Merge pull request #95 from optimizely/pawel/fix_sync_on_close

Pawel/fix sync on close

Author: pawels-optimizely

examples/main.go

@@ -4,10 +4,7 @@
 package main
 
 import (
-	"context"
 	"fmt"
-	"time"
-
 	"github.com/optimizely/go-sdk/optimizely/client"
 	"github.com/optimizely/go-sdk/optimizely/entities"
 	"github.com/optimizely/go-sdk/optimizely/logging"
@@ -40,40 +37,15 @@ func main() {
 	fmt.Printf("Is feature enabled? %v\n", enabled)
 
 	fmt.Println()
-
-	/************* ClientWithOptions - custom context  ********************/
-
-	optimizelyFactory = &client.OptimizelyFactory{
-		SDKKey: "4SLpaJA1r1pgE6T2CoMs9q",
-	}
-	ctx := context.Background()
-	ctx, cancelManager := context.WithCancel(ctx) // user can set up any context
-	clientOptions := client.Options{
-		Context: ctx,
-	}
-
-	app, err = optimizelyFactory.ClientWithOptions(clientOptions)
-	cancelManager() //  user can cancel anytime
-
-	if err != nil {
-		fmt.Printf("Error instantiating client: %s", err)
-		return
-	}
-
-	enabled, _ = app.IsFeatureEnabled("mutext_feat", user)
-	fmt.Printf("Is feature enabled? %v\n", enabled)
-
-	time.Sleep(1000 * time.Millisecond)
+	app.Close() //  user can close dispatcher
 	fmt.Println()
-
 	/************* Client ********************/
 
 	optimizelyFactory = &client.OptimizelyFactory{
 		SDKKey: "4SLpaJA1r1pgE6T2CoMs9q",
 	}
 
 	app, err = optimizelyFactory.Client()
-	app.Close() //  user can cancel anytime
 
 	if err != nil {
 		fmt.Printf("Error instantiating client: %s", err)
@@ -82,5 +54,5 @@ func main() {
 
 	enabled, _ = app.IsFeatureEnabled("mutext_feat", user)
 	fmt.Printf("Is feature enabled? %v\n", enabled)
-
+	app.Close() //  user can close dispatcher
 }

optimizely/client/client.go

@@ -18,18 +18,17 @@
 package client
 
 import (
-	"context"
 	"errors"
 	"fmt"
 	"runtime/debug"
 	"strconv"
 
-	"github.com/optimizely/go-sdk/optimizely/event"
-
 	"github.com/optimizely/go-sdk/optimizely"
 	"github.com/optimizely/go-sdk/optimizely/decision"
 	"github.com/optimizely/go-sdk/optimizely/entities"
+	"github.com/optimizely/go-sdk/optimizely/event"
 	"github.com/optimizely/go-sdk/optimizely/logging"
+	"github.com/optimizely/go-sdk/optimizely/utils"
 )
 
 var logger = logging.GetLogger("Client")
@@ -41,7 +40,7 @@ type OptimizelyClient struct {
 	eventProcessor  event.Processor
 	isValid         bool
 
-	cancelFunc context.CancelFunc
+	executionCtx utils.ExecutionCtx
 }
 
 // IsFeatureEnabled returns true if the feature is enabled for the given user
@@ -279,5 +278,5 @@ func (o *OptimizelyClient) GetProjectConfig() (projectConfig optimizely.ProjectC
 
 // Close closes the Optimizely instance and stops any ongoing tasks from its children components
 func (o *OptimizelyClient) Close() {
-	o.cancelFunc()
+	o.executionCtx.TerminateAndWait()
 }

optimizely/client/factory.go

@@ -18,8 +18,8 @@
 package client
 
 import (
-	"context"
 	"errors"
+	"github.com/optimizely/go-sdk/optimizely/utils"
 
 	"github.com/optimizely/go-sdk/optimizely/event"
 
@@ -32,7 +32,6 @@ import (
 
 // Options are used to create an instance of the OptimizelyClient with custom configuration
 type Options struct {
-	Context              context.Context
 	ProjectConfigManager optimizely.ProjectConfigManager
 	DecisionService      decision.Service
 	EventProcessor       event.Processor
@@ -76,19 +75,11 @@ func (f OptimizelyFactory) StaticClient() (*OptimizelyClient, error) {
 
 // ClientWithOptions returns a client initialized with the given configuration options
 func (f OptimizelyFactory) ClientWithOptions(clientOptions Options) (*OptimizelyClient, error) {
-	client := &OptimizelyClient{
-		isValid: false,
-	}
 
-	var ctx context.Context
-	if clientOptions.Context != nil {
-		ctx = clientOptions.Context
-	} else {
-		// if no context is provided, we create our own cancellable context and hand it over to the client so the client can shut down its child processes
-		ctx = context.Background()
-		var cancel context.CancelFunc
-		ctx, cancel = context.WithCancel(ctx)
-		client.cancelFunc = cancel
+	executionCtx := utils.NewCancelableExecutionCtx()
+	client := &OptimizelyClient{
+		isValid:      false,
+		executionCtx: executionCtx,
 	}
 
 	notificationCenter := notification.NewNotificationCenter()
@@ -100,7 +91,7 @@ func (f OptimizelyFactory) ClientWithOptions(clientOptions Options) (*Optimizely
 		options := config.PollingProjectConfigManagerOptions{
 			Datafile: f.Datafile,
 		}
-		client.configManager = config.NewPollingProjectConfigManagerWithOptions(ctx, f.SDKKey, options)
+		client.configManager = config.NewPollingProjectConfigManagerWithOptions(executionCtx, f.SDKKey, options)
 	case f.Datafile != nil:
 		staticConfigManager, _ := config.NewStaticProjectConfigManagerFromPayload(f.Datafile)
 		client.configManager = staticConfigManager
@@ -117,7 +108,7 @@ func (f OptimizelyFactory) ClientWithOptions(clientOptions Options) (*Optimizely
 	if clientOptions.EventProcessor != nil {
 		client.eventProcessor = clientOptions.EventProcessor
 	} else {
-		client.eventProcessor = event.NewEventProcessor(ctx, event.DefaultBatchSize, event.DefaultEventQueueSize, event.DefaultEventFlushInterval)
+		client.eventProcessor = event.NewEventProcessor(executionCtx, event.DefaultBatchSize, event.DefaultEventQueueSize, event.DefaultEventFlushInterval)
 	}
 
 	client.isValid = true

optimizely/config/polling_manager.go

@@ -18,14 +18,14 @@
 package config
 
 import (
-	"context"
 	"fmt"
 	"sync"
 	"time"
 
 	"github.com/optimizely/go-sdk/optimizely"
 	"github.com/optimizely/go-sdk/optimizely/config/datafileprojectconfig"
 	"github.com/optimizely/go-sdk/optimizely/logging"
+	"github.com/optimizely/go-sdk/optimizely/utils"
 )
 
 const defaultPollingInterval = 5 * time.Minute // default to 5 minutes for polling
@@ -50,7 +50,7 @@ type PollingProjectConfigManager struct {
 	configLock      sync.RWMutex
 	err             error
 
-	ctx context.Context // context used for cancellation
+	exeCtx utils.ExecutionCtx // context used for execution control
 }
 
 func (cm *PollingProjectConfigManager) activate(initialPayload []byte, init bool) {
@@ -89,15 +89,15 @@ func (cm *PollingProjectConfigManager) activate(initialPayload []byte, init bool
 		select {
 		case <-t.C:
 			update()
-		case <-cm.ctx.Done():
+		case <-cm.exeCtx.GetContext().Done():
 			cmLogger.Debug("Polling Config Manager Stopped")
 			return
 		}
 	}
 }
 
 // NewPollingProjectConfigManagerWithOptions returns new instance of PollingProjectConfigManager with the given options
-func NewPollingProjectConfigManagerWithOptions(ctx context.Context, sdkKey string, options PollingProjectConfigManagerOptions) *PollingProjectConfigManager {
+func NewPollingProjectConfigManagerWithOptions(exeCtx utils.ExecutionCtx, sdkKey string, options PollingProjectConfigManagerOptions) *PollingProjectConfigManager {
 
 	var requester Requester
 	if options.Requester != nil {
@@ -112,7 +112,7 @@ func NewPollingProjectConfigManagerWithOptions(ctx context.Context, sdkKey strin
 		pollingInterval = defaultPollingInterval
 	}
 
-	pollingProjectConfigManager := PollingProjectConfigManager{requester: requester, pollingInterval: pollingInterval, ctx: ctx}
+	pollingProjectConfigManager := PollingProjectConfigManager{requester: requester, pollingInterval: pollingInterval, exeCtx: exeCtx}
 
 	pollingProjectConfigManager.activate(options.Datafile, true) // initial poll
 
@@ -122,9 +122,9 @@ func NewPollingProjectConfigManagerWithOptions(ctx context.Context, sdkKey strin
 }
 
 // NewPollingProjectConfigManager returns an instance of the polling config manager with the default configuration
-func NewPollingProjectConfigManager(ctx context.Context, sdkKey string) *PollingProjectConfigManager {
+func NewPollingProjectConfigManager(exeCtx utils.ExecutionCtx, sdkKey string) *PollingProjectConfigManager {
 	options := PollingProjectConfigManagerOptions{}
-	configManager := NewPollingProjectConfigManagerWithOptions(ctx, sdkKey, options)
+	configManager := NewPollingProjectConfigManagerWithOptions(exeCtx, sdkKey, options)
 	return configManager
 }
 

optimizely/config/polling_manager_test.go

@@ -17,7 +17,7 @@
 package config
 
 import (
-	"context"
+	"github.com/optimizely/go-sdk/optimizely/utils"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
@@ -47,7 +47,9 @@ func TestNewPollingProjectConfigManagerWithOptions(t *testing.T) {
 	options := PollingProjectConfigManagerOptions{
 		Requester: mockRequester,
 	}
-	configManager := NewPollingProjectConfigManagerWithOptions(context.Background(), sdkKey, options)
+
+	exeCtx := utils.NewCancelableExecutionCtx()
+	configManager := NewPollingProjectConfigManagerWithOptions(exeCtx, sdkKey, options)
 	mockRequester.AssertExpectations(t)
 
 	actual, err := configManager.GetConfig()
@@ -65,7 +67,8 @@ func TestNewPollingProjectConfigManagerWithNull(t *testing.T) {
 	options := PollingProjectConfigManagerOptions{
 		Requester: mockRequester,
 	}
-	configManager := NewPollingProjectConfigManagerWithOptions(context.Background(), sdkKey, options)
+	exeCtx := utils.NewCancelableExecutionCtx()
+	configManager := NewPollingProjectConfigManagerWithOptions(exeCtx, sdkKey, options)
 	mockRequester.AssertExpectations(t)
 
 	_, err := configManager.GetConfig()

optimizely/event/factory_test.go

@@ -1,14 +1,13 @@
 package event
 
 import (
-	"context"
 	"math/rand"
 	"testing"
 	"time"
 
 	"github.com/optimizely/go-sdk/optimizely"
-
 	"github.com/optimizely/go-sdk/optimizely/entities"
+	"github.com/optimizely/go-sdk/optimizely/utils"
 	"github.com/stretchr/testify/assert"
 )
 
@@ -89,11 +88,10 @@ func BuildTestConversionEvent() UserEvent {
 }
 
 func TestCreateAndSendImpressionEvent(t *testing.T) {
-	ctx := context.Background()
 
 	impressionUserEvent := BuildTestImpressionEvent()
 
-	processor := NewEventProcessor(ctx, 10, 100, 100)
+	processor := NewEventProcessor(utils.NewCancelableExecutionCtx(), 10, 100, 100)
 
 	processor.ProcessEvent(impressionUserEvent)
 
@@ -105,11 +103,10 @@ func TestCreateAndSendImpressionEvent(t *testing.T) {
 }
 
 func TestCreateAndSendConversionEvent(t *testing.T) {
-	ctx := context.Background()
 
 	conversionUserEvent := BuildTestConversionEvent()
 
-	processor := NewEventProcessor(ctx, 10, 100, 100)
+	processor := NewEventProcessor(utils.NewCancelableExecutionCtx(), 10, 100, 100)
 
 	processor.ProcessEvent(conversionUserEvent)
 

optimizely/event/processor.go

@@ -20,6 +20,7 @@ package event
 import (
 	"context"
 	"errors"
+	"github.com/optimizely/go-sdk/optimizely/utils"
 	"sync"
 	"time"
 
@@ -40,6 +41,8 @@ type QueueingEventProcessor struct {
 	Mux             sync.Mutex
 	Ticker          *time.Ticker
 	EventDispatcher Dispatcher
+
+	wg *sync.WaitGroup
 }
 
 // DefaultBatchSize holds the default value for the batch size
@@ -54,19 +57,21 @@ const DefaultEventFlushInterval = 30 * time.Second
 var pLogger = logging.GetLogger("EventProcessor")
 
 // NewEventProcessor returns a new instance of QueueingEventProcessor with queueSize and flushInterval
-func NewEventProcessor(ctx context.Context, batchSize, queueSize int, flushInterval time.Duration) *QueueingEventProcessor {
+func NewEventProcessor(exeCtx utils.ExecutionCtx, batchSize, queueSize int, flushInterval time.Duration) *QueueingEventProcessor {
 	p := &QueueingEventProcessor{
 		MaxQueueSize:    queueSize,
 		FlushInterval:   flushInterval,
 		Q:               NewInMemoryQueue(queueSize),
 		EventDispatcher: &HTTPEventDispatcher{},
+
+		wg: exeCtx.GetWaitSync(),
 	}
 	p.BatchSize = DefaultBatchSize
 	if batchSize > 0 {
 		p.BatchSize = batchSize
 	}
 
-	p.StartTicker(ctx)
+	p.StartTicker(exeCtx.GetContext())
 	return p
 }
 
@@ -102,7 +107,10 @@ func (p *QueueingEventProcessor) StartTicker(ctx context.Context) {
 		return
 	}
 	p.Ticker = time.NewTicker(p.FlushInterval * time.Millisecond)
+	p.wg.Add(1)
 	go func() {
+
+		defer p.wg.Done()
 		for {
 			select {
 			case <-p.Ticker.C: