diff options
author | Grzegorz Bizon <grzesiek.bizon@gmail.com> | 2019-11-14 14:39:46 +0300 |
---|---|---|
committer | Grzegorz Bizon <grzesiek.bizon@gmail.com> | 2019-11-14 14:39:46 +0300 |
commit | 4621a44e1b7a2afa9ccb7f6bd1a0aec7f0215b4d (patch) | |
tree | 60924e0cdf77d6d2e45b98954d3086f212576818 | |
parent | fd0e417eed8f73e6316482a2538074ecf31c3923 (diff) |
Simplify concurrency in retrieving gitlab source response
-rw-r--r-- | internal/source/gitlab/cache/cache.go | 4 | ||||
-rw-r--r-- | internal/source/gitlab/cache/cache_test.go | 4 | ||||
-rw-r--r-- | internal/source/gitlab/cache/entry.go | 39 | ||||
-rw-r--r-- | internal/source/gitlab/cache/retriever.go | 30 |
4 files changed, 36 insertions, 41 deletions
diff --git a/internal/source/gitlab/cache/cache.go b/internal/source/gitlab/cache/cache.go index b3749450..637f27e3 100644 --- a/internal/source/gitlab/cache/cache.go +++ b/internal/source/gitlab/cache/cache.go @@ -32,7 +32,5 @@ func (c *Cache) Resolve(ctx context.Context, domain string) *Lookup { return entry.Lookup() } - <-entry.Retrieve(ctx, c.client) - - return entry.Lookup() + return entry.Retrieve(ctx, c.client) } diff --git a/internal/source/gitlab/cache/cache_test.go b/internal/source/gitlab/cache/cache_test.go index 3a5a2e8f..53e61582 100644 --- a/internal/source/gitlab/cache/cache_test.go +++ b/internal/source/gitlab/cache/cache_test.go @@ -67,9 +67,7 @@ func (cache *Cache) withTestEntry(config entryConfig, block func(*Entry)) { entry := cache.store.LoadOrCreate(domain) if config.retrieved { - newResponse := make(chan Lookup, 1) - newResponse <- Lookup{Domain: Domain{Name: domain}, Status: 200} - entry.setResponse(newResponse) + entry.setResponse(Lookup{Domain: Domain{Name: domain}, Status: 200}) } if config.expired { diff --git a/internal/source/gitlab/cache/entry.go b/internal/source/gitlab/cache/entry.go index 01f44a47..67a1a03a 100644 --- a/internal/source/gitlab/cache/entry.go +++ b/internal/source/gitlab/cache/entry.go @@ -2,6 +2,7 @@ package cache import ( "context" + "errors" "sync" "time" ) @@ -17,7 +18,7 @@ var ( type Entry struct { domain string created time.Time - fetch *sync.Once + retrieve *sync.Once refresh *sync.Once mux *sync.RWMutex retrieved chan struct{} @@ -28,7 +29,7 @@ func newCacheEntry(domain string) *Entry { return &Entry{ domain: domain, created: time.Now(), - fetch: &sync.Once{}, + retrieve: &sync.Once{}, refresh: &sync.Once{}, mux: &sync.RWMutex{}, retrieved: make(chan struct{}), @@ -61,17 +62,22 @@ func (e *Entry) Lookup() *Lookup { return e.response } -// Retrieve schedules a retrieval of a response. It returns a channel that is -// going to be closed when retrieval is done, either successfully or not. -func (e *Entry) Retrieve(ctx context.Context, client Resolver) <-chan struct{} { - // TODO create context with timeout - e.fetch.Do(func() { - retriever := Retriever{client: client, ctx: ctx, timeout: retrievalTimeout} +// Retrieve perform a blocking retrieval of the cache entry response. +func (e *Entry) Retrieve(ctx context.Context, client Resolver) *Lookup { + newctx, cancelctx := context.WithTimeout(ctx, retrievalTimeout) + defer cancelctx() - go e.setResponse(retriever.Retrieve(e.domain)) - }) + e.retrieve.Do(func() { go e.retrieveWithClient(client) }) + + select { + case <-newctx.Done(): + return &Lookup{Status: 502, Error: errors.New("context done")} + case <-e.retrieved: + return e.Lookup() + } - return e.retrieved + // This should not happen + return &Lookup{Status: 500, Error: errors.New("retrieval error")} } // Refresh will update the entry in the store only when it gets resolved. @@ -80,21 +86,24 @@ func (e *Entry) Refresh(client Resolver, store Store) { go func() { entry := newCacheEntry(e.domain) - <-entry.Retrieve(context.Background(), client) + entry.Retrieve(context.Background(), client) store.ReplaceOrCreate(e.domain, entry) }() }) } -func (e *Entry) setResponse(response <-chan Lookup) { - lookup := <-response +func (e *Entry) retrieveWithClient(client Resolver) { + retriever := Retriever{client: client, timeout: retrievalTimeout} + e.setResponse(retriever.Retrieve(e.domain)) +} + +func (e *Entry) setResponse(lookup Lookup) { e.mux.Lock() defer e.mux.Unlock() e.response = &lookup - close(e.retrieved) } diff --git a/internal/source/gitlab/cache/retriever.go b/internal/source/gitlab/cache/retriever.go index de766aa4..e57565ae 100644 --- a/internal/source/gitlab/cache/retriever.go +++ b/internal/source/gitlab/cache/retriever.go @@ -13,42 +13,32 @@ var maxRetrievalInterval = time.Second // case of errors type Retriever struct { client Resolver - ctx context.Context timeout time.Duration } -// Retrieve schedules a retrieval of a response and return a channel that the -// response is going to be sent to -func (r *Retriever) Retrieve(domain string) <-chan Lookup { - response := make(chan Lookup) - - go r.retrieveWithTimeout(domain, response) - - return response -} - -func (r *Retriever) retrieveWithTimeout(domain string, response chan<- Lookup) { - newctx, cancel := context.WithTimeout(r.ctx, r.timeout) +// Retrieve retrieves a lookup response from external source with timeout and +// backoff. It has its own context with timeout. +func (r *Retriever) Retrieve(domain string) Lookup { + ctx, cancel := context.WithTimeout(context.Background(), r.timeout) defer cancel() var lookup Lookup select { - case <-newctx.Done(): - response <- Lookup{Status: 502, Error: errors.New("context done")} + case <-ctx.Done(): fmt.Println("retrieval context done") // TODO logme - case lookup = <-r.resolveWithBackoff(newctx, domain): - response <- lookup + lookup = Lookup{Status: 502, Error: errors.New("retrieval context done")} + case lookup = <-r.resolveWithBackoff(ctx, domain): fmt.Println("retrieval response sent") // TODO logme } - close(response) + return lookup } func (r *Retriever) resolveWithBackoff(ctx context.Context, domain string) <-chan Lookup { response := make(chan Lookup) - go func(response chan<- Lookup) { + go func() { var lookup Lookup for i := 1; i <= 3; i++ { @@ -63,7 +53,7 @@ func (r *Retriever) resolveWithBackoff(ctx context.Context, domain string) <-cha response <- lookup close(response) - }(response) + }() return response } |