unistack-org/micro
3
0
Fork 3
Code Issues 16 Pull Requests 2 Actions Packages Projects Releases 239 Wiki Activity

fix store context issues

Browse Source 
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
This commit is contained in:
Василий Толстов 2020-09-17 15:41:49 +03:00
parent 6021edc855
commit 7e3fac8937
6 changed files with 33 additions and 34 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
api/server/acme/certmagic/storage.go
View File

@ -52,14 +52,14 @@ func (s *storage) Store(key string, value []byte) error {
Key: key, Key: key,
Value: buf.Bytes(), Value: buf.Bytes(),
} }
return s.store.Write(r) return s.store.Write(s.store.Options().Context, r)
} }
func (s *storage) Load(key string) ([]byte, error) { func (s *storage) Load(key string) ([]byte, error) {
if !s.Exists(key) { if !s.Exists(key) {
return nil, certmagic.ErrNotExist(errors.New(key + " doesn't exist")) return nil, certmagic.ErrNotExist(errors.New(key + " doesn't exist"))
} }
records, err := s.store.Read(key) records, err := s.store.Read(s.store.Options().Context, key)
if err != nil { if err != nil {
return nil, err return nil, err
} }
@ -77,18 +77,18 @@ func (s *storage) Load(key string) ([]byte, error) {
} }
func (s *storage) Delete(key string) error { func (s *storage) Delete(key string) error {
return s.store.Delete(key) return s.store.Delete(s.store.Options().Context, key)
} }
func (s *storage) Exists(key string) bool { func (s *storage) Exists(key string) bool {
if _, err := s.store.Read(key); err != nil { if _, err := s.store.Read(s.store.Options().Context, key); err != nil {
return false return false
} }
return true return true
} }
func (s *storage) List(prefix string, recursive bool) ([]string, error) { func (s *storage) List(prefix string, recursive bool) ([]string, error) {
keys, err := s.store.List() keys, err := s.store.List(s.store.Options().Context)
if err != nil { if err != nil {
return nil, err return nil, err
} }
@ -116,7 +116,7 @@ func (s *storage) List(prefix string, recursive bool) ([]string, error) {
} }
func (s *storage) Stat(key string) (certmagic.KeyInfo, error) { func (s *storage) Stat(key string) (certmagic.KeyInfo, error) {
records, err := s.store.Read(key) records, err := s.store.Read(s.store.Options().Context, key)
if err != nil { if err != nil {
return certmagic.KeyInfo{}, err return certmagic.KeyInfo{}, err
} }

2
service.go
View File

@ -86,7 +86,7 @@ func (s *service) Init(opts ...Option) {
} }
if s.opts.Store != nil { if s.opts.Store != nil {
if err := s.opts.Store.Init(); err != nil { if err := s.opts.Store.Init(s.opts.Context); err != nil {
logger.Fatalf("[cmd] init failed: %v", err) logger.Fatalf("[cmd] init failed: %v", err)
} }
} }

17
util/scope/scope.go
View File

@ -1,6 +1,7 @@
package scope package scope
import ( import (
"context"
"fmt" "fmt"
"github.com/unistack-org/micro/v3/store" "github.com/unistack-org/micro/v3/store"
@ -23,22 +24,22 @@ func (s *Scope) Options() store.Options {
return o return o
} }
func (s *Scope) Read(key string, opts ...store.ReadOption) ([]*store.Record, error) { func (s *Scope) Read(ctx context.Context, key string, opts ...store.ReadOption) ([]*store.Record, error) {
key = fmt.Sprintf("%v/%v", s.prefix, key) key = fmt.Sprintf("%v/%v", s.prefix, key)
return s.Store.Read(key, opts...) return s.Store.Read(ctx, key, opts...)
} }
func (s *Scope) Write(r *store.Record, opts ...store.WriteOption) error { func (s *Scope) Write(ctx context.Context, r *store.Record, opts ...store.WriteOption) error {
r.Key = fmt.Sprintf("%v/%v", s.prefix, r.Key) r.Key = fmt.Sprintf("%v/%v", s.prefix, r.Key)
return s.Store.Write(r, opts...) return s.Store.Write(ctx, r, opts...)
} }
func (s *Scope) Delete(key string, opts ...store.DeleteOption) error { func (s *Scope) Delete(ctx context.Context, key string, opts ...store.DeleteOption) error {
key = fmt.Sprintf("%v/%v", s.prefix, key) key = fmt.Sprintf("%v/%v", s.prefix, key)
return s.Store.Delete(key, opts...) return s.Store.Delete(ctx, key, opts...)
} }
func (s *Scope) List(opts ...store.ListOption) ([]string, error) { func (s *Scope) List(ctx context.Context, opts ...store.ListOption) ([]string, error) {
var lops store.ListOptions var lops store.ListOptions
for _, o := range opts { for _, o := range opts {
o(&lops) o(&lops)
@ -47,5 +48,5 @@ func (s *Scope) List(opts ...store.ListOption) ([]string, error) {
key := fmt.Sprintf("%v/%v", s.prefix, lops.Prefix) key := fmt.Sprintf("%v/%v", s.prefix, lops.Prefix)
opts = append(opts, store.ListPrefix(key)) opts = append(opts, store.ListPrefix(key))
return s.Store.List(opts...) return s.Store.List(ctx, opts...)
} }

2
util/sync/manager.go
View File

@ -50,7 +50,7 @@ func (c *syncStore) processQueue(index int) {
nr.Expiry = time.Until(ir.expiresAt) nr.Expiry = time.Until(ir.expiresAt)
} }
// Todo = internal queue also has to hold the corresponding store.WriteOptions // Todo = internal queue also has to hold the corresponding store.WriteOptions
if err := c.syncOpts.Stores[index+1].Write(nr); err != nil { if err := c.syncOpts.Stores[index+1].Write(c.storeOpts.Context, nr); err != nil {
// some error, so queue for retry and bail // some error, so queue for retry and bail
q.PushBack(ir) q.PushBack(ir)
return return

29
util/sync/sync.go
View File

@ -2,7 +2,7 @@
package sync package sync
import ( import (
"errors" "context"
"fmt" "fmt"
"sync" "sync"
"time" "time"
@ -42,23 +42,20 @@ func NewSync(opts ...Option) Sync {
return c return c
} }
func (c *syncStore) Close() error { func (c *syncStore) Close(ctx context.Context) error {
return nil return nil
} }
// Init initialises the storeOptions // Init initialises the storeOptions
func (c *syncStore) Init(opts ...store.Option) error { func (c *syncStore) Init(ctx context.Context, opts ...store.Option) error {
for _, o := range opts { for _, o := range opts {
o(&c.storeOpts) o(&c.storeOpts)
} }
if len(c.syncOpts.Stores) == 0 { if len(c.syncOpts.Stores) == 0 {
return errors.New("the sync has no stores") return fmt.Errorf("the sync has no stores")
}
if c.storeOpts.Context == nil {
return errors.New("please provide a context to the sync. Cancelling the context signals that the sync is being disposed and syncs the sync")
} }
for _, s := range c.syncOpts.Stores { for _, s := range c.syncOpts.Stores {
if err := s.Init(); err != nil { if err := s.Init(ctx); err != nil {
return fmt.Errorf("Store %s failed to Init(): %w", s.String(), err) return fmt.Errorf("Store %s failed to Init(): %w", s.String(), err)
} }
} }
@ -87,21 +84,21 @@ func (c *syncStore) String() string {
return fmt.Sprintf("sync %v", backends) return fmt.Sprintf("sync %v", backends)
} }
func (c *syncStore) List(opts ...store.ListOption) ([]string, error) { func (c *syncStore) List(ctx context.Context, opts ...store.ListOption) ([]string, error) {
return c.syncOpts.Stores[0].List(opts...) return c.syncOpts.Stores[0].List(ctx, opts...)
} }
func (c *syncStore) Read(key string, opts ...store.ReadOption) ([]*store.Record, error) { func (c *syncStore) Read(ctx context.Context, key string, opts ...store.ReadOption) ([]*store.Record, error) {
return c.syncOpts.Stores[0].Read(key, opts...) return c.syncOpts.Stores[0].Read(ctx, key, opts...)
} }
func (c *syncStore) Write(r *store.Record, opts ...store.WriteOption) error { func (c *syncStore) Write(ctx context.Context, r *store.Record, opts ...store.WriteOption) error {
return c.syncOpts.Stores[0].Write(r, opts...) return c.syncOpts.Stores[0].Write(ctx, r, opts...)
} }
// Delete removes a key from the sync // Delete removes a key from the sync
func (c *syncStore) Delete(key string, opts ...store.DeleteOption) error { func (c *syncStore) Delete(ctx context.Context, key string, opts ...store.DeleteOption) error {
return c.syncOpts.Stores[0].Delete(key, opts...) return c.syncOpts.Stores[0].Delete(ctx, key, opts...)
} }
func (c *syncStore) Sync() error { func (c *syncStore) Sync() error {

5
util/token/basic/basic.go
View File

@ -1,6 +1,7 @@
package basic package basic
import ( import (
"context"
"encoding/json" "encoding/json"
"fmt" "fmt"
"time" "time"
@ -46,7 +47,7 @@ func (b *Basic) Generate(acc *auth.Account, opts ...token.GenerateOption) (*toke
// write to the store // write to the store
key := uuid.New().String() key := uuid.New().String()
err = b.store.Write(&store.Record{ err = b.store.Write(context.Background(), &store.Record{
Key: fmt.Sprintf("%v%v", StorePrefix, key), Key: fmt.Sprintf("%v%v", StorePrefix, key),
Value: bytes, Value: bytes,
Expiry: options.Expiry, Expiry: options.Expiry,
@ -66,7 +67,7 @@ func (b *Basic) Generate(acc *auth.Account, opts ...token.GenerateOption) (*toke
// Inspect a token // Inspect a token
func (b *Basic) Inspect(t string) (*auth.Account, error) { func (b *Basic) Inspect(t string) (*auth.Account, error) {
// lookup the token in the store // lookup the token in the store
recs, err := b.store.Read(StorePrefix + t) recs, err := b.store.Read(context.Background(), StorePrefix+t)
if err == store.ErrNotFound { if err == store.ErrNotFound {
return nil, token.ErrInvalidToken return nil, token.ErrInvalidToken
} else if err != nil { } else if err != nil {