add using lazy connect #361

Merged
vtolstov merged 3 commits from devstigneev/micro:store_lazy_connect into v3 2024-11-26 12:18:19 +03:00
3 changed files with 100 additions and 19 deletions
Showing only changes of commit 775429d4d6 - Show all commits

View File

@ -4,6 +4,7 @@ import (
"context" "context"
"sort" "sort"
"strings" "strings"
"sync"
"time" "time"
cache "github.com/patrickmn/go-cache" cache "github.com/patrickmn/go-cache"
@ -20,7 +21,10 @@ func NewStore(opts ...store.Option) store.Store {
} }
func (m *memoryStore) Connect(ctx context.Context) error { func (m *memoryStore) Connect(ctx context.Context) error {
return nil if m.opts.LazyConnect {
return nil
}
return m.connect(ctx)
} }
func (m *memoryStore) Disconnect(ctx context.Context) error { func (m *memoryStore) Disconnect(ctx context.Context) error {
@ -29,13 +33,15 @@ func (m *memoryStore) Disconnect(ctx context.Context) error {
} }
type memoryStore struct { type memoryStore struct {
funcRead store.FuncRead funcRead store.FuncRead
funcWrite store.FuncWrite funcWrite store.FuncWrite
funcExists store.FuncExists funcExists store.FuncExists
funcList store.FuncList funcList store.FuncList
funcDelete store.FuncDelete funcDelete store.FuncDelete
store *cache.Cache store *cache.Cache
opts store.Options opts store.Options
isConnected bool
mutex sync.Mutex
} }
func (m *memoryStore) key(prefix, key string) string { func (m *memoryStore) key(prefix, key string) string {
@ -145,6 +151,11 @@ func (m *memoryStore) Name() string {
} }
func (m *memoryStore) Exists(ctx context.Context, key string, opts ...store.ExistsOption) error { func (m *memoryStore) Exists(ctx context.Context, key string, opts ...store.ExistsOption) error {
if m.opts.LazyConnect {
if err := m.connect(ctx); err != nil {
return err
}
}
return m.funcExists(ctx, key, opts...) return m.funcExists(ctx, key, opts...)
} }
@ -157,6 +168,11 @@ func (m *memoryStore) fnExists(ctx context.Context, key string, opts ...store.Ex
} }
func (m *memoryStore) Read(ctx context.Context, key string, val interface{}, opts ...store.ReadOption) error { func (m *memoryStore) Read(ctx context.Context, key string, val interface{}, opts ...store.ReadOption) error {
if m.opts.LazyConnect {
if err := m.connect(ctx); err != nil {
return err
}
}
return m.funcRead(ctx, key, val, opts...) return m.funcRead(ctx, key, val, opts...)
} }
@ -169,6 +185,11 @@ func (m *memoryStore) fnRead(ctx context.Context, key string, val interface{}, o
} }
func (m *memoryStore) Write(ctx context.Context, key string, val interface{}, opts ...store.WriteOption) error { func (m *memoryStore) Write(ctx context.Context, key string, val interface{}, opts ...store.WriteOption) error {
if m.opts.LazyConnect {
if err := m.connect(ctx); err != nil {
return err
}
}
return m.funcWrite(ctx, key, val, opts...) return m.funcWrite(ctx, key, val, opts...)
} }
@ -193,6 +214,11 @@ func (m *memoryStore) fnWrite(ctx context.Context, key string, val interface{},
} }
func (m *memoryStore) Delete(ctx context.Context, key string, opts ...store.DeleteOption) error { func (m *memoryStore) Delete(ctx context.Context, key string, opts ...store.DeleteOption) error {
if m.opts.LazyConnect {
if err := m.connect(ctx); err != nil {
return err
}
}
return m.funcDelete(ctx, key, opts...) return m.funcDelete(ctx, key, opts...)
} }
@ -211,6 +237,11 @@ func (m *memoryStore) Options() store.Options {
} }
func (m *memoryStore) List(ctx context.Context, opts ...store.ListOption) ([]string, error) { func (m *memoryStore) List(ctx context.Context, opts ...store.ListOption) ([]string, error) {
if m.opts.LazyConnect {
if err := m.connect(ctx); err != nil {
return nil, err
}
}
return m.funcList(ctx, opts...) return m.funcList(ctx, opts...)
} }
@ -244,3 +275,10 @@ func (m *memoryStore) fnList(ctx context.Context, opts ...store.ListOption) ([]s
return keys, nil return keys, nil
} }
func (m *memoryStore) connect(ctx context.Context) error {
m.mutex.Lock()
defer m.mutex.Unlock()
m.isConnected = true
return nil
}

View File

