feat(event): Use the cancelable context in the event processor. (#73)

Author: Michael Ng

optimizely/client/factory.go

@@ -118,8 +118,7 @@ func (f OptimizelyFactory) ClientWithOptions(clientOptions Options) (*Optimizely
 	}
 
 	// @TODO: allow event processor to be passed in
-	// @TODO: pass the context object to the event processor
-	client.eventProcessor = event.NewEventProcessor(defaultEventQueueSize, defaultEventFlushInterval)
+	client.eventProcessor = event.NewEventProcessor(ctx, defaultEventQueueSize, defaultEventFlushInterval)
 	client.isValid = true
 	return client, nil
 }

optimizely/event/factory_test.go

@@ -1,6 +1,7 @@
 package event
 
 import (
+	"context"
 	"math/rand"
 	"testing"
 	"time"
@@ -88,10 +89,11 @@ func BuildTestConversionEvent() UserEvent {
 }
 
 func TestCreateAndSendImpressionEvent(t *testing.T) {
+	ctx := context.Background()
 
 	impressionUserEvent := BuildTestImpressionEvent()
 
-	processor := NewEventProcessor(100, 100)
+	processor := NewEventProcessor(ctx, 100, 100)
 
 	processor.ProcessEvent(impressionUserEvent)
 
@@ -103,10 +105,11 @@ func TestCreateAndSendImpressionEvent(t *testing.T) {
 }
 
 func TestCreateAndSendConversionEvent(t *testing.T) {
+	ctx := context.Background()
 
 	conversionUserEvent := BuildTestConversionEvent()
 
-	processor := NewEventProcessor(100, 100)
+	processor := NewEventProcessor(ctx, 100, 100)
 
 	processor.ProcessEvent(conversionUserEvent)
 

optimizely/event/processor.go

@@ -1,6 +1,7 @@
 package event
 
 import (
+	"context"
 	"errors"
 	"sync"
 	"time"
@@ -27,10 +28,15 @@ type QueueingEventProcessor struct {
 var pLogger = logging.GetLogger("EventProcessor")
 
 // NewEventProcessor returns a new instance of QueueingEventProcessor with queueSize and flushInterval
-func NewEventProcessor(queueSize int, flushInterval time.Duration) *QueueingEventProcessor {
-	p := &QueueingEventProcessor{MaxQueueSize: queueSize, FlushInterval: flushInterval, Q: NewInMemoryQueue(queueSize), EventDispatcher: &HTTPEventDispatcher{}}
+func NewEventProcessor(ctx context.Context, queueSize int, flushInterval time.Duration) *QueueingEventProcessor {
+	p := &QueueingEventProcessor{
+		MaxQueueSize:    queueSize,
+		FlushInterval:   flushInterval,
+		Q:               NewInMemoryQueue(queueSize),
+		EventDispatcher: &HTTPEventDispatcher{},
+	}
 	p.BatchSize = 10
-	p.StartTicker()
+	p.StartTicker(ctx)
 	return p
 }
 
@@ -61,14 +67,21 @@ func (p *QueueingEventProcessor) Remove(count int) []interface{} {
 }
 
 // StartTicker starts new ticker for flushing events
-func (p *QueueingEventProcessor) StartTicker() {
+func (p *QueueingEventProcessor) StartTicker(ctx context.Context) {
 	if p.Ticker != nil {
 		return
 	}
 	p.Ticker = time.NewTicker(p.FlushInterval * time.Millisecond)
 	go func() {
-		for range p.Ticker.C {
-			p.FlushEvents()
+		for {
+			select {
+			case <-p.Ticker.C:
+				p.FlushEvents()
+			case <-ctx.Done():
+				pLogger.Debug("Event processor stopped, flushing events.")
+				p.FlushEvents()
+				return
+			}
 		}
 	}()
 }

optimizely/event/processor_test.go

@@ -1,14 +1,17 @@
 package event
 
 import (
+	"context"
 	"testing"
 	"time"
 
 	"github.com/stretchr/testify/assert"
 )
 
 func TestDefaultEventProcessor_ProcessImpression(t *testing.T) {
-	processor := NewEventProcessor(100, 100)
+	ctx := context.Background()
+
+	processor := NewEventProcessor(ctx, 100, 100)
 
 	impression := BuildTestImpressionEvent()
 
@@ -21,7 +24,6 @@ func TestDefaultEventProcessor_ProcessImpression(t *testing.T) {
 	assert.NotNil(t, processor.Ticker)
 
 	assert.Equal(t, 0, processor.EventsCount())
-
 }
 
 type MockDispatcher struct {
@@ -34,9 +36,14 @@ func (f *MockDispatcher) DispatchEvent(event LogEvent, callback func(success boo
 }
 
 func TestDefaultEventProcessor_ProcessBatch(t *testing.T) {
-	processor := &QueueingEventProcessor{MaxQueueSize: 100, FlushInterval: 100, Q: NewInMemoryQueue(100), EventDispatcher: &MockDispatcher{}}
+	processor := &QueueingEventProcessor{
+		MaxQueueSize:    100,
+		FlushInterval:   100,
+		Q:               NewInMemoryQueue(100),
+		EventDispatcher: &MockDispatcher{},
+	}
 	processor.BatchSize = 10
-	processor.StartTicker()
+	processor.StartTicker(context.TODO())
 
 	impression := BuildTestImpressionEvent()
 	conversion := BuildTestConversionEvent()
@@ -63,10 +70,46 @@ func TestDefaultEventProcessor_ProcessBatch(t *testing.T) {
 	}
 }
 
+func TestBatchEventProcessor_FlushesOnClose(t *testing.T) {
+	ctx, cancelFn := context.WithCancel(context.Background())
+	processor := &QueueingEventProcessor{
+		MaxQueueSize:    100,
+		FlushInterval:   30 * time.Second,
+		Q:               NewInMemoryQueue(100),
+		EventDispatcher: &MockDispatcher{},
+	}
+	processor.BatchSize = 10
+	processor.StartTicker(ctx)
+
+	impression := BuildTestImpressionEvent()
+	conversion := BuildTestConversionEvent()
+
+	processor.ProcessEvent(impression)
+	processor.ProcessEvent(impression)
+	processor.ProcessEvent(conversion)
+	processor.ProcessEvent(conversion)
+
+	assert.Equal(t, 4, processor.EventsCount())
+
+	time.Sleep(500 * time.Millisecond)
+
+	// Triggers the flush in the processor
+	cancelFn()
+
+	time.Sleep(500 * time.Millisecond)
+
+	assert.Equal(t, 0, processor.EventsCount())
+}
+
 func TestDefaultEventProcessor_ProcessBatchRevisionMismatch(t *testing.T) {
-	processor := &QueueingEventProcessor{MaxQueueSize: 100, FlushInterval: 100, Q: NewInMemoryQueue(100), EventDispatcher: &MockDispatcher{}}
+	processor := &QueueingEventProcessor{
+		MaxQueueSize:    100,
+		FlushInterval:   100,
+		Q:               NewInMemoryQueue(100),
+		EventDispatcher: &MockDispatcher{},
+	}
 	processor.BatchSize = 10
-	processor.StartTicker()
+	processor.StartTicker(context.TODO())
 
 	impression := BuildTestImpressionEvent()
 	conversion := BuildTestConversionEvent()
@@ -95,9 +138,14 @@ func TestDefaultEventProcessor_ProcessBatchRevisionMismatch(t *testing.T) {
 }
 
 func TestDefaultEventProcessor_ProcessBatchProjectMismatch(t *testing.T) {
-	processor := &QueueingEventProcessor{MaxQueueSize: 100, FlushInterval: 100, Q: NewInMemoryQueue(100), EventDispatcher: &MockDispatcher{}}
+	processor := &QueueingEventProcessor{
+		MaxQueueSize:    100,
+		FlushInterval:   100,
+		Q:               NewInMemoryQueue(100),
+		EventDispatcher: &MockDispatcher{},
+	}
 	processor.BatchSize = 10
-	processor.StartTicker()
+	processor.StartTicker(context.TODO())
 
 	impression := BuildTestImpressionEvent()
 	conversion := BuildTestConversionEvent()