eosswedenorg/thalos
1
0
Fork 0
mirror of https://github.com/eosswedenorg/thalos synced 2026-07-02 11:43:40 +02:00
Code Releases Activity

transport/redis_pubsub/redis.go rename to publisher.go

Browse source
This commit is contained in:
Henrik Hautakoski 2023-03-07 11:33:53 +01:00
parent b46c070948
commit 19b93495ee
3 changed files with 11 additions and 11 deletions
Download patch file Download diff file Expand all files Collapse all files

2
main.go
View file

@ -222,7 +222,7 @@ func main() {
app.SpawnProccessor( app.SpawnProccessor(
shClient, shClient,
redis_pubsub.New(rdb, redis_pubsub.Namespace{ redis_pubsub.NewPublisher(rdb, redis_pubsub.Namespace{
Prefix: conf.Redis.Prefix, Prefix: conf.Redis.Prefix,
ChainID: chainInfo.ChainID.String(), ChainID: chainInfo.ChainID.String(),
}), }),

10
transport/redis_pubsub/redis.go → transport/redis_pubsub/publisher.go
View file

@ -8,25 +8,25 @@ import (
"github.com/go-redis/redis/v8" "github.com/go-redis/redis/v8"
) )
type RedisPubsub struct { type Publisher struct {
pipeline redis.Pipeliner pipeline redis.Pipeliner
ctx context.Context ctx context.Context
ns Namespace ns Namespace
} }
func New(client *redis.Client, ns Namespace) *RedisPubsub { func NewPublisher(client *redis.Client, ns Namespace) *Publisher {
return &RedisPubsub{ return &Publisher{
pipeline: client.Pipeline(), pipeline: client.Pipeline(),
ctx: client.Context(), ctx: client.Context(),
ns: ns, ns: ns,
} }
} }
func (r *RedisPubsub) Publish(channel transport.ChannelInterface, payload []byte) error { func (r *Publisher) Publish(channel transport.ChannelInterface, payload []byte) error {
return r.pipeline.Publish(r.ctx, r.ns.NewKey(channel).String(), payload).Err() return r.pipeline.Publish(r.ctx, r.ns.NewKey(channel).String(), payload).Err()
} }
func (r *RedisPubsub) Flush() error { func (r *Publisher) Flush() error {
_, err := r.pipeline.Exec(r.ctx) _, err := r.pipeline.Exec(r.ctx)
return err return err
} }

10
transport/redis_pubsub/redis_test.go → transport/redis_pubsub/publisher_test.go
View file

@ -9,18 +9,18 @@ import (
"github.com/stretchr/testify/assert" "github.com/stretchr/testify/assert"
) )
func TestRedisPubsub(t *testing.T) { func TestPublisher_Publish(t *testing.T) {
client, mock := redismock.NewClientMock() client, mock := redismock.NewClientMock()
pubsub := New(client, Namespace{ChainID: "id"}) pub := NewPublisher(client, Namespace{ChainID: "id"})
mock.MatchExpectationsInOrder(true) mock.MatchExpectationsInOrder(true)
mock.ExpectPublish("ship::id::test", []byte("some string")).SetVal(0) mock.ExpectPublish("ship::id::test", []byte("some string")).SetVal(0)
mock.ExpectPublish("ship::id::test2", []byte("some other string")).SetVal(0) mock.ExpectPublish("ship::id::test2", []byte("some other string")).SetVal(0)
assert.NoError(t, pubsub.Publish(transport.Channel{"test"}, []byte("some string"))) assert.NoError(t, pub.Publish(transport.Channel{"test"}, []byte("some string")))
assert.NoError(t, pubsub.Publish(transport.Channel{"test2"}, []byte("some other string"))) assert.NoError(t, pub.Publish(transport.Channel{"test2"}, []byte("some other string")))
assert.NoError(t, pubsub.Flush()) assert.NoError(t, pub.Flush())
assert.NoError(t, mock.ExpectationsWereMet()) assert.NoError(t, mock.ExpectationsWereMet())
} }