2024-11-27 01:57:39 +03:00
|
|
|
package redis
|
|
|
|
|
|
|
|
import (
|
|
|
|
"context"
|
|
|
|
"errors"
|
|
|
|
"net"
|
2024-12-01 19:58:11 +03:00
|
|
|
"time"
|
2024-11-27 01:57:39 +03:00
|
|
|
|
|
|
|
goredis "github.com/redis/go-redis/v9"
|
2024-12-01 19:58:11 +03:00
|
|
|
"go.unistack.org/micro/v3/store"
|
2024-11-27 01:57:39 +03:00
|
|
|
)
|
|
|
|
|
|
|
|
type eventHook struct {
|
2024-12-01 19:58:11 +03:00
|
|
|
s *Store
|
2024-11-27 01:57:39 +03:00
|
|
|
}
|
|
|
|
|
|
|
|
var _ goredis.Hook = (*eventHook)(nil)
|
|
|
|
|
2024-12-01 19:58:11 +03:00
|
|
|
func newEventHook(s *Store) *eventHook {
|
|
|
|
return &eventHook{s: s}
|
2024-11-27 01:57:39 +03:00
|
|
|
}
|
|
|
|
|
|
|
|
func (h *eventHook) DialHook(hook goredis.DialHook) goredis.DialHook {
|
|
|
|
return func(ctx context.Context, network, addr string) (net.Conn, error) {
|
|
|
|
conn, err := hook(ctx, network, addr)
|
2024-11-28 09:49:29 +03:00
|
|
|
if err != nil {
|
|
|
|
if !isRedisError(err) {
|
2024-12-01 19:58:11 +03:00
|
|
|
if h.s.connected.CompareAndSwap(1, 0) {
|
|
|
|
h.s.sendEvent(&event{ts: time.Now(), err: err, t: store.EventTypeDisconnect})
|
|
|
|
}
|
|
|
|
} else {
|
|
|
|
h.s.connected.Store(1)
|
2024-11-28 09:49:29 +03:00
|
|
|
}
|
|
|
|
} else {
|
2024-12-01 19:58:11 +03:00
|
|
|
if h.s.connected.CompareAndSwap(0, 1) {
|
|
|
|
h.s.sendEvent(&event{ts: time.Now(), err: err, t: store.EventTypeConnect})
|
|
|
|
}
|
2024-11-27 01:57:39 +03:00
|
|
|
}
|
|
|
|
return conn, err
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
func (h *eventHook) ProcessHook(hook goredis.ProcessHook) goredis.ProcessHook {
|
|
|
|
return func(ctx context.Context, cmd goredis.Cmder) error {
|
|
|
|
err := hook(ctx, cmd)
|
2024-11-28 09:53:03 +03:00
|
|
|
if err != nil {
|
|
|
|
if !isRedisError(err) {
|
2024-12-01 19:58:11 +03:00
|
|
|
if h.s.connected.CompareAndSwap(1, 0) {
|
|
|
|
h.s.sendEvent(&event{ts: time.Now(), err: err, t: store.EventTypeDisconnect})
|
|
|
|
}
|
|
|
|
} else {
|
|
|
|
h.s.connected.Store(1)
|
2024-11-28 09:53:03 +03:00
|
|
|
}
|
|
|
|
} else {
|
2024-12-01 19:58:11 +03:00
|
|
|
if h.s.connected.CompareAndSwap(0, 1) {
|
|
|
|
h.s.sendEvent(&event{ts: time.Now(), err: err, t: store.EventTypeConnect})
|
|
|
|
}
|
2024-11-27 01:57:39 +03:00
|
|
|
}
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
func (h *eventHook) ProcessPipelineHook(hook goredis.ProcessPipelineHook) goredis.ProcessPipelineHook {
|
|
|
|
return func(ctx context.Context, cmds []goredis.Cmder) error {
|
|
|
|
err := hook(ctx, cmds)
|
2024-11-28 09:53:03 +03:00
|
|
|
if err != nil {
|
|
|
|
if !isRedisError(err) {
|
2024-12-01 19:58:11 +03:00
|
|
|
if h.s.connected.CompareAndSwap(1, 0) {
|
|
|
|
h.s.sendEvent(&event{ts: time.Now(), err: err, t: store.EventTypeDisconnect})
|
|
|
|
}
|
|
|
|
} else {
|
|
|
|
h.s.connected.Store(1)
|
2024-11-28 09:53:03 +03:00
|
|
|
}
|
|
|
|
} else {
|
2024-12-01 19:58:11 +03:00
|
|
|
if h.s.connected.CompareAndSwap(0, 1) {
|
|
|
|
h.s.sendEvent(&event{ts: time.Now(), err: err, t: store.EventTypeConnect})
|
|
|
|
}
|
2024-11-27 01:57:39 +03:00
|
|
|
}
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
func isRedisError(err error) bool {
|
|
|
|
var rerr goredis.Error
|
|
|
|
return errors.As(err, &rerr)
|
|
|
|
}
|