unistack-org/micro-store-redis
4
0
Fork 2
Code Issues Pull Requests Actions 1 Packages Projects Releases 38 Wiki Activity

add store.Watcher interface #123

Merged
vtolstov merged 2 commits from events into v3 2024-12-01 19:58:11 +03:00
Conversation 0 Commits 2 Files Changed 4 +140 -27
4 changed files with 140 additions and 27 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

42
event.go
View File

@ -4,19 +4,20 @@ import (
"context"
"errors"
"net"
"sync/atomic"
"time"
goredis "github.com/redis/go-redis/v9"
"go.unistack.org/micro/v3/store"
)
type eventHook struct {
connected *atomic.Bool
s *Store
}
var _ goredis.Hook = (*eventHook)(nil)
func newEventHook(connected *atomic.Bool) *eventHook {
return &eventHook{connected: connected}
func newEventHook(s *Store) *eventHook {
return &eventHook{s: s}
}
func (h *eventHook) DialHook(hook goredis.DialHook) goredis.DialHook {
@ -24,11 +25,16 @@ func (h *eventHook) DialHook(hook goredis.DialHook) goredis.DialHook {
conn, err := hook(ctx, network, addr)
if err != nil {
if !isRedisError(err) {
h.connected.Store(false)
if h.s.connected.CompareAndSwap(1, 0) {
h.s.sendEvent(&event{ts: time.Now(), err: err, t: store.EventTypeDisconnect})
}
h.connected.Store(true)
} else {
h.connected.Store(true)
h.s.connected.Store(1)
}
} else {
if h.s.connected.CompareAndSwap(0, 1) {
h.s.sendEvent(&event{ts: time.Now(), err: err, t: store.EventTypeConnect})
}
}
return conn, err
}
@ -39,11 +45,16 @@ func (h *eventHook) ProcessHook(hook goredis.ProcessHook) goredis.ProcessHook {
err := hook(ctx, cmd)
if err != nil {
if !isRedisError(err) {
h.connected.Store(false)
if h.s.connected.CompareAndSwap(1, 0) {
h.s.sendEvent(&event{ts: time.Now(), err: err, t: store.EventTypeDisconnect})
}
h.connected.Store(true)
} else {
h.connected.Store(true)
h.s.connected.Store(1)
}
} else {
if h.s.connected.CompareAndSwap(0, 1) {
h.s.sendEvent(&event{ts: time.Now(), err: err, t: store.EventTypeConnect})
}
}
return err
}
@ -54,11 +65,16 @@ func (h *eventHook) ProcessPipelineHook(hook goredis.ProcessPipelineHook) goredi
err := hook(ctx, cmds)
if err != nil {
if !isRedisError(err) {
h.connected.Store(false)
if h.s.connected.CompareAndSwap(1, 0) {
h.s.sendEvent(&event{ts: time.Now(), err: err, t: store.EventTypeDisconnect})
}
h.connected.Store(true)
} else {
h.connected.Store(true)
h.s.connected.Store(1)
}
} else {
if h.s.connected.CompareAndSwap(0, 1) {
h.s.sendEvent(&event{ts: time.Now(), err: err, t: store.EventTypeConnect})
}
}
return err
}

2
go.mod
View File

@ -7,7 +7,7 @@ toolchain go1.22.4
require (
github.com/redis/go-redis/extra/rediscmd/v9 v9.7.0
github.com/redis/go-redis/v9 v9.7.0
go.unistack.org/micro/v3 v3.10.106
go.unistack.org/micro/v3 v3.10.108
)
require (

4
go.sum
View File

@ -14,7 +14,7 @@ github.com/redis/go-redis/v9 v9.7.0 h1:HhLSs+B6O021gwzl+locl0zEDnyNkxMtf/Z3NNBMa
github.com/redis/go-redis/v9 v9.7.0/go.mod h1:f6zhXITC7JUJIlPEiBOTXxJgPLdZcA93GewI7inzyWw=
go.unistack.org/micro-proto/v3 v3.4.1 h1:UTjLSRz2YZuaHk9iSlVqqsA50JQNAEK2ZFboGqtEa9Q=
go.unistack.org/micro-proto/v3 v3.4.1/go.mod h1:okx/cnOhzuCX0ggl/vToatbCupi0O44diiiLLsZ93Zo=
go.unistack.org/micro/v3 v3.10.106 h1:ya4+n58l4PImtrIKrJi1GgkUuJ1gmzLYa9WKYI1JFLs=
go.unistack.org/micro/v3 v3.10.106/go.mod h1:YzMldzHN9Ei+zy5t/Psu7RUWDZwUfrNYiStSQtTz90g=
go.unistack.org/micro/v3 v3.10.108 h1:3L7SkilMVLtH8y3pQIPtr3jjQYrf0AMv1oAkoL3nFkE=
go.unistack.org/micro/v3 v3.10.108/go.mod h1:YzMldzHN9Ei+zy5t/Psu7RUWDZwUfrNYiStSQtTz90g=
google.golang.org/protobuf v1.35.2 h1:8Ar7bF+apOIoThw1EdZl0p1oWvMqTHmpA2fRTyZO8io=
google.golang.org/protobuf v1.35.2/go.mod h1:9fA7Ob0pmnwhb644+1+CVWFRbNajQ6iRojtC/QF5bRE=

117
redis.go
View File

@ -5,18 +5,21 @@ import (
"errors"
"reflect"
"strings"
"sync"
"sync/atomic"
"time"
goredis "github.com/redis/go-redis/v9"
"go.unistack.org/micro/v3/semconv"
"go.unistack.org/micro/v3/store"
"go.unistack.org/micro/v3/util/id"
pool "go.unistack.org/micro/v3/util/xpool"
)
var (
DefaultPathSeparator = "/"
_ store.Store = (*Store)(nil)
_ store.Event = (*event)(nil)
sendEventTime = 10 * time.Millisecond
DefaultUniversalOptions = &goredis.UniversalOptions{
Username: "",
Password: "", // no password set
@ -58,12 +61,14 @@ var (
type Store struct {
cli goredis.UniversalClient
pool *pool.StringsPool
connected *atomic.Bool
connected *atomic.Uint32
opts store.Options
watchers map[string]*watcher
mu sync.RWMutex
}
func (r *Store) Connect(ctx context.Context) error {
if r.connected.Load() {
if r.connected.Load() == 1 {
return nil
}
if r.cli == nil {
@ -76,7 +81,7 @@ func (r *Store) Connect(ctx context.Context) error {
setSpanError(ctx, err)
return err
}
r.connected.Store(true)
r.connected.Store(1)
return nil
}
@ -112,7 +117,7 @@ func (r *Store) ClusterClient() *goredis.ClusterClient {
}
func (r *Store) Disconnect(ctx context.Context) error {
if !r.connected.Load() {
if r.connected.Load() == 0 {
return nil
}
@ -122,7 +127,7 @@ func (r *Store) Disconnect(ctx context.Context) error {
}
}
r.connected.Store(false)
r.connected.Store(1)
return nil
}
@ -650,10 +655,11 @@ func (r *Store) String() string {
}
func NewStore(opts ...store.Option) *Store {
b := atomic.Bool{}
b := atomic.Uint32{}
return &Store{
opts: store.NewOptions(opts...),
connected: &b,
watchers: make(map[string]*watcher),
}
}
@ -745,7 +751,7 @@ func (r *Store) configure() error {
r.cli = goredis.NewUniversalClient(universalOptions)
setTracing(r.cli, r.opts.Tracer)
r.cli.AddHook(newEventHook(r.connected))
r.cli.AddHook(newEventHook(r))
r.pool = pool.NewStringsPool(50)
@ -760,8 +766,99 @@ func (r *Store) getKey(b *strings.Builder, mainNamespace string, opNamespace str
}
if opNamespace != "" {
b.WriteString(opNamespace)
b.WriteString(DefaultPathSeparator)
b.WriteString(r.opts.Separator)
}
b.WriteString(key)
return b.String()
}
func (r *Store) Watch(ctx context.Context, opts ...store.WatchOption) (store.Watcher, error) {
id, err := id.New()
if err != nil {
return nil, err
}
wo, err := store.NewWatchOptions(opts...)
if err != nil {
return nil, err
}
// construct the watcher
w := &watcher{
exit: make(chan bool),
ch: make(chan store.Event),
id: id,
opts: wo,
}
r.mu.Lock()
r.watchers[w.id] = w
r.mu.Unlock()
return w, nil
}
func (r *Store) sendEvent(e store.Event) {
r.mu.RLock()
watchers := make([]*watcher, 0, len(r.watchers))
for _, w := range r.watchers {
watchers = append(watchers, w)
}
r.mu.RUnlock()
for _, w := range watchers {
select {
case <-w.exit:
r.mu.Lock()
delete(r.watchers, w.id)
r.mu.Unlock()
default:
select {
case w.ch <- e:
case <-time.After(sendEventTime):
}
}
}
}
type watcher struct {
ch chan store.Event
exit chan bool
opts store.WatchOptions
id string
}
func (w *watcher) Next() (store.Event, error) {
for {
select {
case e := <-w.ch:
return e, nil
case <-w.exit:
return nil, store.ErrWatcherStopped
}
}
}
func (w *watcher) Stop() {
select {
case <-w.exit:
return
default:
close(w.exit)
}
}
type event struct {
ts time.Time
t store.EventType
err error
}
func (e *event) Error() error {
return e.err
}
func (e *event) Timestamp() time.Time {
return e.ts
}
func (e *event) Type() store.EventType {
return e.t
}