Refactor execution context. (#212)

* Allow OptimizelyFactory to accept a plain old context object (POCO)
* Rework ExecutionCtx into ExecGroup (ala errgroup)
* Refactor Start methods to NOT manage their own goroutines, but assume to be executed within one (this is similar to http.ListenAndServe)

Author: Mike Davis

pkg/client/client.go

@@ -40,7 +40,7 @@ type OptimizelyClient struct {
 	DecisionService    decision.Service
 	EventProcessor     event.Processor
 	notificationCenter notification.Center
-	executionCtx       utils.ExecutionCtx
+	execGroup          *utils.ExecGroup
 }
 
 // Activate returns the key of the variation the user is bucketed into and queues up an impression event to be sent to
@@ -488,5 +488,5 @@ func (o *OptimizelyClient) GetProjectConfig() (projectConfig config.ProjectConfi
 
 // Close closes the Optimizely instance and stops any ongoing tasks from its children components.
 func (o *OptimizelyClient) Close() {
-	o.executionCtx.TerminateAndWait()
+	o.execGroup.TerminateAndWait()
 }

pkg/client/client_test.go

@@ -28,31 +28,13 @@ import (
 	"github.com/optimizely/go-sdk/pkg/entities"
 	"github.com/optimizely/go-sdk/pkg/event"
 	"github.com/optimizely/go-sdk/pkg/notification"
+	"github.com/optimizely/go-sdk/pkg/utils"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/suite"
 )
 
-var exeCtxSignalFlag bool
-
-type ExecutionCtx struct {
-	Wg  *sync.WaitGroup
-	Ctx context.Context
-}
-
-func (ctx ExecutionCtx) TerminateAndWait() {
-	exeCtxSignalFlag = true
-}
-
-func (ctx ExecutionCtx) GetContext() context.Context {
-	return ctx.Ctx
-}
-
-func (ctx ExecutionCtx) GetWaitSync() *sync.WaitGroup {
-	return ctx.Wg
-}
-
 func ValidProjectConfigManager() *MockProjectConfigManager {
 	p := new(MockProjectConfigManager)
 	p.projectConfig = new(TestConfig)
@@ -1934,17 +1916,24 @@ func TestClose(t *testing.T) {
 	mockProcessor := &MockProcessor{}
 	mockDecisionService := new(MockDecisionService)
 
+	eg := utils.NewExecGroup(context.Background())
+
+	wg := &sync.WaitGroup{}
+	wg.Add(1)
+	eg.Go(func(ctx context.Context) {
+		<-ctx.Done()
+		wg.Done()
+	})
+
 	client := OptimizelyClient{
 		ConfigManager:   ValidProjectConfigManager(),
 		DecisionService: mockDecisionService,
 		EventProcessor:  mockProcessor,
-		executionCtx:    new(ExecutionCtx),
+		execGroup:       eg,
 	}
 
-	assert.False(t, exeCtxSignalFlag)
 	client.Close()
-	assert.True(t, exeCtxSignalFlag)
-
+	wg.Wait()
 }
 
 type ClientTestSuiteTrackEvent struct {

pkg/client/factory.go

@@ -18,6 +18,7 @@
 package client
 
 import (
+	"context"
 	"errors"
 	"time"
 
@@ -34,10 +35,10 @@ type OptimizelyFactory struct {
 	Datafile []byte
 
 	configManager      config.ProjectConfigManager
+	ctx                context.Context
 	decisionService    decision.Service
 	eventDispatcher    event.Dispatcher
 	eventProcessor     event.Processor
-	executionCtx       utils.ExecutionCtx
 	userProfileService decision.UserProfileService
 	overrideStore      decision.ExperimentOverrideStore
 }
@@ -56,14 +57,15 @@ func (f OptimizelyFactory) Client(clientOptions ...OptionFunc) (*OptimizelyClien
 		return nil, errors.New("unable to instantiate client: no project config manager, SDK key, or a Datafile provided")
 	}
 
-	var executionCtx utils.ExecutionCtx
-	if f.executionCtx != nil {
-		executionCtx = f.executionCtx
+	var ctx context.Context
+	if f.ctx != nil {
+		ctx = f.ctx
 	} else {
-		executionCtx = utils.NewCancelableExecutionCtx()
+		ctx = context.Background()
 	}
 
-	appClient := &OptimizelyClient{executionCtx: executionCtx, notificationCenter: registry.GetNotificationCenter(f.SDKKey)}
+	eg := utils.NewExecGroup(ctx)
+	appClient := &OptimizelyClient{execGroup: eg, notificationCenter: registry.GetNotificationCenter(f.SDKKey)}
 
 	if f.configManager != nil {
 		appClient.ConfigManager = f.configManager
@@ -103,11 +105,11 @@ func (f OptimizelyFactory) Client(clientOptions ...OptionFunc) (*OptimizelyClien
 
 	// Initialize the default services with the execution context
 	if pollingConfigManager, ok := appClient.ConfigManager.(*config.PollingProjectConfigManager); ok {
-		pollingConfigManager.Start(appClient.executionCtx)
+		eg.Go(pollingConfigManager.Start)
 	}
 
 	if batchProcessor, ok := appClient.EventProcessor.(*event.BatchEventProcessor); ok {
-		batchProcessor.Start(appClient.executionCtx)
+		eg.Go(batchProcessor.Start)
 	}
 
 	return appClient, nil
@@ -171,10 +173,10 @@ func WithEventDispatcher(eventDispatcher event.Dispatcher) OptionFunc {
 	}
 }
 
-// WithExecutionContext allows user to pass in their own execution context to override the default one in the client.
-func WithExecutionContext(executionContext utils.ExecutionCtx) OptionFunc {
+// WithContext allows user to pass in their own context to override the default one in the client.
+func WithContext(ctx context.Context) OptionFunc {
 	return func(f *OptimizelyFactory) {
-		f.executionCtx = executionContext
+		f.ctx = ctx
 	}
 }
 

pkg/client/factory_test.go

@@ -17,8 +17,10 @@
 package client
 
 import (
+	"context"
 	"errors"
 	"net/http"
+	"sync"
 	"testing"
 	"time"
 
@@ -117,14 +119,23 @@ func TestClientWithDecisionServiceAndEventProcessorInOptions(t *testing.T) {
 
 func TestClientWithCustomCtx(t *testing.T) {
 	factory := OptimizelyFactory{}
-	testExecutionCtx := utils.NewCancelableExecutionCtx()
+	ctx, cancel := context.WithCancel(context.Background())
 	mockConfigManager := new(MockProjectConfigManager)
 	client, err := factory.Client(
 		WithConfigManager(mockConfigManager),
-		WithExecutionContext(testExecutionCtx),
+		WithContext(ctx),
 	)
 	assert.NoError(t, err)
-	assert.Equal(t, client.executionCtx, testExecutionCtx)
+
+	wg := &sync.WaitGroup{}
+	wg.Add(1)
+	client.execGroup.Go(func(ctx context.Context) {
+		<-ctx.Done()
+		wg.Done()
+	})
+
+	cancel()
+	wg.Wait()
 }
 
 func TestStaticClient(t *testing.T) {

pkg/config/polling_manager.go

@@ -18,6 +18,7 @@
 package config
 
 import (
+	"context"
 	"fmt"
 	"net/http"
 	"sync"
@@ -170,20 +171,18 @@ func (cm *PollingProjectConfigManager) SyncConfig(datafile []byte) {
 }
 
 // Start starts the polling
-func (cm *PollingProjectConfigManager) Start(exeCtx utils.ExecutionCtx) {
-	go func() {
-		cmLogger.Debug("Polling Config Manager Initiated")
-		t := time.NewTicker(cm.pollingInterval)
-		for {
-			select {
-			case <-t.C:
-				cm.SyncConfig([]byte{})
-			case <-exeCtx.GetContext().Done():
-				cmLogger.Debug("Polling Config Manager Stopped")
-				return
-			}
+func (cm *PollingProjectConfigManager) Start(ctx context.Context) {
+	cmLogger.Debug("Polling Config Manager Initiated")
+	t := time.NewTicker(cm.pollingInterval)
+	for {
+		select {
+		case <-t.C:
+			cm.SyncConfig([]byte{})
+		case <-ctx.Done():
+			cmLogger.Debug("Polling Config Manager Stopped")
+			return
 		}
-	}()
+	}
 }
 
 // NewPollingProjectConfigManager returns an instance of the polling config manager with the customized configuration

pkg/config/polling_manager_test.go

@@ -17,6 +17,7 @@
 package config
 
 import (
+	"context"
 	"net/http"
 	"testing"
 	"time"
@@ -39,6 +40,10 @@ func (m *MockRequester) Get(uri string, headers ...utils.Header) (response []byt
 	return args.Get(0).([]byte), args.Get(1).(http.Header), args.Int(2), args.Error(3)
 }
 
+func newExecGroup() *utils.ExecGroup {
+	return utils.NewExecGroup(context.Background())
+}
+
 func TestNewPollingProjectConfigManagerWithOptions(t *testing.T) {
 
 	mockDatafile := []byte(`{"revision":"42"}`)
@@ -49,17 +54,17 @@ func TestNewPollingProjectConfigManagerWithOptions(t *testing.T) {
 	// Test we fetch using requester
 	sdkKey := "test_sdk_key"
 
-	exeCtx := utils.NewCancelableExecutionCtx()
+	eg := newExecGroup()
 	configManager := NewPollingProjectConfigManager(sdkKey, WithRequester(mockRequester))
-	configManager.Start(exeCtx)
+	eg.Go(configManager.Start)
 	mockRequester.AssertExpectations(t)
 
 	actual, err := configManager.GetConfig()
 	assert.Nil(t, err)
 	assert.NotNil(t, actual)
 	assert.Equal(t, projectConfig, actual)
 
-	exeCtx.TerminateAndWait() // just sending signal and improving coverage
+	eg.TerminateAndWait() // just sending signal and improving coverage
 }
 
 func TestNewPollingProjectConfigManagerWithNull(t *testing.T) {
@@ -70,9 +75,9 @@ func TestNewPollingProjectConfigManagerWithNull(t *testing.T) {
 	// Test we fetch using requester
 	sdkKey := "test_sdk_key"
 
-	exeCtx := utils.NewCancelableExecutionCtx()
+	eg := newExecGroup()
 	configManager := NewPollingProjectConfigManager(sdkKey, WithRequester(mockRequester))
-	configManager.Start(exeCtx)
+	eg.Go(configManager.Start)
 	mockRequester.AssertExpectations(t)
 
 	_, err := configManager.GetConfig()
@@ -88,9 +93,9 @@ func TestNewPollingProjectConfigManagerWithSimilarDatafileRevisions(t *testing.T
 
 	sdkKey := "test_sdk_key"
 
-	exeCtx := utils.NewCancelableExecutionCtx()
+	eg := newExecGroup()
 	configManager := NewPollingProjectConfigManager(sdkKey, WithRequester(mockRequester))
-	configManager.Start(exeCtx)
+	eg.Go(configManager.Start)
 	mockRequester.AssertExpectations(t)
 
 	actual, err := configManager.GetConfig()
@@ -116,9 +121,9 @@ func TestNewPollingProjectConfigManagerWithLastModifiedDates(t *testing.T) {
 
 	sdkKey := "test_sdk_key"
 
-	exeCtx := utils.NewCancelableExecutionCtx()
+	eg := newExecGroup()
 	configManager := NewPollingProjectConfigManager(sdkKey, WithRequester(mockRequester))
-	configManager.Start(exeCtx)
+	eg.Go(configManager.Start)
 
 	// Fetch valid config
 	actual, err := configManager.GetConfig()
@@ -146,9 +151,9 @@ func TestNewPollingProjectConfigManagerWithDifferentDatafileRevisions(t *testing
 	// Test we fetch using requester
 	sdkKey := "test_sdk_key"
 
-	exeCtx := utils.NewCancelableExecutionCtx()
+	eg := newExecGroup()
 	configManager := NewPollingProjectConfigManager(sdkKey, WithRequester(mockRequester))
-	configManager.Start(exeCtx)
+	eg.Go(configManager.Start)
 	mockRequester.AssertExpectations(t)
 
 	actual, err := configManager.GetConfig()
@@ -173,9 +178,9 @@ func TestNewPollingProjectConfigManagerWithErrorHandling(t *testing.T) {
 	// Test we fetch using requester
 	sdkKey := "test_sdk_key"
 
-	exeCtx := utils.NewCancelableExecutionCtx()
+	eg := newExecGroup()
 	configManager := NewPollingProjectConfigManager(sdkKey, WithRequester(mockRequester))
-	configManager.Start(exeCtx)
+	eg.Go(configManager.Start)
 	mockRequester.AssertExpectations(t)
 
 	actual, err := configManager.GetConfig() // polling for bad file
@@ -204,9 +209,9 @@ func TestNewPollingProjectConfigManagerOnDecision(t *testing.T) {
 	// Test we fetch using requester
 	sdkKey := "test_sdk_key"
 
-	exeCtx := utils.NewCancelableExecutionCtx()
+	eg := newExecGroup()
 	configManager := NewPollingProjectConfigManager(sdkKey, WithRequester(mockRequester))
-	configManager.Start(exeCtx)
+	eg.Go(configManager.Start)
 
 	var numberOfCalls = 0
 	callback := func(notification notification.ProjectConfigUpdateNotification) {
@@ -238,19 +243,19 @@ func TestPollingInterval(t *testing.T) {
 
 	sdkKey := "test_sdk_key"
 
-	exeCtx := utils.NewCancelableExecutionCtx()
+	eg := newExecGroup()
 	configManager := NewPollingProjectConfigManager(sdkKey, WithPollingInterval(5*time.Second))
-	configManager.Start(exeCtx)
+	eg.Go(configManager.Start)
 
 	assert.Equal(t, configManager.pollingInterval, 5*time.Second)
 }
 
 func TestInitialDatafile(t *testing.T) {
 
 	sdkKey := "test_sdk_key"
-	exeCtx := utils.NewCancelableExecutionCtx()
+	eg := newExecGroup()
 	configManager := NewPollingProjectConfigManager(sdkKey, WithInitialDatafile([]byte("test")))
-	configManager.Start(exeCtx)
+	eg.Go(configManager.Start)
 
 	assert.Equal(t, configManager.initDatafile, []byte("test"))
 }

pkg/event/dispatcher.go

@@ -18,7 +18,6 @@
 package event
 
 import (
-	"context"
 	"fmt"
 	"net/http"
 	"sync"
@@ -166,13 +165,10 @@ func (ed *QueueEventDispatcher) flushEvents() {
 }
 
 // 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{requester: utils.NewHTTPRequester()}, metrics: &DefaultMetrics{}}
-
-	go func() {
-		<-ctx.Done()
-		dispatcher.flushEvents()
-	}()
-
-	return dispatcher
+func NewQueueEventDispatcher() *QueueEventDispatcher {
+	return &QueueEventDispatcher{
+		eventQueue: NewInMemoryQueue(defaultQueueSize),
+		Dispatcher: &HTTPEventDispatcher{requester: utils.NewHTTPRequester()},
+		metrics:    &DefaultMetrics{},
+	}
 }