Streaming connections are now closed when you call Close on a client

Author: jcox250

client/client.go

@@ -3,6 +3,7 @@ package client
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"net/http"
 	"strings"
@@ -46,7 +47,6 @@ type CfClient struct {
 	config              *config
 	environmentID       string
 	token               string
-	cancelFunc          context.CancelFunc
 	streamConnected     bool
 	streamConnectedLock sync.RWMutex
 	authenticated       chan struct{}
@@ -55,17 +55,14 @@ type CfClient struct {
 	initializedLock     sync.RWMutex
 	analyticsService    *analyticsservice.AnalyticsService
 	clusterIdentifier   string
+	stop                chan struct{}
+	stopped             bool
 }
 
 // NewCfClient creates a new client instance that connects to CF with the default configuration.
 // For advanced configuration options use ConfigOptions functions
 func NewCfClient(sdkKey string, options ...ConfigOption) (*CfClient, error) {
 
-	var (
-		ctx context.Context
-		err error
-	)
-
 	//  functional options for config
 	config := newDefaultConfig()
 	for _, opt := range options {
@@ -81,8 +78,8 @@ func NewCfClient(sdkKey string, options ...ConfigOption) (*CfClient, error) {
 		analyticsService:  analyticsService,
 		clusterIdentifier: "1",
 		postEvalChan:      make(chan evaluation.PostEvalData),
+		stop:              make(chan struct{}),
 	}
-	ctx, client.cancelFunc = context.WithCancel(context.Background())
 
 	if sdkKey == "" {
 		return client, types.ErrSdkCantBeEmpty
@@ -97,13 +94,20 @@ func NewCfClient(sdkKey string, options ...ConfigOption) (*CfClient, error) {
 		return nil, err
 	}
 
-	go client.initAuthentication(ctx)
-
-	go client.setAnalyticsServiceClient(ctx)
+	client.start()
+	return client, nil
+}
 
-	go client.pullCronJob(ctx)
+func (c *CfClient) start() {
+	ctx, cancel := context.WithCancel(context.Background())
+	go func() {
+		<-c.stop
+		cancel()
+	}()
 
-	return client, nil
+	go c.initAuthentication(ctx)
+	go c.setAnalyticsServiceClient(ctx)
+	go c.pullCronJob(ctx)
 }
 
 // PostEvaluateProcessor push the data to the analytics service
@@ -432,7 +436,14 @@ func (c *CfClient) JSONVariation(key string, target *evaluation.Target, defaultV
 // Close shuts down the Feature Flag client. After calling this, the client
 // should no longer be used
 func (c *CfClient) Close() error {
-	c.cancelFunc()
+	c.mux.Lock()
+	defer c.mux.Unlock()
+	if c.stopped {
+		return errors.New("client already closed")
+	}
+	close(c.stop)
+
+	c.stopped = true
 	return nil
 }
 

client/client_test.go

@@ -199,3 +199,16 @@ var FeatureConfigsResponse = func(req *http.Request) (*http.Response, error) {
 
 	return httpmock.NewJsonResponse(200, FeatureConfigResponse)
 }
+
+func TestCfClient_Close(t *testing.T) {
+	client, err := newClient(&http.Client{})
+	if err != nil {
+		t.Error(err)
+	}
+
+	t.Log("When I close the client for the first time I should not get an error")
+	assert.Nil(t, client.Close())
+
+	t.Log("When I close the client for the second time I should an error")
+	assert.NotNil(t, client.Close())
+}