fix(go-client): parllel schema loads if descriptor accessed from different go routines (#162)

goburrow cache calling loaderFunc multiple times if key is accessed from multiple go routines.
Removed goburrowcache.Implemented simple reloader with Ticker.

Author: h4rikris

clients/go/client.go

@@ -5,9 +5,9 @@ package stencil
 
 import (
 	"encoding/json"
+	"sync"
 	"time"
 
-	"github.com/goburrow/cache"
 	"github.com/pkg/errors"
 	"google.golang.org/protobuf/encoding/protojson"
 	"google.golang.org/protobuf/proto"
@@ -77,22 +77,22 @@ func (o *Options) setDefaults() {
 // NewClient creates stencil client. Downloads proto descriptor file from given url and stores the definitions.
 // It will throw error if download fails or downloaded file is not fully contained descriptor file
 func NewClient(urls []string, options Options) (Client, error) {
-	cacheOptions := []cache.Option{cache.WithMaximumSize(len(urls))}
 	options.setDefaults()
-	if options.AutoRefresh {
-		cacheOptions = append(cacheOptions, cache.WithRefreshAfterWrite(options.RefreshInterval), cache.WithExpireAfterWrite(options.RefreshInterval))
-	}
-	newCache := cache.NewLoadingCache(options.RefreshStrategy.getLoader(options), cacheOptions...)
-	s, err := newStore(urls, newCache)
-	if err != nil {
-		return nil, err
+	stores := []*store{}
+	for _, url := range urls {
+		s, err := newStore(url, options)
+		if err != nil {
+			return nil, err
+		}
+		stores = append(stores, s)
 	}
-	return &stencilClient{urls: urls, store: s, options: options}, nil
+
+	return &stencilClient{urls: urls, stores: stores, options: options}, nil
 }
 
 type stencilClient struct {
 	urls    []string
-	store   *store
+	stores  []*store
 	options Options
 }
 
@@ -133,8 +133,8 @@ func (s *stencilClient) Serialize(className string, data interface{}) (bytes []b
 }
 
 func (s *stencilClient) getMatchingResolver(className string) (*Resolver, bool) {
-	for _, url := range s.urls {
-		resolver, ok := s.store.getResolver(url)
+	for _, store := range s.stores {
+		resolver, ok := store.getResolver()
 		if !ok {
 			return nil, false
 		}
@@ -156,13 +156,21 @@ func (s *stencilClient) GetDescriptor(className string) (protoreflect.MessageDes
 }
 
 func (s *stencilClient) Close() {
-	if s.store != nil {
-		s.store.Close()
+	for _, store := range s.stores {
+		if store != nil {
+			store.Close()
+		}
 	}
 }
 
 func (s *stencilClient) Refresh() {
-	for _, url := range s.urls {
-		s.store.Refresh(url)
+	var wg sync.WaitGroup
+	for _, st := range s.stores {
+		wg.Add(1)
+		go func(s *store) {
+			defer wg.Done()
+			s.refresh()
+		}(st)
 	}
+	wg.Wait()
 }

clients/go/client_test.go

@@ -181,7 +181,7 @@ func TestNewClient(t *testing.T) {
 		}))
 		client, _ := stencil.NewClient([]string{ts.URL}, stencil.Options{AutoRefresh: true, RefreshInterval: 2 * time.Millisecond})
 		// wait for interval to end
-		time.Sleep(2 * time.Millisecond)
+		time.Sleep(3 * time.Millisecond)
 		client.GetDescriptor("test.One")
 		time.Sleep(1 * time.Millisecond)
 		client.Close()
@@ -381,7 +381,7 @@ func TestRefreshStrategies(t *testing.T) {
 		assert.NoError(t, err)
 		assert.NotNil(t, client)
 		// wait for refresh interval
-		time.Sleep(2 * time.Millisecond)
+		time.Sleep(3 * time.Millisecond)
 		desc, err := client.GetDescriptor("test.stencil.One")
 		assert.Nil(t, err)
 		assert.NotNil(t, desc)
@@ -392,7 +392,7 @@ func TestRefreshStrategies(t *testing.T) {
 		// simulates version update
 		versions = `{"versions": [1,2]}`
 		// wait for refresh interval
-		time.Sleep(2 * time.Millisecond)
+		time.Sleep(3 * time.Millisecond)
 		desc, err = client.GetDescriptor("test.stencil.One")
 		assert.Nil(t, err)
 		assert.NotNil(t, desc)

clients/go/go.mod

@@ -3,8 +3,6 @@ module github.com/odpf/stencil/clients/go
 go 1.16
 
 require (
-	github.com/davecgh/go-spew v1.1.1 // indirect
-	github.com/goburrow/cache v0.1.4
 	github.com/pkg/errors v0.9.1
 	github.com/stretchr/testify v1.7.0
 	google.golang.org/protobuf v1.26.0

clients/go/go.sum

@@ -1,8 +1,5 @@
+github.com/davecgh/go-spew v1.1.0 h1:ZDRjVQ15GmhC3fiQ8ni8+OwkZQO4DARzQgrnXU1Liz8=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
-github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
-github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
-github.com/goburrow/cache v0.1.4 h1:As4KzO3hgmzPlnaMniZU9+VmoNYseUhuELbxy9mRBfw=
-github.com/goburrow/cache v0.1.4/go.mod h1:cDFesZDnIlrHoNlMYqqMpCRawuXulgx+y7mXU8HZ+/c=
 github.com/golang/protobuf v1.5.0/go.mod h1:FsONVRAS9T7sI+LIUmWTfcYkHO4aIWwzhcaSAoJOfIk=
 github.com/google/go-cmp v0.5.5 h1:Khx7svrCpmxxtHBq5j2mp/xVjsi8hQMfNLvJFAlrGgU=
 github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=

clients/go/refresh_strategy.go

@@ -4,8 +4,6 @@ import (
 	"encoding/json"
 	"fmt"
 	"strings"
-
-	"github.com/goburrow/cache"
 )
 
 // RefreshStrategy clients can configure which refresh strategy to use to download latest schema.
@@ -21,7 +19,7 @@ const (
 	VersionBasedRefresh
 )
 
-func (r RefreshStrategy) getLoader(opts Options) cache.LoaderFunc {
+func (r RefreshStrategy) getLoader(opts Options) loaderFunc {
 	switch r {
 	case VersionBasedRefresh:
 		return versionBasedRefresh(opts)
@@ -42,9 +40,8 @@ func loadFromURL(url string, opts Options) (*Resolver, error) {
 	return NewResolver(data)
 }
 
-func longPollingRefresh(opts Options) cache.LoaderFunc {
-	return func(k cache.Key) (cache.Value, error) {
-		url := k.(string)
+func longPollingRefresh(opts Options) loaderFunc {
+	return func(url string) (*Resolver, error) {
 		return loadFromURL(url, opts)
 	}
 }
@@ -53,11 +50,10 @@ type versionsModel struct {
 	Versions []int `json:"versions"`
 }
 
-func versionBasedRefresh(opts Options) cache.LoaderFunc {
+func versionBasedRefresh(opts Options) loaderFunc {
 	lastVersion := 0
 	logger := wrapLogger(opts.Logger)
-	return func(k cache.Key) (cache.Value, error) {
-		url := k.(string)
+	return func(url string) (*Resolver, error) {
 		versionsURL := fmt.Sprintf("%s/versions", strings.TrimRight(url, "/"))
 		data, err := downloader(versionsURL, opts.HTTPOptions)
 		if err != nil {

clients/go/store.go

@@ -1,27 +1,93 @@
 package stencil
 
 import (
-	"github.com/goburrow/cache"
+	"io"
+	"sync"
+	"time"
 )
 
+type loaderFunc func(string) (*Resolver, error)
+type timer struct {
+	ticker *time.Ticker
+	done   chan bool
+}
+
+func (t *timer) Close() error {
+	t.ticker.Stop()
+	t.done <- true
+	return nil
+}
+
+func setInterval(d time.Duration, f func(), waitForReader <-chan bool) io.Closer {
+	ticker := time.NewTicker(d)
+	done := make(chan bool)
+	go (func() {
+		for {
+			select {
+			case <-done:
+				return
+			case <-ticker.C:
+				// wait for access
+				refresh := <-waitForReader
+				if refresh {
+					f()
+				}
+			}
+		}
+	})()
+	return &timer{ticker: ticker, done: done}
+}
+
 type store struct {
-	cache.LoadingCache
+	autoRefresh bool
+	timer       io.Closer
+	access      chan bool
+	loader      loaderFunc
+	url         string
+	data        *Resolver
+	lock        sync.RWMutex
 }
 
-func (s *store) getResolver(key string) (*Resolver, bool) {
-	val, err := s.Get(key)
-	if err != nil {
-		return nil, false
+func (s *store) refresh() {
+	val, err := s.loader(s.url)
+	if err == nil {
+		s.lock.Lock()
+		defer s.lock.Unlock()
+		s.data = val
 	}
-	return val.(*Resolver), true
 }
 
-func newStore(urls []string, loadingCache cache.LoadingCache) (*store, error) {
-	s := &store{loadingCache}
-	for _, url := range urls {
-		if _, err := loadingCache.Get(url); err != nil {
-			return s, err
-		}
+func (s *store) notify() {
+	select {
+	case s.access <- true:
+	default:
+	}
+}
+
+func (s *store) getResolver() (*Resolver, bool) {
+	s.notify()
+	s.lock.RLock()
+	defer s.lock.RUnlock()
+	return s.data, s.data != nil
+}
+
+func (s *store) Close() {
+	close(s.access)
+	if s.timer != nil {
+		s.timer.Close()
+	}
+}
+
+func newStore(url string, options Options) (*store, error) {
+	loader := options.RefreshStrategy.getLoader(options)
+	s := &store{loader: loader, access: make(chan bool), url: url, autoRefresh: options.AutoRefresh}
+	val, err := loader(url)
+	if err != nil {
+		return s, err
+	}
+	s.data = val
+	if options.AutoRefresh {
+		s.timer = setInterval(options.RefreshInterval, s.refresh, s.access)
 	}
 	return s, nil
 }