mirror of
https://github.com/hanzoai/psrpc.git
synced 2026-09-30 23:47:22 +00:00
refactor: kv vocabulary throughout, on hanzokv/go v9.22.0
hanzokv/go v9.22.0 renames its Go package from `redis` to `kv` (import path unchanged). Follow that through the fork so there is one vocabulary for the key-value bus: - bus_redis.go -> bus_kv.go, bus_redis_test.go -> bus_kv_test.go, bustest/redis.go -> bustest/kv.go - redisMessageBus -> kvMessageBus, and the same for its subscription, sublist, write-op, publish-queue and bustest server types - psrpc.NewRedisMessageBus -> psrpc.NewKVMessageBus, bus.NewRedisMessageBus -> bus.NewKVMessageBus - log messages and README say kv The dockertest image stays `redis:latest` and the pipelining doc link stays on redis.io: those name third-party artifacts, not our types. Co-authored-by: Hanzo Dev <[email protected]>
This commit is contained in:
Hanzo Dev
@@ -4,7 +4,7 @@ Create custom protobuf-based golang RPCs built on pub/sub.
|
||||
|
||||
Supports:
|
||||
* Protobuf service definitions
|
||||
* Use Redis, Nats, or a local communication layer
|
||||
* Use KV, Nats, or a local communication layer
|
||||
* Custom server selection for RPC handling based on user-defined [affinity](#Affinity)
|
||||
* RPC topics - any RPC can be divided into topics, (e.g. by region)
|
||||
* Single RPCs - one request is handled by one server, used for normal RPCs
|
||||
|
||||
@@ -15,8 +15,8 @@
|
||||
package psrpc
|
||||
|
||||
import (
|
||||
"github.com/nats-io/nats.go"
|
||||
"github.com/hanzokv/go/v9"
|
||||
"github.com/nats-io/nats.go"
|
||||
|
||||
"github.com/hanzoai/psrpc/internal/bus"
|
||||
)
|
||||
@@ -32,6 +32,6 @@ func NewNatsMessageBus(nc *nats.Conn) MessageBus {
|
||||
return bus.NewNatsMessageBus(nc)
|
||||
}
|
||||
|
||||
func NewRedisMessageBus(rc redis.UniversalClient) MessageBus {
|
||||
return bus.NewRedisMessageBus(rc)
|
||||
func NewKVMessageBus(rc kv.UniversalClient) MessageBus {
|
||||
return bus.NewKVMessageBus(rc)
|
||||
}
|
||||
|
||||
@@ -6,7 +6,7 @@ require (
|
||||
github.com/frostbyte73/core v0.1.1
|
||||
github.com/gammazero/deque v1.2.1
|
||||
github.com/go-logr/logr v1.4.3
|
||||
github.com/hanzokv/go/v9 v9.21.1
|
||||
github.com/hanzokv/go/v9 v9.22.0
|
||||
github.com/livekit/mageutil v0.0.0-20250511045019-0f1ff63f7731
|
||||
github.com/nats-io/nats.go v1.52.0
|
||||
github.com/ory/dockertest/v4 v4.0.0
|
||||
|
||||
@@ -41,6 +41,8 @@ github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
|
||||
github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
|
||||
github.com/hanzokv/go/v9 v9.21.1 h1:MUIAJTV1pI4pkKWSzrgJRzKA7W3aHUJ1WO0mtZRhhPc=
|
||||
github.com/hanzokv/go/v9 v9.21.1/go.mod h1:GV+nw+jX60sIrJ7LBkmOQw2WASXzkMstt52blcwNw6w=
|
||||
github.com/hanzokv/go/v9 v9.22.0 h1:zD4fh0NLBuVa8njIrXUivJCijlratzS1Yf7Y/uD5T00=
|
||||
github.com/hanzokv/go/v9 v9.22.0/go.mod h1:GV+nw+jX60sIrJ7LBkmOQw2WASXzkMstt52blcwNw6w=
|
||||
github.com/klauspost/compress v1.18.6 h1:2jupLlAwFm95+YDR+NwD2MEfFO9d4z4Prjl1XXDjuao=
|
||||
github.com/klauspost/compress v1.18.6/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ=
|
||||
github.com/klauspost/cpuid/v2 v2.3.0 h1:S4CRMLnYUhGeDFDqkGriYKdfoFlDnMtqTiI/sFzhA9Y=
|
||||
|
||||
@@ -41,46 +41,46 @@ const (
|
||||
publishBuckets = 17
|
||||
)
|
||||
|
||||
type redisMessageBus struct {
|
||||
rc redis.UniversalClient
|
||||
type kvMessageBus struct {
|
||||
rc kv.UniversalClient
|
||||
ctx context.Context
|
||||
ps *redis.PubSub
|
||||
ps *kv.PubSub
|
||||
|
||||
mu sync.Mutex
|
||||
subs map[string]*redisSubList
|
||||
queues map[string]*redisSubList
|
||||
subs map[string]*kvSubList
|
||||
queues map[string]*kvSubList
|
||||
|
||||
wakeup chan struct{}
|
||||
ops *redisWriteOpQueue
|
||||
ops *kvWriteOpQueue
|
||||
dirtyChannels map[string]struct{}
|
||||
currentChannels map[string]struct{}
|
||||
|
||||
publishQueues [publishBuckets]*redisPublishQueue
|
||||
publishQueues [publishBuckets]*kvPublishQueue
|
||||
}
|
||||
|
||||
func NewRedisMessageBus(rc redis.UniversalClient) MessageBus {
|
||||
func NewKVMessageBus(rc kv.UniversalClient) MessageBus {
|
||||
ctx := context.Background()
|
||||
r := &redisMessageBus{
|
||||
r := &kvMessageBus{
|
||||
rc: rc,
|
||||
ctx: ctx,
|
||||
ps: rc.Subscribe(ctx),
|
||||
subs: map[string]*redisSubList{},
|
||||
queues: map[string]*redisSubList{},
|
||||
subs: map[string]*kvSubList{},
|
||||
queues: map[string]*kvSubList{},
|
||||
|
||||
wakeup: make(chan struct{}, 1),
|
||||
ops: &redisWriteOpQueue{},
|
||||
ops: &kvWriteOpQueue{},
|
||||
dirtyChannels: map[string]struct{}{},
|
||||
currentChannels: map[string]struct{}{},
|
||||
}
|
||||
for i := range len(r.publishQueues) {
|
||||
r.publishQueues[i] = newRedisPublishQueue(r.ctx, r.rc)
|
||||
r.publishQueues[i] = newKVPublishQueue(r.ctx, r.rc)
|
||||
}
|
||||
go r.readWorker()
|
||||
go r.writeWorker()
|
||||
return r
|
||||
}
|
||||
|
||||
func (r *redisMessageBus) Publish(_ context.Context, channel Channel, msg proto.Message) error {
|
||||
func (r *kvMessageBus) Publish(_ context.Context, channel Channel, msg proto.Message) error {
|
||||
b, err := serialize(msg, "")
|
||||
if err != nil {
|
||||
return err
|
||||
@@ -91,22 +91,22 @@ func (r *redisMessageBus) Publish(_ context.Context, channel Channel, msg proto.
|
||||
return nil
|
||||
}
|
||||
|
||||
func (r *redisMessageBus) Subscribe(ctx context.Context, channel Channel, size int) (Reader, error) {
|
||||
func (r *kvMessageBus) Subscribe(ctx context.Context, channel Channel, size int) (Reader, error) {
|
||||
return r.subscribe(ctx, channel.Legacy, size, r.subs, false)
|
||||
}
|
||||
|
||||
func (r *redisMessageBus) SubscribeQueue(ctx context.Context, channel Channel, size int) (Reader, error) {
|
||||
func (r *kvMessageBus) SubscribeQueue(ctx context.Context, channel Channel, size int) (Reader, error) {
|
||||
return r.subscribe(ctx, channel.Legacy, size, r.queues, true)
|
||||
}
|
||||
|
||||
func (r *redisMessageBus) subscribe(ctx context.Context, channel string, size int, subLists map[string]*redisSubList, queue bool) (Reader, error) {
|
||||
func (r *kvMessageBus) subscribe(ctx context.Context, channel string, size int, subLists map[string]*kvSubList, queue bool) (Reader, error) {
|
||||
ctx, cancel := context.WithCancel(ctx)
|
||||
sub := &redisSubscription{
|
||||
sub := &kvSubscription{
|
||||
bus: r,
|
||||
ctx: ctx,
|
||||
cancel: cancel,
|
||||
channel: channel,
|
||||
msgChan: make(chan *redis.Message, size),
|
||||
msgChan: make(chan *kv.Message, size),
|
||||
queue: queue,
|
||||
}
|
||||
|
||||
@@ -115,7 +115,7 @@ func (r *redisMessageBus) subscribe(ctx context.Context, channel string, size in
|
||||
|
||||
subList, ok := subLists[channel]
|
||||
if !ok {
|
||||
subList = &redisSubList{}
|
||||
subList = &kvSubList{}
|
||||
subLists[channel] = subList
|
||||
r.reconcileSubscriptions(channel)
|
||||
}
|
||||
@@ -124,11 +124,11 @@ func (r *redisMessageBus) subscribe(ctx context.Context, channel string, size in
|
||||
return sub, nil
|
||||
}
|
||||
|
||||
func (r *redisMessageBus) unsubscribe(channel string, queue bool, sub *redisSubscription) {
|
||||
func (r *kvMessageBus) unsubscribe(channel string, queue bool, sub *kvSubscription) {
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
|
||||
var subLists map[string]*redisSubList
|
||||
var subLists map[string]*kvSubList
|
||||
if queue {
|
||||
subLists = r.queues
|
||||
} else {
|
||||
@@ -152,12 +152,12 @@ func (r *redisMessageBus) unsubscribe(channel string, queue bool, sub *redisSubs
|
||||
}
|
||||
}
|
||||
|
||||
func (r *redisMessageBus) readWorker() {
|
||||
func (r *kvMessageBus) readWorker() {
|
||||
var delay time.Duration
|
||||
for {
|
||||
msg, err := r.ps.ReceiveMessage(r.ctx)
|
||||
if err != nil {
|
||||
logger.Error(err, "redis receive message failed")
|
||||
logger.Error(err, "kv receive message failed")
|
||||
|
||||
time.Sleep(delay)
|
||||
if delay *= 2; delay == 0 {
|
||||
@@ -180,12 +180,12 @@ func (r *redisMessageBus) readWorker() {
|
||||
}
|
||||
}
|
||||
|
||||
func (r *redisMessageBus) reconcileSubscriptions(channel string) {
|
||||
func (r *kvMessageBus) reconcileSubscriptions(channel string) {
|
||||
r.dirtyChannels[channel] = struct{}{}
|
||||
r.enqueueWriteOp(&redisReconcileSubscriptionsOp{r})
|
||||
r.enqueueWriteOp(&kvReconcileSubscriptionsOp{r})
|
||||
}
|
||||
|
||||
func (r *redisMessageBus) enqueueWriteOp(op redisWriteOp) {
|
||||
func (r *kvMessageBus) enqueueWriteOp(op kvWriteOp) {
|
||||
r.ops.push(op)
|
||||
select {
|
||||
case r.wakeup <- struct{}{}:
|
||||
@@ -193,7 +193,7 @@ func (r *redisMessageBus) enqueueWriteOp(op redisWriteOp) {
|
||||
}
|
||||
}
|
||||
|
||||
func (r *redisMessageBus) writeWorker() {
|
||||
func (r *kvMessageBus) writeWorker() {
|
||||
for range r.wakeup {
|
||||
r.ops.drain()
|
||||
}
|
||||
@@ -201,30 +201,30 @@ func (r *redisMessageBus) writeWorker() {
|
||||
|
||||
// ----------------------------------------------
|
||||
|
||||
type redisWriteOpQueue struct {
|
||||
type kvWriteOpQueue struct {
|
||||
mu sync.Mutex
|
||||
ops deque.Deque[redisWriteOp]
|
||||
ops deque.Deque[kvWriteOp]
|
||||
}
|
||||
|
||||
func (q *redisWriteOpQueue) empty() bool {
|
||||
func (q *kvWriteOpQueue) empty() bool {
|
||||
q.mu.Lock()
|
||||
defer q.mu.Unlock()
|
||||
return q.ops.Len() == 0
|
||||
}
|
||||
|
||||
func (q *redisWriteOpQueue) push(op redisWriteOp) {
|
||||
func (q *kvWriteOpQueue) push(op kvWriteOp) {
|
||||
q.mu.Lock()
|
||||
defer q.mu.Unlock()
|
||||
q.ops.PushBack(op)
|
||||
}
|
||||
|
||||
func (q *redisWriteOpQueue) drain() {
|
||||
func (q *kvWriteOpQueue) drain() {
|
||||
q.mu.Lock()
|
||||
for q.ops.Len() > 0 {
|
||||
op := q.ops.PopFront()
|
||||
q.mu.Unlock()
|
||||
if err := op.run(); err != nil {
|
||||
logger.Error(err, "redis write message failed")
|
||||
logger.Error(err, "kv write message failed")
|
||||
}
|
||||
q.mu.Lock()
|
||||
}
|
||||
@@ -233,17 +233,17 @@ func (q *redisWriteOpQueue) drain() {
|
||||
|
||||
//-----------------------------------------------------
|
||||
|
||||
type redisWriteOp interface {
|
||||
type kvWriteOp interface {
|
||||
run() error
|
||||
}
|
||||
|
||||
// ----------------------------------------------------
|
||||
|
||||
type redisReconcileSubscriptionsOp struct {
|
||||
*redisMessageBus
|
||||
type kvReconcileSubscriptionsOp struct {
|
||||
*kvMessageBus
|
||||
}
|
||||
|
||||
func (r *redisReconcileSubscriptionsOp) run() error {
|
||||
func (r *kvReconcileSubscriptionsOp) run() error {
|
||||
r.mu.Lock()
|
||||
for len(r.dirtyChannels) > 0 {
|
||||
subscribe := make(map[string]struct{}, len(r.dirtyChannels))
|
||||
@@ -269,7 +269,7 @@ func (r *redisReconcileSubscriptionsOp) run() error {
|
||||
}
|
||||
|
||||
if err := multierr.Combine(subscribeErr, unsubscribeErr); err != nil {
|
||||
logger.Error(err, "redis subscription reconciliation failed")
|
||||
logger.Error(err, "kv subscription reconciliation failed")
|
||||
time.Sleep(reconcilerRetryInterval)
|
||||
}
|
||||
|
||||
@@ -293,12 +293,12 @@ func (r *redisReconcileSubscriptionsOp) run() error {
|
||||
|
||||
// ----------------------------------------------------
|
||||
|
||||
type redisSubList struct {
|
||||
subs []*redisSubscription
|
||||
type kvSubList struct {
|
||||
subs []*kvSubscription
|
||||
next int
|
||||
}
|
||||
|
||||
func (r *redisSubList) dispatchQueue(msg *redis.Message) {
|
||||
func (r *kvSubList) dispatchQueue(msg *kv.Message) {
|
||||
if r.next >= len(r.subs) {
|
||||
r.next = 0
|
||||
}
|
||||
@@ -306,7 +306,7 @@ func (r *redisSubList) dispatchQueue(msg *redis.Message) {
|
||||
r.next++
|
||||
}
|
||||
|
||||
func (r *redisSubList) dispatch(msg *redis.Message) {
|
||||
func (r *kvSubList) dispatch(msg *kv.Message) {
|
||||
for _, sub := range r.subs {
|
||||
sub.write(msg)
|
||||
}
|
||||
@@ -314,25 +314,25 @@ func (r *redisSubList) dispatch(msg *redis.Message) {
|
||||
|
||||
// ----------------------------------------------------
|
||||
|
||||
type redisSubscription struct {
|
||||
bus *redisMessageBus
|
||||
type kvSubscription struct {
|
||||
bus *kvMessageBus
|
||||
ctx context.Context
|
||||
cancel context.CancelFunc
|
||||
channel string
|
||||
msgChan chan *redis.Message
|
||||
msgChan chan *kv.Message
|
||||
queue bool
|
||||
}
|
||||
|
||||
func (r *redisSubscription) write(msg *redis.Message) {
|
||||
func (r *kvSubscription) write(msg *kv.Message) {
|
||||
select {
|
||||
case r.msgChan <- msg:
|
||||
case <-r.ctx.Done():
|
||||
}
|
||||
}
|
||||
|
||||
func (r *redisSubscription) read() ([]byte, bool) {
|
||||
func (r *kvSubscription) read() ([]byte, bool) {
|
||||
for {
|
||||
var msg *redis.Message
|
||||
var msg *kv.Message
|
||||
var ok bool
|
||||
select {
|
||||
case msg, ok = <-r.msgChan:
|
||||
@@ -356,7 +356,7 @@ func (r *redisSubscription) read() ([]byte, bool) {
|
||||
}
|
||||
}
|
||||
|
||||
func (r *redisSubscription) Close() error {
|
||||
func (r *kvSubscription) Close() error {
|
||||
r.cancel()
|
||||
r.bus.unsubscribe(r.channel, r.queue, r)
|
||||
close(r.msgChan)
|
||||
@@ -365,22 +365,22 @@ func (r *redisSubscription) Close() error {
|
||||
|
||||
// ----------------------------------------------------
|
||||
|
||||
type redisPublishMessage struct {
|
||||
type kvPublishMessage struct {
|
||||
channel string
|
||||
payload []byte
|
||||
}
|
||||
|
||||
type redisPublishQueue struct {
|
||||
type kvPublishQueue struct {
|
||||
ctx context.Context
|
||||
rc redis.UniversalClient
|
||||
rc kv.UniversalClient
|
||||
|
||||
lock sync.Mutex
|
||||
messages []redisPublishMessage
|
||||
messages []kvPublishMessage
|
||||
wakeup chan struct{}
|
||||
}
|
||||
|
||||
func newRedisPublishQueue(ctx context.Context, rc redis.UniversalClient) *redisPublishQueue {
|
||||
r := &redisPublishQueue{
|
||||
func newKVPublishQueue(ctx context.Context, rc kv.UniversalClient) *kvPublishQueue {
|
||||
r := &kvPublishQueue{
|
||||
ctx: ctx,
|
||||
rc: rc,
|
||||
wakeup: make(chan struct{}, 1),
|
||||
@@ -390,18 +390,18 @@ func newRedisPublishQueue(ctx context.Context, rc redis.UniversalClient) *redisP
|
||||
return r
|
||||
}
|
||||
|
||||
func (r *redisPublishQueue) Enqueue(channel string, payload []byte) {
|
||||
func (r *kvPublishQueue) Enqueue(channel string, payload []byte) {
|
||||
r.lock.Lock()
|
||||
defer r.lock.Unlock()
|
||||
|
||||
r.messages = append(r.messages, redisPublishMessage{channel, payload})
|
||||
r.messages = append(r.messages, kvPublishMessage{channel, payload})
|
||||
select {
|
||||
case r.wakeup <- struct{}{}:
|
||||
default:
|
||||
}
|
||||
}
|
||||
|
||||
func (r *redisPublishQueue) worker() {
|
||||
func (r *kvPublishQueue) worker() {
|
||||
for {
|
||||
select {
|
||||
case <-r.wakeup:
|
||||
@@ -414,7 +414,7 @@ func (r *redisPublishQueue) worker() {
|
||||
r.messages = nil
|
||||
r.lock.Unlock()
|
||||
|
||||
// using a pipeline to handle redis servers with a high RTT
|
||||
// using a pipeline to handle kv servers with a high RTT
|
||||
// (https://redis.io/docs/latest/develop/using-commands/pipelining/).
|
||||
//
|
||||
// This is doing oppotunistic batching + pipelining.
|
||||
@@ -29,25 +29,25 @@ import (
|
||||
"github.com/hanzoai/psrpc/internal/bus/bustest"
|
||||
)
|
||||
|
||||
func redisTestChannel(channel string) bus.Channel {
|
||||
func kvTestChannel(channel string) bus.Channel {
|
||||
return bus.Channel{Legacy: channel}
|
||||
}
|
||||
|
||||
func TestRedisMessageBus(t *testing.T) {
|
||||
srv := bustest.NewRedis(t, bustest.Docker(t))
|
||||
func TestKVMessageBus(t *testing.T) {
|
||||
srv := bustest.NewKV(t, bustest.Docker(t))
|
||||
|
||||
t.Run("published messages are received by subscribers", func(t *testing.T) {
|
||||
b0 := srv.Connect(t)
|
||||
b1 := srv.Connect(t)
|
||||
|
||||
r, err := b0.Subscribe(context.Background(), redisTestChannel("test"), 100)
|
||||
r, err := b0.Subscribe(context.Background(), kvTestChannel("test"), 100)
|
||||
require.NoError(t, err)
|
||||
|
||||
time.Sleep(100 * time.Millisecond)
|
||||
|
||||
src := wrapperspb.String("test")
|
||||
|
||||
err = b1.Publish(context.Background(), redisTestChannel("test"), src)
|
||||
err = b1.Publish(context.Background(), kvTestChannel("test"), src)
|
||||
require.NoError(t, err)
|
||||
|
||||
b, ok := bus.RawRead(r)
|
||||
@@ -63,16 +63,16 @@ func TestRedisMessageBus(t *testing.T) {
|
||||
b1 := srv.Connect(t)
|
||||
b2 := srv.Connect(t)
|
||||
|
||||
r1, err := b1.SubscribeQueue(context.Background(), redisTestChannel("test"), 100)
|
||||
r1, err := b1.SubscribeQueue(context.Background(), kvTestChannel("test"), 100)
|
||||
require.NoError(t, err)
|
||||
r2, err := b2.SubscribeQueue(context.Background(), redisTestChannel("test"), 100)
|
||||
r2, err := b2.SubscribeQueue(context.Background(), kvTestChannel("test"), 100)
|
||||
require.NoError(t, err)
|
||||
|
||||
time.Sleep(100 * time.Millisecond)
|
||||
|
||||
src := wrapperspb.String("test")
|
||||
|
||||
err = b0.Publish(context.Background(), redisTestChannel("test"), src)
|
||||
err = b0.Publish(context.Background(), kvTestChannel("test"), src)
|
||||
require.NoError(t, err)
|
||||
|
||||
var n atomic.Int64
|
||||
@@ -98,16 +98,16 @@ func TestRedisMessageBus(t *testing.T) {
|
||||
b1 := srv.Connect(t)
|
||||
b2 := srv.Connect(t)
|
||||
|
||||
r1, err := b1.Subscribe(context.Background(), redisTestChannel("test"), 100)
|
||||
r1, err := b1.Subscribe(context.Background(), kvTestChannel("test"), 100)
|
||||
require.NoError(t, err)
|
||||
r2, err := b2.Subscribe(context.Background(), redisTestChannel("test"), 100)
|
||||
r2, err := b2.Subscribe(context.Background(), kvTestChannel("test"), 100)
|
||||
require.NoError(t, err)
|
||||
|
||||
time.Sleep(100 * time.Millisecond)
|
||||
|
||||
src := wrapperspb.String("test")
|
||||
|
||||
err = b0.Publish(context.Background(), redisTestChannel("test"), src)
|
||||
err = b0.Publish(context.Background(), kvTestChannel("test"), src)
|
||||
require.NoError(t, err)
|
||||
|
||||
_, ok := bus.RawRead(r1)
|
||||
@@ -120,7 +120,7 @@ func TestRedisMessageBus(t *testing.T) {
|
||||
|
||||
time.Sleep(time.Second)
|
||||
|
||||
err = b0.Publish(context.Background(), redisTestChannel("test"), src)
|
||||
err = b0.Publish(context.Background(), kvTestChannel("test"), src)
|
||||
require.NoError(t, err)
|
||||
|
||||
_, ok = bus.RawRead(r1)
|
||||
@@ -130,13 +130,13 @@ func TestRedisMessageBus(t *testing.T) {
|
||||
})
|
||||
}
|
||||
|
||||
func BenchmarkRedisMessageBus(b *testing.B) {
|
||||
srv := bustest.NewRedis(b, bustest.Docker(b))
|
||||
func BenchmarkKVMessageBus(b *testing.B) {
|
||||
srv := bustest.NewKV(b, bustest.Docker(b))
|
||||
|
||||
b0 := srv.Connect(b)
|
||||
b1 := srv.Connect(b)
|
||||
|
||||
r, _ := b0.Subscribe(context.Background(), redisTestChannel("test"), 100)
|
||||
r, _ := b0.Subscribe(context.Background(), kvTestChannel("test"), 100)
|
||||
|
||||
time.Sleep(100 * time.Millisecond)
|
||||
|
||||
@@ -152,7 +152,7 @@ func BenchmarkRedisMessageBus(b *testing.B) {
|
||||
|
||||
src := wrapperspb.String("test")
|
||||
for i := 0; i < b.N; i++ {
|
||||
b1.Publish(context.Background(), redisTestChannel("test"), src)
|
||||
b1.Publish(context.Background(), kvTestChannel("test"), src)
|
||||
}
|
||||
|
||||
<-done
|
||||
@@ -7,23 +7,23 @@ import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/ory/dockertest/v4"
|
||||
"github.com/hanzokv/go/v9"
|
||||
"github.com/ory/dockertest/v4"
|
||||
|
||||
"github.com/hanzoai/psrpc/internal/bus"
|
||||
)
|
||||
|
||||
func init() {
|
||||
RegisterServer("Redis", NewRedis)
|
||||
RegisterServer("KV", NewKV)
|
||||
}
|
||||
|
||||
var redisLast = baseID
|
||||
var kvLast = baseID
|
||||
|
||||
func NewRedis(t testing.TB, pool dockertest.Pool) Server {
|
||||
func NewKV(t testing.TB, pool dockertest.Pool) Server {
|
||||
ctx := context.Background()
|
||||
c, err := pool.Run(ctx, "redis",
|
||||
dockertest.WithTag("latest"),
|
||||
dockertest.WithName(fmt.Sprintf("psrpc-redis-%d", atomic.AddUint32(&redisLast, 1))),
|
||||
dockertest.WithName(fmt.Sprintf("psrpc-kv-%d", atomic.AddUint32(&kvLast, 1))),
|
||||
)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
@@ -34,9 +34,9 @@ func NewRedis(t testing.TB, pool dockertest.Pool) Server {
|
||||
addr := c.GetHostPort("6379/tcp")
|
||||
waitTCPPort(t, pool, addr)
|
||||
|
||||
t.Log("Redis running on", addr)
|
||||
t.Log("KV running on", addr)
|
||||
|
||||
s := &redisServer{addr: addr}
|
||||
s := &kvServer{addr: addr}
|
||||
|
||||
err = pool.Retry(ctx, 0, func() error {
|
||||
rc, err := s.connect()
|
||||
@@ -53,12 +53,12 @@ func NewRedis(t testing.TB, pool dockertest.Pool) Server {
|
||||
return s
|
||||
}
|
||||
|
||||
type redisServer struct {
|
||||
type kvServer struct {
|
||||
addr string
|
||||
}
|
||||
|
||||
func (s *redisServer) connect() (redis.UniversalClient, error) {
|
||||
rc := redis.NewUniversalClient(&redis.UniversalOptions{Addrs: []string{s.addr}})
|
||||
func (s *kvServer) connect() (kv.UniversalClient, error) {
|
||||
rc := kv.NewUniversalClient(&kv.UniversalOptions{Addrs: []string{s.addr}})
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), time.Second)
|
||||
defer cancel()
|
||||
@@ -71,10 +71,10 @@ func (s *redisServer) connect() (redis.UniversalClient, error) {
|
||||
return rc, nil
|
||||
}
|
||||
|
||||
func (s *redisServer) Connect(t testing.TB) bus.MessageBus {
|
||||
func (s *kvServer) Connect(t testing.TB) bus.MessageBus {
|
||||
rc, err := s.connect()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return bus.NewRedisMessageBus(rc)
|
||||
return bus.NewKVMessageBus(rc)
|
||||
}
|
||||
Reference in New Issue
Block a user