@ -2,6 +2,7 @@ package store
import ( import (
"context" "context"
"sync"
"go.unistack.org/micro/v3/options" "go.unistack.org/micro/v3/options"
) )
@ -9,12 +10,14 @@ import (
var _ Store = (*noopStore)(nil) var _ Store = (*noopStore)(nil)
type noopStore struct { type noopStore struct {
funcRead FuncRead funcRead FuncRead
funcWrite FuncWrite funcWrite FuncWrite
funcExists FuncExists funcExists FuncExists
funcList FuncList funcList FuncList
funcDelete FuncDelete funcDelete FuncDelete
opts Options opts Options
isConnected bool
mutex sync.Mutex
} }
func NewStore(opts ...Option) *noopStore { func NewStore(opts ...Option) *noopStore {
@ -52,12 +55,10 @@ func (n *noopStore) Init(opts ...Option) error {
} }
func (n *noopStore) Connect(ctx context.Context) error { func (n *noopStore) Connect(ctx context.Context) error {
select { if n.opts.LazyConnect {
case <-ctx.Done(): return nil
return ctx.Err()
default:
} }
return nil return n.connect(ctx)
} }
func (n *noopStore) Disconnect(ctx context.Context) error { func (n *noopStore) Disconnect(ctx context.Context) error {
@ -70,6 +71,11 @@ func (n *noopStore) Disconnect(ctx context.Context) error {
} }
func (n *noopStore) Read(ctx context.Context, key string, val interface{}, opts ...ReadOption) error { func (n *noopStore) Read(ctx context.Context, key string, val interface{}, opts ...ReadOption) error {
if n.opts.LazyConnect {
if err := n.connect(ctx); err != nil {
return err
}
}
return n.funcRead(ctx, key, val, opts...) return n.funcRead(ctx, key, val, opts...)
} }
@ -83,6 +89,11 @@ func (n *noopStore) fnRead(ctx context.Context, key string, val interface{}, opt
} }
func (n *noopStore) Delete(ctx context.Context, key string, opts ...DeleteOption) error { func (n *noopStore) Delete(ctx context.Context, key string, opts ...DeleteOption) error {
if n.opts.LazyConnect {
if err := n.connect(ctx); err != nil {
return err
}
}
return n.funcDelete(ctx, key, opts...) return n.funcDelete(ctx, key, opts...)
} }
@ -96,6 +107,11 @@ func (n *noopStore) fnDelete(ctx context.Context, key string, opts ...DeleteOpti
} }
func (n *noopStore) Exists(ctx context.Context, key string, opts ...ExistsOption) error { func (n *noopStore) Exists(ctx context.Context, key string, opts ...ExistsOption) error {
if n.opts.LazyConnect {
if err := n.connect(ctx); err != nil {
return err
}
}
return n.funcExists(ctx, key, opts...) return n.funcExists(ctx, key, opts...)
} }
@ -109,6 +125,11 @@ func (n *noopStore) fnExists(ctx context.Context, key string, opts ...ExistsOpti
} }
func (n *noopStore) Write(ctx context.Context, key string, val interface{}, opts ...WriteOption) error { func (n *noopStore) Write(ctx context.Context, key string, val interface{}, opts ...WriteOption) error {
if n.opts.LazyConnect {
if err := n.connect(ctx); err != nil {
return err
}
}
return n.funcWrite(ctx, key, val, opts...) return n.funcWrite(ctx, key, val, opts...)
} }
@ -122,6 +143,11 @@ func (n *noopStore) fnWrite(ctx context.Context, key string, val interface{}, op
} }
func (n *noopStore) List(ctx context.Context, opts ...ListOption) ([]string, error) { func (n *noopStore) List(ctx context.Context, opts ...ListOption) ([]string, error) {
if n.opts.LazyConnect {
if err := n.connect(ctx); err != nil {
return nil, err
}
}
return n.funcList(ctx, opts...) return n.funcList(ctx, opts...)
} }
@ -145,3 +171,18 @@ func (n *noopStore) String() string {
func (n *noopStore) Options() Options { func (n *noopStore) Options() Options {
return n.opts return n.opts
} }
func (n *noopStore) connect(ctx context.Context) error {
n.mutex.Lock()
defer n.mutex.Unlock()
if n.isConnected {
return nil
}
n.isConnected = true
select {
case <-ctx.Done():
return ctx.Err()
default:
}
return nil
}

View File

@ -41,6 +41,8 @@ type Options struct {
Timeout time.Duration Timeout time.Duration
// Hooks can be run before/after store Read/List/Write/Exists/Delete // Hooks can be run before/after store Read/List/Write/Exists/Delete
Hooks options.Hooks Hooks options.Hooks
// LazyConnect creates a connection when using store
LazyConnect bool
} }
// NewOptions creates options struct // NewOptions creates options struct