added blocking call to flush dispatcher

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

@@ -23,6 +23,7 @@ import (
 	"fmt"
 	"runtime/debug"
 	"strconv"
+	"sync"
 
 	"github.com/optimizely/go-sdk/optimizely/event"
 
@@ -42,6 +43,7 @@ type OptimizelyClient struct {
 	isValid         bool
 
 	cancelFunc context.CancelFunc
+	wg         *sync.WaitGroup
 }
 
 // IsFeatureEnabled returns true if the feature is enabled for the given user
@@ -280,4 +282,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.wg.Wait() //blocking call
 }

optimizely/client/factory.go

@@ -20,6 +20,7 @@ package client
 import (
 	"context"
 	"errors"
+	"sync"
 
 	"github.com/optimizely/go-sdk/optimizely/event"
 
@@ -76,8 +77,12 @@ func (f OptimizelyFactory) StaticClient() (*OptimizelyClient, error) {
 
 // ClientWithOptions returns a client initialized with the given configuration options
 func (f OptimizelyFactory) ClientWithOptions(clientOptions Options) (*OptimizelyClient, error) {
+
+	var wg sync.WaitGroup
+
 	client := &OptimizelyClient{
 		isValid: false,
+		wg:      &wg,
 	}
 
 	var ctx context.Context
@@ -117,7 +122,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(ctx, event.DefaultBatchSize, event.DefaultEventQueueSize, event.DefaultEventFlushInterval, &wg)
 	}
 
 	client.isValid = true

optimizely/event/processor.go

@@ -40,6 +40,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,18 +56,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(ctx context.Context, batchSize, queueSize int, flushInterval time.Duration, wg *sync.WaitGroup) *QueueingEventProcessor {
 	p := &QueueingEventProcessor{
 		MaxQueueSize:    queueSize,
 		FlushInterval:   flushInterval,
 		Q:               NewInMemoryQueue(queueSize),
 		EventDispatcher: &HTTPEventDispatcher{},
+
+		wg: wg,
 	}
 	p.BatchSize = DefaultBatchSize
 	if batchSize > 0 {
 		p.BatchSize = batchSize
 	}
 
+	p.wg.Add(1)
 	p.StartTicker(ctx)
 	return p
 }
@@ -103,6 +108,8 @@ func (p *QueueingEventProcessor) StartTicker(ctx context.Context) {
 	}
 	p.Ticker = time.NewTicker(p.FlushInterval * time.Millisecond)
 	go func() {
+
+		defer p.wg.Done()
 		for {
 			select {
 			case <-p.Ticker.C: