(feat): first pass at using sdk key with processor for notification events. … (#140)

* first pass at using sdk key with processor for notification events.  this should be refactored to use the sdk key in the project config.

* add setting sdk key to client factory

* fix lint errors

* lint

* remove wrong package that was auto added

* cleanup code a little

* added some more testing for dispatch

* refactor notification to hide log event until registered callback

* rename callback parameter

Author: thomaszurkan-optimizely

pkg/client/factory.go

@@ -46,7 +46,8 @@ func (f OptimizelyFactory) Client(clientOptions ...OptionFunc) (*OptimizelyClien
 		executionCtx:    executionCtx,
 		DecisionService: decision.NewCompositeService(f.SDKKey),
 		EventProcessor: event.NewEventProcessor(event.BatchSize(event.DefaultBatchSize),
-			event.QueueSize(event.DefaultEventQueueSize), event.FlushInterval(event.DefaultEventFlushInterval)),
+			event.QueueSize(event.DefaultEventQueueSize), event.FlushInterval(event.DefaultEventFlushInterval),
+			event.SDKKey(f.SDKKey)),
 	}
 
 	for _, opt := range clientOptions {

pkg/event/dispatcher.go

@@ -128,6 +128,11 @@ func (ed *QueueEventDispatcher) flushEvents() {
 			}
 		} else {
 			dispatcherLogger.Error("Error dispatching ", err)
+			// we failed.  Sleep some seconds and try again.
+			time.Sleep(sleepTime)
+			// increase retryCount.  We exit if we have retried x times.
+			// we will retry again next event that is added.
+			retryCount++
 		}
 	}
 }

pkg/event/dispatcher_test.go

@@ -58,3 +58,59 @@ func TestQueueEventDispatcher_DispatchEvent(t *testing.T) {
 	assert.Equal(t, 0, qd.eventQueue.Size())
 
 }
+
+func TestQueueEventDispatcher_InvalidEvent(t *testing.T) {
+	ctx := context.TODO()
+	q := NewQueueEventDispatcher(ctx)
+
+
+	config := TestConfig{}
+
+	if qed, ok := q.(*QueueEventDispatcher); ok {
+		qed.Dispatcher = &MockDispatcher{Events: NewInMemoryQueue(100)}
+		qed.eventQueue.Add(config)
+	}
+
+	qd, _ := q.(*QueueEventDispatcher)
+
+	assert.Equal(t, 1, qd.eventQueue.Size())
+
+	// give the queue a chance to run
+	qd.flushEvents()
+
+	// check the queue. bad event type should be removed.  but, not sent.
+	assert.Equal(t, 0, qd.eventQueue.Size())
+
+}
+
+func TestQueueEventDispatcher_FailDispath(t *testing.T) {
+	ctx := context.TODO()
+	q := NewQueueEventDispatcher(ctx)
+
+	if qed, ok := q.(*QueueEventDispatcher); ok {
+		qed.Dispatcher = &MockDispatcher{ShouldFail: true, Events: NewInMemoryQueue(100)}
+	}
+
+	eventTags := map[string]interface{}{"revenue": 55.0, "value": 25.1}
+	config := TestConfig{}
+
+	conversionUserEvent := CreateConversionUserEvent(config, entities.Event{ExperimentIds: []string{"15402980349"}, ID: "15368860886", Key: "sample_conversion"}, userContext, eventTags)
+
+	batch := createBatchEvent(conversionUserEvent, createVisitorFromUserEvent(conversionUserEvent))
+
+	logEvent := createLogEvent(batch)
+
+	q.DispatchEvent(logEvent)
+
+	qd, _ := q.(*QueueEventDispatcher)
+
+	assert.Equal(t, 1, qd.eventQueue.Size())
+
+	// give the queue a chance to run
+	qd.flushEvents()
+
+	// check the queue. bad event type should be removed.  but, not sent.
+	assert.Equal(t, 1, qd.eventQueue.Size())
+
+}
+

pkg/event/processor.go

@@ -19,12 +19,13 @@ package event
 
 import (
 	"errors"
+	"fmt"
+	"github.com/optimizely/go-sdk/pkg/logging"
+	"github.com/optimizely/go-sdk/pkg/notification"
+	"github.com/optimizely/go-sdk/pkg/registry"
+	"github.com/optimizely/go-sdk/pkg/utils"
 	"sync"
 	"time"
-
-	"github.com/optimizely/go-sdk/pkg/utils"
-
-	"github.com/optimizely/go-sdk/pkg/logging"
 )
 
 // Processor processes events
