// Package syncs will sync multiple stores package sync import ( "context" "fmt" "sync" "time" "github.com/ef-ds/deque" "github.com/unistack-org/micro/v3/store" ) // Sync implements a sync in for stores type Sync interface { // Implements the store interface store.Store // Force a full sync Sync() error } type syncStore struct { storeOpts store.Options syncOpts Options pendingWrites []*deque.Deque pendingWriteTickers []*time.Ticker sync.RWMutex } // NewSync returns a new Sync func NewSync(opts ...Option) Sync { c := &syncStore{} for _, o := range opts { o(&c.syncOpts) } if c.syncOpts.SyncInterval == 0 { c.syncOpts.SyncInterval = 1 * time.Minute } if c.syncOpts.SyncMultiplier == 0 { c.syncOpts.SyncMultiplier = 5 } return c } func (c *syncStore) Connect(ctx context.Context) error { return nil } func (c *syncStore) Disconnect(ctx context.Context) error { return nil } func (c *syncStore) Close(ctx context.Context) error { return nil } // Init initialises the storeOptions func (c *syncStore) Init(opts ...store.Option) error { for _, o := range opts { o(&c.storeOpts) } if len(c.syncOpts.Stores) == 0 { return fmt.Errorf("the sync has no stores") } for _, s := range c.syncOpts.Stores { if err := s.Init(); err != nil { return fmt.Errorf("Store %s failed to Init(): %w", s.String(), err) } } c.pendingWrites = make([]*deque.Deque, len(c.syncOpts.Stores)-1) c.pendingWriteTickers = make([]*time.Ticker, len(c.syncOpts.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.syncOpts.SyncInterval * time.Duration(intpow(c.syncOpts.SyncMultiplier, int64(i)))) } go c.syncManager() return nil } // Options returns the sync's store options func (c *syncStore) Options() store.Options { return c.storeOpts } // String returns a printable string describing the sync func (c *syncStore) String() string { backends := make([]string, len(c.syncOpts.Stores)) for i, s := range c.syncOpts.Stores { backends[i] = s.String() } return fmt.Sprintf("sync %v", backends) } func (c *syncStore) List(ctx context.Context, opts ...store.ListOption) ([]string, error) { return c.syncOpts.Stores[0].List(ctx, opts...) } func (c *syncStore) Read(ctx context.Context, key string, opts ...store.ReadOption) ([]*store.Record, error) { return c.syncOpts.Stores[0].Read(ctx, key, opts...) } func (c *syncStore) Write(ctx context.Context, r *store.Record, opts ...store.WriteOption) error { return c.syncOpts.Stores[0].Write(ctx, r, opts...) } // Delete removes a key from the sync func (c *syncStore) Delete(ctx context.Context, key string, opts ...store.DeleteOption) error { return c.syncOpts.Stores[0].Delete(ctx, key, opts...) } func (c *syncStore) Sync() error { return nil } type internalRecord struct { key string value []byte expiresAt time.Time }