micro/sync/store/cache.go

111 lines
2.7 KiB
Go
Raw Normal View History

// Package store syncs multiple go-micro stores
package store
import (
"fmt"
"sync"
"time"
"github.com/ef-ds/deque"
"github.com/micro/go-micro/v2/store"
"github.com/pkg/errors"
)
// Cache implements a cache in front of go-micro Stores
type Cache interface {
store.Store
// Force a full sync
Sync() error
}
type cache struct {
sOptions store.Options
cOptions Options
pendingWrites []*deque.Deque
pendingWriteTickers []*time.Ticker
sync.RWMutex
}
// NewCache returns a new Cache
func NewCache(opts ...Option) Cache {
c := &cache{}
for _, o := range opts {
o(&c.cOptions)
}
if c.cOptions.SyncInterval == 0 {
c.cOptions.SyncInterval = 1 * time.Minute
}
if c.cOptions.SyncMultiplier == 0 {
c.cOptions.SyncMultiplier = 5
}
return c
}
// Init initialises the storeOptions
func (c *cache) Init(opts ...store.Option) error {
for _, o := range opts {
o(&c.sOptions)
}
if len(c.cOptions.Stores) == 0 {
return errors.New("the cache has no stores")
}
if c.sOptions.Context == nil {
return errors.New("please provide a context to the cache. Cancelling the context signals that the cache is being disposed and syncs the cache")
}
for _, s := range c.cOptions.Stores {
if err := s.Init(); err != nil {
return errors.Wrapf(err, "Store %s failed to Init()", s.String())
}
}
c.pendingWrites = make([]*deque.Deque, len(c.cOptions.Stores)-1)
c.pendingWriteTickers = make([]*time.Ticker, len(c.cOptions.Stores)-1)
for i := 0; i < len(c.pendingWrites); i++ {
c.pendingWrites[i] = deque.New()
c.pendingWrites[i].Init()
c.pendingWriteTickers[i] = time.NewTicker(c.cOptions.SyncInterval * time.Duration(intpow(c.cOptions.SyncMultiplier, int64(i))))
}
go c.cacheManager()
return nil
}
// Options returns the cache's store options
func (c *cache) Options() store.Options {
return c.sOptions
}
// String returns a printable string describing the cache
func (c *cache) String() string {
backends := make([]string, len(c.cOptions.Stores))
for i, s := range c.cOptions.Stores {
backends[i] = s.String()
}
return fmt.Sprintf("cache %v", backends)
}
func (c *cache) List(opts ...store.ListOption) ([]string, error) {
return c.cOptions.Stores[0].List(opts...)
}
func (c *cache) Read(key string, opts ...store.ReadOption) ([]*store.Record, error) {
return c.cOptions.Stores[0].Read(key, opts...)
}
func (c *cache) Write(r *store.Record, opts ...store.WriteOption) error {
return c.cOptions.Stores[0].Write(r, opts...)
}
// Delete removes a key from the cache
func (c *cache) Delete(key string, opts ...store.DeleteOption) error {
return c.cOptions.Stores[0].Delete(key, opts...)
}
func (c *cache) Sync() error {
return nil
}
type internalRecord struct {
key string
value []byte
expiresAt time.Time
}