@@ -34,6 +35,7 @@ type Processor interface {
 
 // QueueingEventProcessor is used out of the box by the SDK
 type QueueingEventProcessor struct {
+	sdkKey 			string
 	MaxQueueSize    int           // max size of the queue before flush
 	FlushInterval   time.Duration // in milliseconds
 	BatchSize       int
@@ -92,6 +94,14 @@ func PDispatcher(d Dispatcher) QPConfigOption {
 	}
 }
 
+// SDKKey sets the SDKKey used to register for notifications.  This should be removed when the project
+// config supports sdk key.
+func SDKKey(sdkKey string) QPConfigOption {
+	return func(qp *QueueingEventProcessor) {
+		qp.sdkKey = sdkKey
+	}
+}
+
 // NewEventProcessor returns a new instance of QueueingEventProcessor with queueSize and flushInterval
 func NewEventProcessor(options ...QPConfigOption) *QueueingEventProcessor {
 	p := &QueueingEventProcessor{}
@@ -242,7 +252,15 @@ func (p *QueueingEventProcessor) FlushEvents() {
 		}
 		if batchEventCount > 0 {
 			// TODO: figure out what to do with the error
-			if success, _ := p.EventDispatcher.DispatchEvent(createLogEvent(batchEvent)); success {
+			logEvent := createLogEvent(batchEvent)
+			notificationCenter := registry.GetNotificationCenter(p.sdkKey)
+
+			err := notificationCenter.Send(notification.LogEvent, logEvent)
+
+			if err != nil {
+				pLogger.Error("Send Log Event notification failed.", err)
+			}
+			if success, _ := p.EventDispatcher.DispatchEvent(logEvent); success {
 				pLogger.Debug("Dispatched event successfully")
 				p.Remove(batchEventCount)
 				batchEventCount = 0
@@ -255,3 +273,33 @@ func (p *QueueingEventProcessor) FlushEvents() {
 	}
 	p.Mux.Unlock()
 }
+
+// OnEventDispatch registers a handler for LogEvent notifications
+func (p *QueueingEventProcessor) OnEventDispatch(callback func(logEvent LogEvent)) (int, error) {
+	notificationCenter := registry.GetNotificationCenter(p.sdkKey)
+
+	handler := func(payload interface{}) {
+		if ev, ok := payload.(LogEvent); ok {
+			callback(ev)
+		} else {
+			pLogger.Warning(fmt.Sprintf("Unable to convert notification payload %v into LogEventNotification", payload))
+		}
+	}
+	id, err := notificationCenter.AddHandler(notification.LogEvent, handler)
+	if err != nil {
+		pLogger.Error("Problem with adding notification handler.", err)
+		return 0, err
+	}
+	return id, nil
+}
+
+// RemoveOnEventDispatch removes handler for LogEvent notification with given id
+func (p *QueueingEventProcessor) RemoveOnEventDispatch(id int) error {
+	notificationCenter := registry.GetNotificationCenter(p.sdkKey)
+
+	if err := notificationCenter.RemoveHandler(id, notification.LogEvent); err != nil {
+		pLogger.Warning("Problem with removing notification handler.")
+		return err
+	}
+	return nil
+}

pkg/event/processor_test.go

@@ -77,6 +77,40 @@ func (f *MockDispatcher) DispatchEvent(event LogEvent) (bool, error) {
 	return true, nil
 }
 
+func TestDefaultEventProcessor_LogEventNotification(t *testing.T) {
+	exeCtx := utils.NewCancelableExecutionCtx()
+	processor := NewEventProcessor(FlushInterval(100), QueueSize(100),
+		PQ(NewInMemoryQueue(100)), PDispatcher(&MockDispatcher{Events: NewInMemoryQueue(100)}),
+		SDKKey("fakeSDKKey"))
+
+	var logEvent LogEvent
+
+	id, _ := processor.OnEventDispatch(func(eventNotification LogEvent) {
+		logEvent = eventNotification
+	})
+	processor.Start(exeCtx)
+
+	impression := BuildTestImpressionEvent()
+	conversion := BuildTestConversionEvent()
+
+	processor.ProcessEvent(impression)
+	processor.ProcessEvent(impression)
+	processor.ProcessEvent(conversion)
+	processor.ProcessEvent(conversion)
+
+	assert.Equal(t, 4, processor.EventsCount())
+
+	exeCtx.TerminateAndWait()
+
+	assert.NotNil(t, logEvent)
+	assert.Equal(t, 4, len(logEvent.Event.Visitors))
+
+	err := processor.RemoveOnEventDispatch(id)
+
+	assert.Nil(t, err)
+
+}
+
 func TestDefaultEventProcessor_ProcessBatch(t *testing.T) {
 	exeCtx := utils.NewCancelableExecutionCtx()
 	processor := NewEventProcessor(FlushInterval(100), QueueSize(100),

pkg/notification/center.go

@@ -35,9 +35,11 @@ type DefaultCenter struct {
 func NewNotificationCenter() *DefaultCenter {
 	decisionNotificationManager := NewAtomicManager()
 	projectConfigUpdateNotificationManager := NewAtomicManager()
+	processLogEventNotificationManager := NewAtomicManager()
 	managerMap := make(map[Type]Manager)
 	managerMap[Decision] = decisionNotificationManager
 	managerMap[ProjectConfigUpdate] = projectConfigUpdateNotificationManager
+	managerMap[LogEvent] = processLogEventNotificationManager
 	return &DefaultCenter{
 		managerMap: managerMap,
 	}

pkg/notification/entities.go

@@ -17,7 +17,9 @@
 // Package notification //
 package notification
 
-import "github.com/optimizely/go-sdk/pkg/entities"
+import (
+	"github.com/optimizely/go-sdk/pkg/entities"
+)
 
 // Type is the type of notification
 type Type string
@@ -35,6 +37,8 @@ const (
 	ABTest DecisionNotificationType = "ab-test"
 	// Feature is used when the decision is returned as part of evaluating a feature
 	Feature DecisionNotificationType = "feature"
+	// LogEvent notification type
+	LogEvent Type = "log_event_notification"
 )
 
 // DecisionNotification is a notification triggered when a decision is made for either a feature or an experiment
@@ -49,3 +53,9 @@ type ProjectConfigUpdateNotification struct {
 	Type     Type
 	Revision string
 }
+
+// LogEventNotification is the notification triggered before log event is dispatched.
+type LogEventNotification struct {
+	Type         Type
+	LogEvent  	 interface{}
+}