Compare commits
No commits in common. "v3" and "master" have entirely different histories.
25
.gitignore
vendored
25
.gitignore
vendored
@ -1,25 +0,0 @@
|
||||
# Binaries for programs and plugins
|
||||
*.exe
|
||||
*.exe~
|
||||
*.dll
|
||||
*.so
|
||||
*.dylib
|
||||
bin
|
||||
|
||||
# Test binary, built with `go test -c`
|
||||
*.test
|
||||
cmd/test
|
||||
|
||||
# Output of the go coverage tool, specifically when used with LiteIDE
|
||||
*.out
|
||||
|
||||
# Dependency directories (remove the comment below to include it)
|
||||
# vendor/
|
||||
|
||||
# Go workspace file
|
||||
go.work
|
||||
|
||||
# General
|
||||
.DS_Store
|
||||
.idea
|
||||
.vscode
|
||||
12
go.mod
12
go.mod
@ -1,17 +1,13 @@
|
||||
module go.unistack.org/micro-broker-redis/v3
|
||||
|
||||
go 1.22
|
||||
go 1.23.1
|
||||
|
||||
require (
|
||||
github.com/redis/go-redis/extra/rediscmd/v9 v9.6.2
|
||||
github.com/redis/go-redis/v9 v9.6.2
|
||||
go.unistack.org/micro/v3 v3.10.97
|
||||
github.com/redis/go-redis/v9 v9.6.1
|
||||
go.unistack.org/micro/v3 v3.10.84
|
||||
)
|
||||
|
||||
require (
|
||||
github.com/cespare/xxhash/v2 v2.3.0 // indirect
|
||||
github.com/cespare/xxhash/v2 v2.2.0 // indirect
|
||||
github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
|
||||
github.com/google/go-cmp v0.5.9 // indirect
|
||||
go.unistack.org/micro-proto/v3 v3.4.1 // indirect
|
||||
google.golang.org/protobuf v1.35.1 // indirect
|
||||
)
|
||||
|
||||
20
go.sum
20
go.sum
@ -2,19 +2,11 @@ github.com/bsm/ginkgo/v2 v2.12.0 h1:Ny8MWAHyOepLGlLKYmXG4IEkioBysk6GpaRTLC8zwWs=
|
||||
github.com/bsm/ginkgo/v2 v2.12.0/go.mod h1:SwYbGRRDovPVboqFv0tPTcG1sN61LM1Z4ARdbAV9g4c=
|
||||
github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA=
|
||||
github.com/bsm/gomega v1.27.10/go.mod h1:JyEr/xRbxbtgWNi8tIEVPUYZ5Dzef52k01W3YH0H+O0=
|
||||
github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
|
||||
github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
|
||||
github.com/cespare/xxhash/v2 v2.2.0 h1:DC2CZ1Ep5Y4k3ZQ899DldepgrayRUGE6BBZ/cd9Cj44=
|
||||
github.com/cespare/xxhash/v2 v2.2.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
|
||||
github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f h1:lO4WD4F/rVNCu3HqELle0jiPLLBs70cWOduZpkS1E78=
|
||||
github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f/go.mod h1:cuUVRXasLTGF7a8hSLbxyZXjz+1KgoB3wDUb6vlszIc=
|
||||
github.com/google/go-cmp v0.5.9 h1:O2Tfq5qg4qc4AmwVlvv0oLiVAGB7enBSJ2x2DqQFi38=
|
||||
github.com/google/go-cmp v0.5.9/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
|
||||
github.com/redis/go-redis/extra/rediscmd/v9 v9.6.2 h1:oBlErygFka9FAVfowzW7sRUfB7n31YX0aFzfaOflM3w=
|
||||
github.com/redis/go-redis/extra/rediscmd/v9 v9.6.2/go.mod h1:jYLUE5tC8UsFnpSclhEjZlFLMMtHH4jTfFMnshNWkoo=
|
||||
github.com/redis/go-redis/v9 v9.6.2 h1:w0uvkRbc9KpgD98zcvo5IrVUsn0lXpRMuhNgiHDJzdk=
|
||||
github.com/redis/go-redis/v9 v9.6.2/go.mod h1:0C0c6ycQsdpVNQpxb1njEQIqkx5UcsM8FJCQLgE9+RA=
|
||||
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.97 h1:8l7fv+i06/PjPrBBhRC/ZQkWGIOuHPg3jJN0vktYE78=
|
||||
go.unistack.org/micro/v3 v3.10.97/go.mod h1:YzMldzHN9Ei+zy5t/Psu7RUWDZwUfrNYiStSQtTz90g=
|
||||
google.golang.org/protobuf v1.35.1 h1:m3LfL6/Ca+fqnjnlqQXNpFPABW1UD7mjh8KO2mKFytA=
|
||||
google.golang.org/protobuf v1.35.1/go.mod h1:9fA7Ob0pmnwhb644+1+CVWFRbNajQ6iRojtC/QF5bRE=
|
||||
github.com/redis/go-redis/v9 v9.6.1 h1:HHDteefn6ZkTtY5fGUE8tj8uy85AHk6zP7CpzIAM0y4=
|
||||
github.com/redis/go-redis/v9 v9.6.1/go.mod h1:0C0c6ycQsdpVNQpxb1njEQIqkx5UcsM8FJCQLgE9+RA=
|
||||
go.unistack.org/micro/v3 v3.10.84 h1:Fc38VoRnL+sFyVn8V/lx5T0sP/I4TKuQ61ium0fs6l4=
|
||||
go.unistack.org/micro/v3 v3.10.84/go.mod h1:erMgt3Bl7vQQ0e9UpQyR5NlLiZ9pKeEJ9+1tfYFaqUg=
|
||||
|
||||
18
options.go
18
options.go
@ -7,13 +7,20 @@ import (
|
||||
"go.unistack.org/micro/v3/broker"
|
||||
"go.unistack.org/micro/v3/logger"
|
||||
"go.unistack.org/micro/v3/meter"
|
||||
"go.unistack.org/micro/v3/store"
|
||||
"go.unistack.org/micro/v3/tracer"
|
||||
)
|
||||
|
||||
type configKey struct{}
|
||||
|
||||
func Config(c *redis.UniversalOptions) broker.Option {
|
||||
return broker.SetOption(configKey{}, c)
|
||||
func Config(c *redis.Options) store.Option {
|
||||
return store.SetOption(configKey{}, c)
|
||||
}
|
||||
|
||||
type clusterConfigKey struct{}
|
||||
|
||||
func ClusterConfig(c *redis.ClusterOptions) store.Option {
|
||||
return store.SetOption(clusterConfigKey{}, c)
|
||||
}
|
||||
|
||||
var (
|
||||
@ -45,12 +52,19 @@ func NewOptions(opts ...Option) Options {
|
||||
Meter: meter.DefaultMeter,
|
||||
Tracer: tracer.DefaultTracer,
|
||||
MeterStatsInterval: DefaultMeterStatsInterval,
|
||||
MeterMetricPrefix: DefaultMeterMetricPrefix,
|
||||
}
|
||||
|
||||
for _, o := range opts {
|
||||
o(&options)
|
||||
}
|
||||
|
||||
options.Meter = options.Meter.Clone(
|
||||
meter.MetricPrefix(options.MeterMetricPrefix),
|
||||
)
|
||||
|
||||
options.Logger = options.Logger.Clone(logger.WithCallerSkipCount(1))
|
||||
|
||||
return options
|
||||
}
|
||||
|
||||
|
||||
38
redis.go
38
redis.go
@ -35,9 +35,9 @@ var (
|
||||
// Event is an broker.Event
|
||||
type Event struct {
|
||||
ctx context.Context
|
||||
err error
|
||||
msg *broker.Message
|
||||
topic string
|
||||
msg *broker.Message
|
||||
err error
|
||||
}
|
||||
|
||||
// Topic returns the topic this Event applies to.
|
||||
@ -106,8 +106,6 @@ func (s *Subscriber) loop() {
|
||||
}
|
||||
|
||||
err := s.opts.Codec.Unmarshal([]byte(msg.Payload), p.msg)
|
||||
p.ctx = metadata.NewIncomingContext(s.ctx, p.msg.Header)
|
||||
|
||||
if err != nil {
|
||||
p.msg.Body = codec.RawMessage(msg.Payload)
|
||||
if eh != nil {
|
||||
@ -225,7 +223,6 @@ func (b *Broker) BatchSubscribe(ctx context.Context, topic string, handler broke
|
||||
// Subscribe returns a broker.Subscriber for the topic and handler
|
||||
func (b *Broker) Subscribe(ctx context.Context, topic string, handler broker.Handler, opts ...broker.SubscribeOption) (broker.Subscriber, error) {
|
||||
s := &Subscriber{
|
||||
ctx: ctx,
|
||||
topic: topic,
|
||||
handle: handler,
|
||||
opts: b.opts,
|
||||
@ -233,41 +230,45 @@ func (b *Broker) Subscribe(ctx context.Context, topic string, handler broker.Han
|
||||
done: make(chan struct{}),
|
||||
}
|
||||
|
||||
// Run the receiver routine.
|
||||
go s.loop()
|
||||
|
||||
s.sub = b.cli.Subscribe(s.ctx, s.topic)
|
||||
if err := s.sub.Ping(ctx, ""); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
go s.loop()
|
||||
|
||||
return s, nil
|
||||
}
|
||||
|
||||
func (b *Broker) configure() error {
|
||||
redisOptions := DefaultOptions
|
||||
|
||||
if b.cli != nil && b.opts.Context == nil {
|
||||
return nil
|
||||
}
|
||||
|
||||
if b.opts.Context != nil {
|
||||
if c, ok := b.opts.Context.Value(configKey{}).(*redis.UniversalOptions); ok && c != nil {
|
||||
if c, ok := b.opts.Context.Value(configKey{}).(*redis.UniversalOptions); ok {
|
||||
redisOptions = c
|
||||
if b.opts.TLSConfig != nil {
|
||||
redisOptions.TLSConfig = b.opts.TLSConfig
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if len(b.opts.Addrs) > 0 {
|
||||
redisOptions.Addrs = b.opts.Addrs
|
||||
}
|
||||
|
||||
if b.opts.TLSConfig != nil {
|
||||
redisOptions.TLSConfig = b.opts.TLSConfig
|
||||
}
|
||||
|
||||
if redisOptions == nil && b.cli != nil {
|
||||
return nil
|
||||
}
|
||||
|
||||
if redisOptions == nil {
|
||||
redisOptions.Addrs = b.opts.Addrs
|
||||
redisOptions.TLSConfig = b.opts.TLSConfig
|
||||
}
|
||||
|
||||
c := redis.NewUniversalClient(redisOptions)
|
||||
setTracing(c, b.opts.Tracer)
|
||||
|
||||
b.cli = c
|
||||
b.statsMeter()
|
||||
|
||||
return nil
|
||||
@ -279,7 +280,6 @@ func (b *Broker) Connect(ctx context.Context) error {
|
||||
err = b.cli.Ping(ctx).Err()
|
||||
}
|
||||
setSpanError(ctx, err)
|
||||
b.done = make(chan struct{})
|
||||
return err
|
||||
}
|
||||
|
||||
@ -318,5 +318,5 @@ func (b *Broker) Disconnect(ctx context.Context) error {
|
||||
}
|
||||
|
||||
func NewBroker(opts ...broker.Option) *Broker {
|
||||
return &Broker{opts: broker.NewOptions(opts...)}
|
||||
return &Broker{done: make(chan struct{}), opts: broker.NewOptions(opts...)}
|
||||
}
|
||||
|
||||
4
stats.go
4
stats.go
@ -13,6 +13,10 @@ var (
|
||||
PoolConnTotalCurrent = "pool_conn_total_current"
|
||||
PoolConnIdleCurrent = "pool_conn_idle_current"
|
||||
PoolConnStaleTotal = "pool_conn_stale_total"
|
||||
|
||||
meterRequestTotal = "request_total"
|
||||
meterRequestLatencyMicroseconds = "latency_microseconds"
|
||||
meterRequestDurationSeconds = "request_duration_seconds"
|
||||
)
|
||||
|
||||
type Statser interface {
|
||||
|
||||
Loading…
Reference in New Issue
Block a user