diff options
author | Grzegorz Bizon <grzesiek.bizon@gmail.com> | 2019-11-12 16:17:29 +0300 |
---|---|---|
committer | Grzegorz Bizon <grzesiek.bizon@gmail.com> | 2019-11-12 16:17:29 +0300 |
commit | b60ee3425b4b6d54154de1629075b52b8599d869 (patch) | |
tree | a96eae38f4427a45bdbe00af0dbc39656871e001 /internal/source/gitlab/cache/entry.go | |
parent | 4e0be9393f33d27ea381cae3e6a6aeda88032153 (diff) |
Refactor gitlab source cache to make it more thread safe
Diffstat (limited to 'internal/source/gitlab/cache/entry.go')
-rw-r--r-- | internal/source/gitlab/cache/entry.go | 120 |
1 files changed, 120 insertions, 0 deletions
diff --git a/internal/source/gitlab/cache/entry.go b/internal/source/gitlab/cache/entry.go new file mode 100644 index 00000000..65da4779 --- /dev/null +++ b/internal/source/gitlab/cache/entry.go @@ -0,0 +1,120 @@ +package cache + +import ( + "context" + "sync" + "time" +) + +var ( + retrievalTimeout = 5 * time.Second + shortCacheExpiry = 10 * time.Second +) + +// Entry represents a cache object that can be retrieved asynchronously and +// holds a pointer to *Response when the domain lookup has been retrieved +// succesfully +type Entry struct { + domain string + created time.Time + ctx context.Context + cancel context.CancelFunc + fetch *sync.Once + refresh *sync.Once + mux *sync.RWMutex + retrieved chan struct{} + response *Response +} + +func newCacheEntry(ctx context.Context, domain string) *Entry { + newctx, cancel := context.WithCancel(ctx) + + return &Entry{ + domain: domain, + ctx: newctx, + cancel: cancel, + created: time.Now(), + fetch: &sync.Once{}, + refresh: &sync.Once{}, + mux: &sync.RWMutex{}, + retrieved: make(chan struct{}), + } +} + +// IsUpToDate returns true if the entry has been resolved correctly and has not +// expired yet. False otherwise. +func (e *Entry) IsUpToDate() bool { + e.mux.RLock() + defer e.mux.RUnlock() + + return e.isResolved() && !e.isExpired() +} + +// NeedsRefresh return true if the entry has been resolved correctly but it has +// expired since then. +func (e *Entry) NeedsRefresh() bool { + e.mux.RLock() + defer e.mux.RUnlock() + + return e.isResolved() && e.isExpired() +} + +// Lookup return a retrieval response. TODO consider returning *Response. +func (e *Entry) Lookup() (*Lookup, int, error) { + e.mux.RLock() + defer e.mux.RUnlock() + + return e.response.Lookup() +} + +// 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(client Resolver) <-chan struct{} { + e.fetch.Do(func() { + retriever := Retriever{ + client: client, ctx: e.ctx, timeout: retrievalTimeout, + } + + go e.setResponse(retriever.Retrieve(e.domain)) + }) + + return e.retrieved +} + +// Refresh will update the entry in the store only when it gets resolved. +func (e *Entry) Refresh(ctx context.Context, client Resolver, store Store) { + e.refresh.Do(func() { + go func() { + newEntry := newCacheEntry(ctx, e.domain) + + <-newEntry.Retrieve(client) + + store.ReplaceOrCreate(ctx, e.domain) + }() + }) +} + +// CancelContexts cancels all cancelable contexts. Typically used when the +// entry is evicted from cache. +func (e *Entry) CancelContexts() { + e.cancel() +} + +func (e *Entry) setResponse(response <-chan Response) { + newResponse := <-response + + e.mux.Lock() + defer e.mux.Unlock() + + e.response = &newResponse + + close(e.retrieved) +} + +func (e *Entry) isExpired() bool { + return e.created.Add(shortCacheExpiry).Before(time.Now()) +} + +func (e *Entry) isResolved() bool { + return e.response != nil +} |