store: refactor interface #11

Merged
vtolstov merged 1 commits from issue-40 into master 2020-12-10 22:08:57 +03:00
8 changed files with 73 additions and 62 deletions

View File

@ -4,7 +4,6 @@ import (
"bytes" "bytes"
"encoding/gob" "encoding/gob"
"errors" "errors"
"fmt"
"path" "path"
"strings" "strings"
"time" "time"
@ -48,25 +47,19 @@ func (s *storage) Store(key string, value []byte) error {
if err := e.Encode(f); err != nil { if err := e.Encode(f); err != nil {
return err return err
} }
r := &store.Record{ return s.store.Write(s.store.Options().Context, key, buf.Bytes())
Key: key,
Value: buf.Bytes(),
}
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(s.store.Options().Context, key) var val []byte
err := s.store.Read(s.store.Options().Context, key, val)
if err != nil { if err != nil {
return nil, err return nil, err
} }
if len(records) != 1 { b := bytes.NewBuffer(val)
return nil, fmt.Errorf("ACME Storage: multiple records matched key %s", key)
}
b := bytes.NewBuffer(records[0].Value)
d := gob.NewDecoder(b) d := gob.NewDecoder(b)
var f File var f File
err = d.Decode(&f) err = d.Decode(&f)
@ -81,7 +74,7 @@ func (s *storage) Delete(key string) error {
} }
func (s *storage) Exists(key string) bool { func (s *storage) Exists(key string) bool {
if _, err := s.store.Read(s.store.Options().Context, key); err != nil { if err := s.store.Read(s.store.Options().Context, key, nil); err != nil {
return false return false
} }
return true return true
@ -116,14 +109,12 @@ 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(s.store.Options().Context, key) var val []byte
err := s.store.Read(s.store.Options().Context, key, val)
if err != nil { if err != nil {
return certmagic.KeyInfo{}, err return certmagic.KeyInfo{}, err
} }
if len(records) != 1 { b := bytes.NewBuffer(val)
return certmagic.KeyInfo{}, fmt.Errorf("ACME Storage: multiple records matched key %s", key)
}
b := bytes.NewBuffer(records[0].Value)
d := gob.NewDecoder(b) d := gob.NewDecoder(b)
var f File var f File
err = d.Decode(&f) err = d.Decode(&f)

View File

@ -29,12 +29,17 @@ func (n *noopStore) String() string {
} }
// Read reads store value by key // Read reads store value by key
func (n *noopStore) Read(ctx context.Context, key string, opts ...ReadOption) ([]*Record, error) { func (n *noopStore) Exists(ctx context.Context, key string) error {
return []*Record{}, nil return ErrNotFound
}
// Read reads store value by key
func (n *noopStore) Read(ctx context.Context, key string, val interface{}, opts ...ReadOption) error {
return ErrNotFound
} }
// Write writes store record // Write writes store record
func (n *noopStore) Write(ctx context.Context, r *Record, opts ...WriteOption) error { func (n *noopStore) Write(ctx context.Context, key string, val interface{}, opts ...WriteOption) error {
return nil return nil
} }

View File

@ -4,7 +4,9 @@ import (
"context" "context"
"time" "time"
"github.com/unistack-org/micro/v3/codec"
"github.com/unistack-org/micro/v3/logger" "github.com/unistack-org/micro/v3/logger"
"github.com/unistack-org/micro/v3/metadata"
) )
// Options contains configuration for the Store // Options contains configuration for the Store
@ -17,6 +19,8 @@ type Options struct {
Database string Database string
// Table is analag for a table in database backends or a key prefix in KV backends // Table is analag for a table in database backends or a key prefix in KV backends
Table string Table string
// Codec that used for marshal/unmarshal value
Codec codec.Codec
// Logger the logger // Logger the logger
Logger logger.Logger Logger logger.Logger
// Context should contain all implementation specific options, using context.WithValue. // Context should contain all implementation specific options, using context.WithValue.
@ -45,6 +49,13 @@ func Context(ctx context.Context) Option {
} }
} }
// Codec sets the codec
func Codec(c codec.Codec) Option {
return func(o *Options) {
o.Codec = c
}
}
// Logger sets the logger // Logger sets the logger
func Logger(l logger.Logger) Option { func Logger(l logger.Logger) Option {
return func(o *Options) { return func(o *Options) {
@ -130,11 +141,13 @@ func ReadOffset(o uint) ReadOption {
// WriteOptions configures an individual Write operation // WriteOptions configures an individual Write operation
// If Expiry and TTL are set TTL takes precedence // If Expiry and TTL are set TTL takes precedence
type WriteOptions struct { type WriteOptions struct {
Database, Table string Database string
Table string
// Expiry is the time the record expires // Expiry is the time the record expires
Expiry time.Time Expiry time.Time
// TTL is the time until the record expires // TTL is the time until the record expires
TTL time.Duration TTL time.Duration
Metadata metadata.Metadata
} }
// WriteOption sets values in WriteOptions // WriteOption sets values in WriteOptions
@ -162,6 +175,13 @@ func WriteTTL(d time.Duration) WriteOption {
} }
} }
// WriteMetadata add metadata.Metadata
func WriteMetadata(md metadata.Metadata) WriteOption {
return func(w *WriteOptions) {
w.Metadata = metadata.Copy(md)
}
}
// DeleteOptions configures an individual Delete operation // DeleteOptions configures an individual Delete operation
type DeleteOptions struct { type DeleteOptions struct {
Database, Table string Database, Table string

View File

@ -5,7 +5,8 @@ package store
import ( import (
"context" "context"
"errors" "errors"
"time"
"github.com/unistack-org/micro/v3/metadata"
) )
var ( var (
@ -23,10 +24,12 @@ type Store interface {
Connect(ctx context.Context) error Connect(ctx context.Context) error
// Options allows you to view the current options. // Options allows you to view the current options.
Options() Options Options() Options
// Read takes a single key name and optional ReadOptions. It returns matching []*Record or an error. // Exists check that key exists in store
Read(ctx context.Context, key string, opts ...ReadOption) ([]*Record, error) Exists(ctx context.Context, key string) error
// Write() writes a record to the store, and returns an error if the record was not written. // Read reads a single key name to provided value with optional ReadOptions
Write(ctx context.Context, r *Record, opts ...WriteOption) error Read(ctx context.Context, key string, val interface{}, opts ...ReadOption) error
// Write writes a value to key name to the store with optional WriteOption
Write(ctx context.Context, key string, val interface{}, opts ...WriteOption) error
// Delete removes the record with the corresponding key from the store. // Delete removes the record with the corresponding key from the store.
Delete(ctx context.Context, key string, opts ...DeleteOption) error Delete(ctx context.Context, key string, opts ...DeleteOption) error
// List returns any keys that match, or an empty list with no error if none matched. // List returns any keys that match, or an empty list with no error if none matched.
@ -37,14 +40,11 @@ type Store interface {
String() string String() string
} }
// Record is an item stored or retrieved from a Store // Value is an item stored or retrieved from a Store
type Record struct { // may be used in store implementations to provide metadata
// The key to store the record type Value struct {
Key string `json:"key"` // Data holds underline struct
// The value within the record Data interface{} `json:"data"`
Value []byte `json:"value"` // Metadata associated with data for indexing
// Any associated metadata for indexing Metadata metadata.Metadata `json:"metadata"`
Metadata map[string]interface{} `json:"metadata"`
// Time to expire a record: TODO: change to timestamp
Expiry time.Duration `json:"expiry,omitempty"`
} }

View File

@ -24,14 +24,14 @@ func (s *Scope) Options() store.Options {
return o return o
} }
func (s *Scope) Read(ctx context.Context, key string, opts ...store.ReadOption) ([]*store.Record, error) { func (s *Scope) Read(ctx context.Context, key string, val interface{}, opts ...store.ReadOption) error {
key = fmt.Sprintf("%v/%v", s.prefix, key) key = fmt.Sprintf("%v/%v", s.prefix, key)
return s.Store.Read(ctx, key, opts...) return s.Store.Read(ctx, key, val, opts...)
} }
func (s *Scope) Write(ctx context.Context, r *store.Record, opts ...store.WriteOption) error { func (s *Scope) Write(ctx context.Context, key string, val interface{}, opts ...store.WriteOption) error {
r.Key = fmt.Sprintf("%v/%v", s.prefix, r.Key) key = fmt.Sprintf("%v/%v", s.prefix, key)
return s.Store.Write(ctx, r, opts...) return s.Store.Write(ctx, key, val, opts...)
} }
func (s *Scope) Delete(ctx context.Context, key string, opts ...store.DeleteOption) error { func (s *Scope) Delete(ctx context.Context, key string, opts ...store.DeleteOption) error {

View File

@ -41,16 +41,12 @@ func (c *syncStore) processQueue(index int) {
if !ir.expiresAt.IsZero() && time.Now().After(ir.expiresAt) { if !ir.expiresAt.IsZero() && time.Now().After(ir.expiresAt) {
continue continue
} }
nr := &store.Record{ var opts []store.WriteOption
Key: ir.key,
}
nr.Value = make([]byte, len(ir.value))
copy(nr.Value, ir.value)
if !ir.expiresAt.IsZero() { if !ir.expiresAt.IsZero() {
nr.Expiry = time.Until(ir.expiresAt) opts = append(opts, store.WriteTTL(ir.expiresAt.Sub(time.Now())))
} }
// 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(c.storeOpts.Context, nr); err != nil { if err := c.syncOpts.Stores[index+1].Write(c.storeOpts.Context, ir.key, ir.value, opts...); 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

View File

@ -96,12 +96,16 @@ func (c *syncStore) List(ctx context.Context, opts ...store.ListOption) ([]strin
return c.syncOpts.Stores[0].List(ctx, opts...) return c.syncOpts.Stores[0].List(ctx, opts...)
} }
func (c *syncStore) Read(ctx context.Context, key string, opts ...store.ReadOption) ([]*store.Record, error) { func (c *syncStore) Exists(ctx context.Context, key string) error {
return c.syncOpts.Stores[0].Read(ctx, key, opts...) return c.syncOpts.Stores[0].Exists(ctx, key)
} }
func (c *syncStore) Write(ctx context.Context, r *store.Record, opts ...store.WriteOption) error { func (c *syncStore) Read(ctx context.Context, key string, val interface{}, opts ...store.ReadOption) error {
return c.syncOpts.Stores[0].Write(ctx, r, opts...) return c.syncOpts.Stores[0].Read(ctx, key, val, opts...)
}
func (c *syncStore) Write(ctx context.Context, key string, val interface{}, opts ...store.WriteOption) error {
return c.syncOpts.Stores[0].Write(ctx, key, val, opts...)
} }
// Delete removes a key from the sync // Delete removes a key from the sync

View File

@ -47,11 +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(context.Background(), &store.Record{ err = b.store.Write(context.Background(), fmt.Sprintf("%v%v", StorePrefix, key), bytes, store.WriteTTL(options.Expiry))
Key: fmt.Sprintf("%v%v", StorePrefix, key),
Value: bytes,
Expiry: options.Expiry,
})
if err != nil { if err != nil {
return nil, err return nil, err
} }
@ -67,17 +63,16 @@ 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(context.Background(), StorePrefix+t) var val []byte
err := b.store.Read(context.Background(), StorePrefix+t, val)
if err == store.ErrNotFound { if err == store.ErrNotFound {
return nil, token.ErrInvalidToken return nil, token.ErrInvalidToken
} else if err != nil { } else if err != nil {
return nil, err return nil, err
} }
bytes := recs[0].Value
// unmarshal the bytes // unmarshal the bytes
var acc *auth.Account var acc *auth.Account
if err := json.Unmarshal(bytes, &acc); err != nil { if err := json.Unmarshal(val, &acc); err != nil {
return nil, err return nil, err
} }