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 93 additions and 19 deletions

View File

@ -4,6 +4,7 @@ import (
"context" "context"
"sort" "sort"
"strings" "strings"
"sync/atomic"
"time" "time"
cache "github.com/patrickmn/go-cache" cache "github.com/patrickmn/go-cache"
@ -20,8 +21,11 @@ func NewStore(opts ...store.Option) store.Store {
} }
func (m *memoryStore) Connect(ctx context.Context) error { func (m *memoryStore) Connect(ctx context.Context) error {
if m.opts.LazyConnect {
return nil return nil
} }
return m.connect(ctx)
}
func (m *memoryStore) Disconnect(ctx context.Context) error { func (m *memoryStore) Disconnect(ctx context.Context) error {
m.store.Flush() m.store.Flush()
@ -36,6 +40,7 @@ type memoryStore struct {
funcDelete store.FuncDelete funcDelete store.FuncDelete
store *cache.Cache store *cache.Cache
opts store.Options opts store.Options
isConnected atomic.Int32
} }
func (m *memoryStore) key(prefix, key string) string { func (m *memoryStore) key(prefix, key string) string {
@ -145,6 +150,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 +167,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 +184,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 +213,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 +236,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 +274,8 @@ 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.isConnected.CompareAndSwap(0, 1)
return nil
}

View File

@ -2,6 +2,7 @@ package store
import ( import (
"context" "context"
"sync/atomic"
"go.unistack.org/micro/v3/options" "go.unistack.org/micro/v3/options"
) )
@ -15,6 +16,7 @@ type noopStore struct {
funcList FuncList funcList FuncList
funcDelete FuncDelete funcDelete FuncDelete
opts Options opts Options
isConnected atomic.Int32
} }
func NewStore(opts ...Option) *noopStore { func NewStore(opts ...Option) *noopStore {
@ -52,13 +54,11 @@ 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 ctx.Err()
default:
}
return nil return nil
} }
return n.connect(ctx)
}
func (n *noopStore) Disconnect(ctx context.Context) error { func (n *noopStore) Disconnect(ctx context.Context) error {
select { select {
@ -70,6 +70,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 +88,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 +106,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 +124,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 +142,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 +170,15 @@ 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 {
if n.isConnected.CompareAndSwap(0, 1) {
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