(feat): Queue event dispatcher (#88)

thomaszurkan-optimizely · web-flow · commit ae6d34e197f1 · 2019-09-11T10:28:53.000-07:00
* first pass at event queued event dispatcher

* fix typo

* add unit test and fix dispatch

* fix removing bad event

* cleanup from Mike Ng's comments

* add context

* update dispatcher to use context correctly

* pass in context

* update lint fixes

* fix lint issues

* fix lint issues.

* remove sleep

* add some logging

* cleanup and add header

* merge in fm-beta.  remove dispatch callback

* update license

* add headers after merge
diff --git a/optimizely/event/dispatcher.go b/optimizely/event/dispatcher.go
@@ -19,14 +19,22 @@ package event
 
 import (
 	"bytes"
+	"context"
 	"encoding/json"
 	"fmt"
 	"net/http"
+	"sync"
+	"time"
 
 	"github.com/optimizely/go-sdk/optimizely/logging"
 )
 
 const jsonContentType = "application/json"
+const maxRetries = 3
+const defaultQueueSize = 1000
+const sleepTime = 5 * time.Second
+
+var dispatcherLogger = logging.GetLogger("EventDispatcher")
 
 // Dispatcher dispatches events
 type Dispatcher interface {
@@ -37,8 +45,6 @@ type Dispatcher interface {
 type HTTPEventDispatcher struct {
 }
 
-var dispatcherLogger = logging.GetLogger("EventDispatcher")
-
 // DispatchEvent dispatches event with callback
 func (*HTTPEventDispatcher) DispatchEvent(event LogEvent) (bool, error) {
 
@@ -54,9 +60,88 @@ func (*HTTPEventDispatcher) DispatchEvent(event LogEvent) (bool, error) {
 		if resp.StatusCode == 204 {
 			success = true
 		} else {
-			dispatcherLogger.Error(fmt.Sprintf("http.Post invalid response %d", resp.StatusCode), err)
+			dispatcherLogger.Error(fmt.Sprintf("http.Post invalid response %d", resp.StatusCode), nil)
 			success = false
 		}
 	}
 	return success, err
 }
+
+// QueueEventDispatcher is a queued version of the event dispatcher that queues, returns success, and dispatches events in the background
+type QueueEventDispatcher struct {
+	eventQueue     Queue
+	eventFlushLock sync.Mutex
+	dispatcher     *HTTPEventDispatcher
+}
+
+// DispatchEvent queues event with callback and calls flush in a go routine.
+func (ed *QueueEventDispatcher) DispatchEvent(event LogEvent) (bool, error) {
+	ed.eventQueue.Add(event)
+	go func() {
+		ed.flushEvents()
+	}()
+	return true,nil
+}
+
+// flush the events
+func (ed *QueueEventDispatcher) flushEvents() {
+
+	ed.eventFlushLock.Lock()
+
+	defer func(){
+		ed.eventFlushLock.Unlock()
+	}()
+
+	retryCount := 0
+
+	for ed.eventQueue.Size() > 0 {
+		if retryCount > maxRetries {
+			dispatcherLogger.Error(fmt.Sprintf("event failed to send %d times. It will retry on next event sent", maxRetries), nil)
+			break
+		}
+
+		items := ed.eventQueue.Get(1)
+		if len(items) == 0 {
+			// something happened.  Just continue and you should expect size to be zero.
+			continue
+		}
+		event, ok := items[0].(LogEvent)
+		if !ok {
+			// remove it
+			dispatcherLogger.Error("invalid type passed to event dispatcher", nil)
+			ed.eventQueue.Remove(1)
+			continue
+		}
+
+		success, err := ed.dispatcher.DispatchEvent(event)
+
+		if err == nil {
+			if success {
+				dispatcherLogger.Debug(fmt.Sprintf("Dispatched log event %+v", event))
+				ed.eventQueue.Remove(1)
+				retryCount = 0
+			} else {
+				dispatcherLogger.Warning("dispatch event failed")
+				// 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++
+			}
+		} else {
+			dispatcherLogger.Error("Error dispatching ", err)
+		}
+	}
+}
+
+// NewQueueEventDispatcher creates a dispatcher that queues in memory and then sends via go routine.
+func NewQueueEventDispatcher(ctx context.Context) Dispatcher {
+	dispatcher := &QueueEventDispatcher{eventQueue: NewInMemoryQueue(defaultQueueSize), dispatcher:&HTTPEventDispatcher{}}
+
+	go func() {
+		<-ctx.Done()
+		dispatcher.flushEvents()
+	}()
+
+	return dispatcher
+}
diff --git a/optimizely/event/dispatcher_test.go b/optimizely/event/dispatcher_test.go
@@ -0,0 +1,56 @@
+/****************************************************************************
+ * Copyright 2019, Optimizely, Inc. and contributors                        *
+ *                                                                          *
+ * Licensed under the Apache License, Version 2.0 (the "License");          *
+ * you may not use this file except in compliance with the License.         *
+ * You may obtain a copy of the License at                                  *
+ *                                                                          *
+ *    http://www.apache.org/licenses/LICENSE-2.0                            *
+ *                                                                          *
+ * Unless required by applicable law or agreed to in writing, software      *
+ * distributed under the License is distributed on an "AS IS" BASIS,        *
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. *
+ * See the License for the specific language governing permissions and      *
+ * limitations under the License.                                           *
+ ***************************************************************************/
+
+// Package event //
+package event
+
+import (
+	"context"
+	"github.com/optimizely/go-sdk/optimizely/entities"
+	"github.com/stretchr/testify/assert"
+	"testing"
+	"time"
+)
+
+func TestQueueEventDispatcher_DispatchEvent(t *testing.T) {
+	ctx := context.TODO()
+	q := NewQueueEventDispatcher(ctx)
+
+	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)
+
+	qd, _ := q.(*QueueEventDispatcher)
+
+	success, _ := qd.DispatchEvent(logEvent)
+
+	assert.True(t, success)
+
+	// its been queued
+	assert.Equal(t,1, qd.eventQueue.Size())
+
+	// give the queue a chance to run
+	time.Sleep(1 * time.Second)
+
+	// check the queue
+	assert.Equal(t,0, qd.eventQueue.Size())
+
+}
diff --git a/optimizely/event/factory_test.go b/optimizely/event/factory_test.go
@@ -1,3 +1,20 @@
+/****************************************************************************
+ * Copyright 2019, Optimizely, Inc. and contributors                        *
+ *                                                                          *
+ * Licensed under the Apache License, Version 2.0 (the "License");          *
+ * you may not use this file except in compliance with the License.         *
+ * You may obtain a copy of the License at                                  *
+ *                                                                          *
+ *    http://www.apache.org/licenses/LICENSE-2.0                            *
+ *                                                                          *
+ * Unless required by applicable law or agreed to in writing, software      *
+ * distributed under the License is distributed on an "AS IS" BASIS,        *
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. *
+ * See the License for the specific language governing permissions and      *
+ * limitations under the License.                                           *
+ ***************************************************************************/
+
+// Package event //
 package event
 
 import (
diff --git a/optimizely/event/processor.go b/optimizely/event/processor.go
@@ -62,8 +62,7 @@ func NewEventProcessor(exeCtx utils.ExecutionCtx, batchSize, queueSize int, flus
 		MaxQueueSize:    queueSize,
 		FlushInterval:   flushInterval,
 		Q:               NewInMemoryQueue(queueSize),
-		EventDispatcher: &HTTPEventDispatcher{},
-
+		EventDispatcher: NewQueueEventDispatcher(exeCtx.GetContext()),
 		wg: exeCtx.GetWaitSync(),
 	}
 	p.BatchSize = DefaultBatchSize
@@ -118,6 +117,10 @@ func (p *QueueingEventProcessor) StartTicker(ctx context.Context) {
 			case <-ctx.Done():
 				pLogger.Debug("Event processor stopped, flushing events.")
 				p.FlushEvents()
+				d, ok := p.EventDispatcher.(*QueueEventDispatcher)
+				if ok {
+					d.flushEvents()
+				}
 				return
 			}
 		}
diff --git a/optimizely/event/processor_test.go b/optimizely/event/processor_test.go
@@ -1,3 +1,20 @@
+/****************************************************************************
+ * Copyright 2019, Optimizely, Inc. and contributors                        *
+ *                                                                          *
+ * Licensed under the Apache License, Version 2.0 (the "License");          *
+ * you may not use this file except in compliance with the License.         *
+ * You may obtain a copy of the License at                                  *
+ *                                                                          *
+ *    http://www.apache.org/licenses/LICENSE-2.0                            *
+ *                                                                          *
+ * Unless required by applicable law or agreed to in writing, software      *
+ * distributed under the License is distributed on an "AS IS" BASIS,        *
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. *
+ * See the License for the specific language governing permissions and      *
+ * limitations under the License.                                           *
+ ***************************************************************************/
+
+// Package event //
 package event
 
 import (
diff --git a/optimizely/event/queue_test.go b/optimizely/event/queue_test.go
@@ -1,3 +1,20 @@
+/****************************************************************************
+ * Copyright 2019, Optimizely, Inc. and contributors                        *
+ *                                                                          *
+ * Licensed under the Apache License, Version 2.0 (the "License");          *
+ * you may not use this file except in compliance with the License.         *
+ * You may obtain a copy of the License at                                  *
+ *                                                                          *
+ *    http://www.apache.org/licenses/LICENSE-2.0                            *
+ *                                                                          *
+ * Unless required by applicable law or agreed to in writing, software      *
+ * distributed under the License is distributed on an "AS IS" BASIS,        *
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. *
+ * See the License for the specific language governing permissions and      *
+ * limitations under the License.                                           *
+ ***************************************************************************/
+
+// Package event //
 package event
 
 import (
