Repository: Unleash/unleash-client-go
Pull request #51 (2019-09-14 20:30:15)

Make go test -race ./... pass
This PR makes Client.Close return *after* all resources have been freed
that's used by the client. In particular, when Client.Close returns all
goroutines started by the client have exited, any errors and warnings reported
by the client have been sent to their respective channels or listeners, and no
HTTP requests started by the client will still be in flight.

https://github.com/Unleash/unleash-client-go/issues/17 isn't fixed by this change, but I believe it's a step in the right direction.


Patch:
diff --git a/README.md b/README.md
index b09f048..6b4a0c5 100644
--- a/README.md
+++ b/README.md
@@ -106,6 +106,6 @@ Run code-style checks:(currently failing)
     
     make strict-check
     
-Run race-tests(currently failing):
+Run race-tests:
  
     make test-all
diff --git a/client.go b/client.go
index df78f72..c490a57 100644
--- a/client.go
+++ b/client.go
@@ -2,11 +2,12 @@ package unleash
 
 import (
 	"fmt"
-	s "github.com/Unleash/unleash-client-go/v3/internal/strategies"
-	"github.com/Unleash/unleash-client-go/v3/strategy"
 	"net/url"
 	"strings"
 	"time"
+
+	s "github.com/Unleash/unleash-client-go/v3/internal/strategies"
+	"github.com/Unleash/unleash-client-go/v3/strategy"
 )
 
 const (
@@ -37,6 +38,7 @@ type Client struct {
 	repositoryListener RepositoryListener
 	ready              chan bool
 	onReady            chan struct{}
+	close              chan struct{}
 	closed             chan struct{}
 	count              chan metric
 	sent               chan MetricsData
@@ -90,6 +92,7 @@ func NewClient(options ...ConfigOption) (*Client, error) {
 		count:         make(chan metric),
 		sent:          make(chan MetricsData),
 		registered:    make(chan ClientData, 1),
+		close:         make(chan struct{}),
 		closed:        make(chan struct{}),
 	}
 
@@ -212,7 +215,8 @@ func (uc *Client) sync() {
 			if uc.metricsListener != nil {
 				uc.metricsListener.OnRegistered(cd)
 			}
-		case <-uc.closed:
+		case <-uc.close:
+			close(uc.closed)
 			return
 		}
 	}
@@ -263,7 +267,11 @@ func (uc Client) IsEnabled(feature string, options ...FeatureOption) (enabled bo
 func (uc *Client) Close() error {
 	uc.repository.Close()
 	uc.metrics.Close()
-	close(uc.closed)
+	if uc.options.listener != nil {
+		// Wait for sync to exit.
+		close(uc.close)
+		<-uc.closed
+	}
 	return nil
 }
 
diff --git a/metrics.go b/metrics.go
index bb915f8..0e3c3ed 100644
--- a/metrics.go
+++ b/metrics.go
@@ -2,6 +2,7 @@ package unleash
 
 import (
 	"bytes"
+	"context"
 	"encoding/json"
 	"fmt"
 	"net/http"
@@ -60,8 +61,11 @@ type metrics struct {
 	started  time.Time
 	bucketMu sync.Mutex
 	bucket   api.Bucket
-	stopped  chan bool
 	timer    *time.Timer
+	close    chan struct{}
+	closed   chan struct{}
+	ctx      context.Context
+	cancel   func()
 }
 
 func newMetrics(options metricsOptions, channels metricsChannels) *metrics {
@@ -69,8 +73,12 @@ func newMetrics(options metricsOptions, channels metricsChannels) *metrics {
 		metricsChannels: channels,
 		options:         options,
 		started:         time.Now(),
-		stopped:         make(chan bool),
+		close:           make(chan struct{}),
+		closed:          make(chan struct{}),
 	}
+	ctx, cancel := context.WithCancel(context.Background())
+	m.ctx = ctx
+	m.cancel = cancel
 
 	if m.options.httpClient == nil {
 		m.options.httpClient = http.DefaultClient
@@ -80,7 +88,10 @@ func newMetrics(options metricsOptions, channels metricsChannels) *metrics {
 	m.resetBucket()
 	m.bucketMu.Unlock()
 
-	if m.options.metricsInterval > 0 {
+	if m.options.metricsInterval <= 0 {
+		m.options.disableMetrics = true
+	}
+	if !m.options.disableMetrics {
 		m.startTimer()
 		m.registerInstance()
 		go m.sync()
@@ -90,47 +101,32 @@ func newMetrics(options metricsOptions, channels metricsChannels) *metrics {
 }
 
 func (m *metrics) Close() error {
-	m.stop()
+	if !m.options.disableMetrics {
+		m.timer.Stop()
+		m.cancel()
+		close(m.close)
+		<-m.closed
+	}
 	return nil
 }
 
 func (m *metrics) startTimer() {
-	if m.options.disableMetrics {
-		return
-	}
-
 	m.timer = time.NewTimer(m.options.metricsInterval)
 }
 
-func (m *metrics) stop() {
-	if !m.timer.Stop() {
-		<-m.timer.C
-	}
-	m.stopped <- true
-}
-
 func (m *metrics) sync() {
-	if m.options.disableMetrics {
-		return
-	}
-
 	for {
 		select {
 		case <-m.timer.C:
 			m.sendMetrics()
-		case <-m.stopped:
-			m.options.disableMetrics = true
+		case <-m.close:
+			close(m.closed)
 			return
 		}
 	}
-
 }
 
 func (m *metrics) registerInstance() {
-	if m.options.disableMetrics {
-		return
-	}
-
 	u, _ := m.options.url.Parse("./client/register")
 	payload := m.getClientData()
 	resp, err := m.doPost(u, payload)
@@ -149,10 +145,6 @@ func (m *metrics) registerInstance() {
 }
 
 func (m *metrics) sendMetrics() {
-	if m.options.disableMetrics {
-		return
-	}
-
 	m.bucketMu.Lock()
 	if m.bucket.IsEmpty() {
 		m.resetBucket()
@@ -175,7 +167,7 @@ func (m *metrics) sendMetrics() {
 
 	if resp.StatusCode == http.StatusNotFound {
 		m.warn(fmt.Errorf("%s return 404, stopping metrics", u.String()))
-		m.stop()
+		m.Close()
 		return
 	}
 
@@ -197,6 +189,7 @@ func (m *metrics) doPost(url *url.URL, payload interface{}) (*http.Response, err
 	if err != nil {
 		return nil, err
 	}
+	req = req.WithContext(m.ctx)
 	req.Header.Set("Content-Type", "application/json")
 	req.Header.Add("UNLEASH-APPNAME", m.options.appName)
 	req.Header.Add("UNLEASH-INSTANCEID", m.options.instanceId)
diff --git a/repository.go b/repository.go
index 4ca01c2..59e8078 100644
--- a/repository.go
+++ b/repository.go
@@ -1,6 +1,7 @@
 package unleash
 
 import (
+	"context"
 	"encoding/json"
 	"net/http"
 	"sync"
@@ -15,6 +16,9 @@ type repository struct {
 	options repositoryOptions
 	etag    string
 	close   chan struct{}
+	closed  chan struct{}
+	ctx     context.Context
+	cancel  func()
 }
 
 func newRepository(options repositoryOptions, channels repositoryChannels) *repository {
@@ -22,7 +26,11 @@ func newRepository(options repositoryOptions, channels repositoryChannels) *repo
 		options:            options,
 		repositoryChannels: channels,
 		close:              make(chan struct{}),
+		closed:             make(chan struct{}),
 	}
+	ctx, cancel := context.WithCancel(context.Background())
+	repo.ctx = ctx
+	repo.cancel = cancel
 
 	if options.httpClient == nil {
 		repo.options.httpClient = http.DefaultClient
@@ -40,8 +48,6 @@ func newRepository(options repositoryOptions, channels repositoryChannels) *repo
 }
 
 func (r *repository) sync() {
-	defer r.cleanup()
-
 	r.fetch()
 	r.ready <- true
 
@@ -50,6 +56,10 @@ func (r *repository) sync() {
 
 		select {
 		case <-r.close:
+			if err := r.options.storage.Persist(); err != nil {
+				r.err(err)
+			}
+			close(r.closed)
 			return
 		case <-refreshTimer.C:
 			r.fetch()
@@ -57,12 +67,6 @@ func (r *repository) sync() {
 	}
 }
 
-func (r *repository) cleanup() {
-	if err := r.options.storage.Persist(); err != nil {
-		r.err(err)
-	}
-}
-
 func (r *repository) fetch() {
 	u, _ := r.options.url.Parse("./client/features")
 
@@ -71,6 +75,7 @@ func (r *repository) fetch() {
 		r.err(err)
 		return
 	}
+	req = req.WithContext(r.ctx)
 
 	req.Header.Add("UNLEASH-APPNAME", r.options.appName)
 	req.Header.Add("UNLEASH-INSTANCEID", r.options.instanceId)
@@ -123,5 +128,7 @@ func (r *repository) getToggle(key string) *api.Feature {
 
 func (r *repository) Close() error {
 	close(r.close)
+	r.cancel()
+	<-r.closed
 	return nil
 }


Test patch:
diff --git a/client_test.go b/client_test.go
new file mode 100644
index 0000000..4718d25
--- /dev/null
+++ b/client_test.go
@@ -0,0 +1,47 @@
+package unleash
+
+import (
+	"testing"
+
+	"github.com/Unleash/unleash-client-go/v3/internal/api"
+	"github.com/stretchr/testify/assert"
+	"gopkg.in/h2non/gock.v1"
+)
+
+func TestClientWithoutListener(t *testing.T) {
+	assert := assert.New(t)
+	defer gock.OffAll()
+
+	gock.New(mockerServer).
+		Post("/client/register").
+		Reply(200)
+
+	gock.New(mockerServer).
+		Get("/client/features").
+		Reply(200).
+		JSON(api.FeatureResponse{})
+
+	client, err := NewClient(
+		WithUrl(mockerServer),
+		WithAppName(mockAppName),
+		WithInstanceId(mockInstanceId),
+	)
+	assert.Nil(err, "client should not return an error")
+
+	go func() {
+		for {
+			select {
+			case e := <-client.Errors():
+				t.Fatalf("Unexpected error: %v", e)
+			case w := <-client.Warnings():
+				t.Fatalf("Unexpected warning: %v", w)
+			case <-client.Count():
+			case <-client.Sent():
+			}
+		}
+	}()
+	<-client.Registered()
+	<-client.Ready()
+	client.Close()
+	assert.True(gock.IsDone(), "there should be no more mocks")
+}
diff --git a/metrics_test.go b/metrics_test.go
index 85a62de..6bb759a 100644
--- a/metrics_test.go
+++ b/metrics_test.go
@@ -1,18 +1,19 @@
 package unleash
 
 import (
+	"net/url"
+	"testing"
+	"time"
+
 	"github.com/Unleash/unleash-client-go/v3/internal/api"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
 	"gopkg.in/h2non/gock.v1"
-	"net/url"
-	"testing"
-	"time"
 )
 
 func TestMetrics_RegisterInstance(t *testing.T) {
 	assert := assert.New(t)
-	defer gock.Off()
+	defer gock.OffAll()
 
 	gock.New(mockerServer).
 		Post("/client/register").
@@ -29,7 +30,7 @@ func TestMetrics_RegisterInstance(t *testing.T) {
 	mockListener.On("OnReady").Return()
 	mockListener.On("OnRegistered", mock.AnythingOfType("ClientData"))
 
-	_, err := NewClient(
+	client, err := NewClient(
 		WithUrl(mockerServer),
 		WithAppName(mockAppName),
 		WithInstanceId(mockInstanceId),
@@ -37,6 +38,7 @@ func TestMetrics_RegisterInstance(t *testing.T) {
 	)
 
 	time.Sleep(1 * time.Second)
+	client.Close()
 
 	assert.Nil(err, "client should not return an error")
 
@@ -45,22 +47,32 @@ func TestMetrics_RegisterInstance(t *testing.T) {
 
 func TestMetrics_DoPost(t *testing.T) {
 	assert := assert.New(t)
-	defer gock.Off()
+	defer gock.OffAll()
 
 	gock.New(mockerServer).
 		Post("/client/register").
 		Reply(200)
 
+	gock.New(mockerServer).
+		Get("/client/features").
+		Reply(200).
+		JSON(api.FeatureResponse{})
+
 	gock.New(mockerServer).
 		Post("").
 		MatchHeader("UNLEASH-APPNAME", mockAppName).
 		MatchHeader("UNLEASH-INSTANCEID", mockInstanceId).
 		Reply(200)
 
+	mockListener := &MockedListener{}
+	mockListener.On("OnReady").Return()
+	mockListener.On("OnRegistered", mock.AnythingOfType("ClientData"))
+
 	client, err := NewClient(
 		WithUrl(mockerServer),
 		WithAppName(mockAppName),
 		WithInstanceId(mockInstanceId),
+		WithListener(mockListener),
 	)
 
 	assert.Nil(err, "client should not return an error")
@@ -69,9 +81,41 @@ func TestMetrics_DoPost(t *testing.T) {
 
 	serverUrl, _ := url.Parse(mockerServer)
 	res, err := m.doPost(serverUrl, &struct{}{})
+	client.Close()
 
 	assert.Nil(err, "doPost should not return an error")
 	assert.Equal(200, res.StatusCode, "statusCode should be 200")
 	assert.True(gock.IsDone(), "there should be no more mocks")
 }
 
+func TestMetrics_DisabledMetrics(t *testing.T) {
+	assert := assert.New(t)
+	defer gock.OffAll()
+
+	gock.New(mockerServer).
+		Get("/client/features").
+		Reply(200).
+		JSON(api.FeatureResponse{})
+
+	mockListener := &MockedListener{}
+	mockListener.On("OnReady").Return()
+
+	client, err := NewClient(
+		WithUrl(mockerServer),
+		WithDisableMetrics(true),
+		WithMetricsInterval(100*time.Millisecond),
+		WithAppName(mockAppName),
+		WithInstanceId(mockInstanceId),
+		WithListener(mockListener),
+	)
+	assert.Nil(err, "client should not return an error")
+
+	client.WaitForReady()
+	client.IsEnabled("foo")
+	client.IsEnabled("bar")
+	client.IsEnabled("baz")
+
+	time.Sleep(300 * time.Millisecond)
+	client.Close()
+	assert.True(gock.IsDone(), "there should be no more mocks")
+}