(feat): add a way to create a custom event processor taking advantage of Exec… (#113)

* add a way to create a custom event processor taking advantage of ExecutionContext

* remove redundant ticker start and bath setting

* add full test coverage

* fix race condition

* add option config stuff

* add comment for options function

* up code coverage

* refactor to include all option configs

* fix lint

* rename and drop timers package

Author: thomaszurkan-optimizely

optimizely/client/factory.go

@@ -45,7 +45,8 @@ func (f OptimizelyFactory) Client(clientOptions ...OptionFunc) (*OptimizelyClien
 	appClient := &OptimizelyClient{
 		executionCtx:    executionCtx,
 		decisionService: decision.NewCompositeService(f.SDKKey),
-		eventProcessor:  event.NewEventProcessor(executionCtx, event.DefaultBatchSize, event.DefaultEventQueueSize, event.DefaultEventFlushInterval),
+		eventProcessor:  event.NewEventProcessor(executionCtx, event.BatchSize(event.DefaultBatchSize),
+		event.QueueSize(event.DefaultEventQueueSize), event.FlushInterval(event.DefaultEventFlushInterval)),
 	}
 
 	for _, opt := range clientOptions {
@@ -97,7 +98,8 @@ func DecisionService(decisionService decision.Service) OptionFunc {
 // BatchEventProcessor sets event processor on a client
 func BatchEventProcessor(batchSize, queueSize int, flushInterval time.Duration) OptionFunc {
 	return func(f *OptimizelyClient, executionCtx utils.ExecutionCtx) {
-		f.eventProcessor = event.NewEventProcessor(executionCtx, batchSize, queueSize, flushInterval)
+		f.eventProcessor = event.NewEventProcessor(executionCtx, event.BatchSize(batchSize),
+			event.QueueSize(queueSize), event.FlushInterval(flushInterval))
 	}
 }
 

optimizely/event/factory_test.go

@@ -104,11 +104,24 @@ func BuildTestConversionEvent() UserEvent {
 	return conversionUserEvent
 }
 
+func TestCreateEmptyEvent(t *testing.T) {
+
+	impressionUserEvent := BuildTestImpressionEvent()
+
+	impressionUserEvent.Impression = nil
+	impressionUserEvent.Conversion = nil
+
+	visitor := createVisitorFromUserEvent(impressionUserEvent)
+
+	assert.Nil(t, visitor.Snapshots)
+
+}
+
 func TestCreateAndSendImpressionEvent(t *testing.T) {
 
 	impressionUserEvent := BuildTestImpressionEvent()
 
-	processor := NewEventProcessor(utils.NewCancelableExecutionCtx(), 10, 100, 100)
+	processor := NewEventProcessor(utils.NewCancelableExecutionCtx(), BatchSize(10), QueueSize(100), FlushInterval(100))
 
 	processor.ProcessEvent(impressionUserEvent)
 
@@ -123,7 +136,7 @@ func TestCreateAndSendConversionEvent(t *testing.T) {
 
 	conversionUserEvent := BuildTestConversionEvent()
 
-	processor := NewEventProcessor(utils.NewCancelableExecutionCtx(), 10, 100, 100)
+	processor := NewEventProcessor(utils.NewCancelableExecutionCtx(), FlushInterval(100))
 
 	processor.ProcessEvent(conversionUserEvent)
 

optimizely/event/processor.go

@@ -56,21 +56,75 @@ const DefaultEventFlushInterval = 30 * time.Second
 
 var pLogger = logging.GetLogger("EventProcessor")
 
+// QPConfigOption is the QueuingProcessor options that give you the ability to add one more more options before the processor is initialized.
+type QPConfigOption func(qp *QueueingEventProcessor)
+
+// BatchSize sets the batch size as a config option to be passed into the NewProcessor method
+func BatchSize(bsize int) QPConfigOption {
+	return func(qp *QueueingEventProcessor) {
+		qp.BatchSize = bsize
+	}
+}
+
+// QueueSize sets the queue size as a config option to be passed into the NewProcessor method
+func QueueSize(qsize int) QPConfigOption {
+	return func(qp *QueueingEventProcessor) {
+		qp.MaxQueueSize = qsize
+	}
+}
+
+// FlushInterval sets the flush interval as a config option to be passed into the NewProcessor method
+func FlushInterval(flushInterval time.Duration) QPConfigOption {
+	return func(qp *QueueingEventProcessor) {
+		qp.FlushInterval = flushInterval
+	}
+}
+
+// PQ sets the Processor Queue as a config option to be passed into the NewProcessor method
+func PQ(q Queue) QPConfigOption {
+	return func(qp *QueueingEventProcessor) {
+		qp.Q = q
+	}
+}
+// PDispatcher sets the Processor Dispatcher as a config option to be passed into the NewProcessor method
+func PDispatcher(d Dispatcher) QPConfigOption {
+	return func(qp *QueueingEventProcessor) {
+		qp.EventDispatcher = d
+	}
+}
+
 // NewEventProcessor returns a new instance of QueueingEventProcessor with queueSize and flushInterval
-func NewEventProcessor(exeCtx utils.ExecutionCtx, batchSize, queueSize int, flushInterval time.Duration) *QueueingEventProcessor {
+func NewEventProcessor(exeCtx utils.ExecutionCtx, options ...QPConfigOption) *QueueingEventProcessor {
 	p := &QueueingEventProcessor{
-		MaxQueueSize:    queueSize,
-		FlushInterval:   flushInterval,
-		Q:               NewInMemoryQueue(queueSize),
-		EventDispatcher: NewQueueEventDispatcher(exeCtx.GetContext()),
 		wg: exeCtx.GetWaitSync(),
 	}
-	p.BatchSize = DefaultBatchSize
-	if batchSize > 0 {
-		p.BatchSize = batchSize
+
+	for _, opt := range options {
+		opt(p)
+	}
+
+	if p.MaxQueueSize == 0 {
+		p.MaxQueueSize = defaultQueueSize
+	}
+
+	if p.FlushInterval == 0 {
+		p.FlushInterval = DefaultEventFlushInterval
+	}
+
+	if p.BatchSize == 0 {
+		p.BatchSize = DefaultBatchSize
+	}
+
+	if p.Q == nil {
+		p.Q = NewInMemoryQueue(p.MaxQueueSize)
+	}
+
+	if p.EventDispatcher == nil {
+		p.EventDispatcher = NewQueueEventDispatcher(exeCtx.GetContext())
 	}
 
 	p.StartTicker(exeCtx.GetContext())
+
 	return p
 }