-
Notifications
You must be signed in to change notification settings - Fork 162
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
refactor Redis publish-subscribe connector (#79)
- Loading branch information
Showing
7 changed files
with
140 additions
and
130 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,14 +1,10 @@ | ||
| github.com/bsm/ginkgo/v2 v2.5.0 h1:aOAnND1T40wEdAtkGSkvSICWeQ8L3UASX7YVCqQx+eQ= | ||
| github.com/bsm/gomega v1.20.0 h1:JhAwLmtRzXFTx2AkALSLa8ijZafntmhSoU63Ok18Uq8= | ||
| github.com/bsm/ginkgo/v2 v2.7.0 h1:ItPMPH90RbmZJt5GtkcNvIRuGEdwlBItdNVoyzaNQao= | ||
| github.com/bsm/gomega v1.26.0 h1:LhQm+AFcgV2M0WyKroMASzAzCAJVpAxQXv4SaI9a69Y= | ||
| 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/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= | ||
| 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/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= | ||
| github.com/redis/go-redis/v9 v9.0.2 h1:BA426Zqe/7r56kCcvxYLWe1mkaz71LKF77GwgFzSxfE= | ||
| github.com/redis/go-redis/v9 v9.0.2/go.mod h1:/xDTe9EF1LM61hek62Poq2nzQSGj0xSrEtEHbBQevps= | ||
| github.com/redis/go-redis/v9 v9.0.5 h1:CuQcn5HIEeK7BgElubPP8CGtE0KakrnbBSTLjathl5o= | ||
| github.com/redis/go-redis/v9 v9.0.5/go.mod h1:WqMKv5vnQbRuZstUwxQI195wHy+t4PuXDOjzMvcuQHk= | ||
| github.com/reugn/go-streams v0.9.0 h1:OSOBi8V4B8wb66g0+/0RyA8BkyMcxIj+b1ERVI6l/eQ= | ||
| github.com/reugn/go-streams v0.9.0/go.mod h1:QI5XXifJkVJl2jQ6Cra8I9DvWdJTgqcFYR7amvXZ9Lg= | ||
| github.com/stretchr/testify v1.8.1 h1:w7B6lhMri9wdJUVmEZPGGhZzrYTPvgJArz7wNPgYKsk= | ||
| gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= |
This file was deleted.
Oops, something went wrong.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,126 @@ | ||
| package redis | ||
|
|
||
| import ( | ||
| "context" | ||
| "log" | ||
|
|
||
| "github.com/redis/go-redis/v9" | ||
| "github.com/reugn/go-streams" | ||
| "github.com/reugn/go-streams/flow" | ||
| ) | ||
|
|
||
| // PubSubSource represents a Redis Pub/Sub source connector. | ||
| // | ||
| // In the Publish/Subscribe messaging paradigm senders (publishers) | ||
| // are not programmed to send their messages to specific receivers (subscribers). | ||
| // Rather, published messages are characterized into channels, without knowledge | ||
| // of what (if any) subscribers there may be. | ||
| type PubSubSource struct { | ||
| ctx context.Context | ||
| redisClient *redis.Client | ||
| channel string | ||
| out chan interface{} | ||
| } | ||
|
|
||
| // NewPubSubSource returns a new PubSubSource instance. | ||
| // | ||
| // The given redisClient is subscribed to the provided channel. | ||
| // The replies to subscription and unsubscribing operations are sent in the form of messages | ||
| // so that the client reads a coherent stream of messages where the first element | ||
| // indicates the type of message. | ||
| func NewPubSubSource(ctx context.Context, redisClient *redis.Client, channel string) (*PubSubSource, error) { | ||
| pubsub := redisClient.Subscribe(ctx, channel) | ||
|
|
||
| // Wait for a confirmation that subscription is created before publishing anything | ||
| _, err := pubsub.Receive(ctx) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| source := &PubSubSource{ | ||
| ctx: ctx, | ||
| redisClient: redisClient, | ||
| channel: channel, | ||
| out: make(chan interface{}), | ||
| } | ||
|
|
||
| go source.init(pubsub.Channel()) | ||
| return source, nil | ||
| } | ||
|
|
||
| // init starts the main loop | ||
| func (ps *PubSubSource) init(ch <-chan *redis.Message) { | ||
| loop: | ||
| for { | ||
| select { | ||
| case <-ps.ctx.Done(): | ||
| break loop | ||
|
|
||
| case msg := <-ch: | ||
| ps.out <- msg | ||
| } | ||
| } | ||
|
|
||
| log.Printf("Closing Redis Pub/Sub consumer") | ||
| close(ps.out) | ||
| ps.redisClient.Close() | ||
| } | ||
|
|
||
| // Via streams data through the given flow | ||
| func (ps *PubSubSource) Via(_flow streams.Flow) streams.Flow { | ||
| flow.DoStream(ps, _flow) | ||
| return _flow | ||
| } | ||
|
|
||
| // Out returns an output channel for sending data | ||
| func (ps *PubSubSource) Out() <-chan interface{} { | ||
| return ps.out | ||
| } | ||
|
|
||
| // PubSubSink represents a Redis Pub/Sub sink connector. | ||
| type PubSubSink struct { | ||
| ctx context.Context | ||
| redisClient *redis.Client | ||
| channel string | ||
| in chan interface{} | ||
| } | ||
|
|
||
| // NewPubSubSink returns a new PubSubSink instance. | ||
| // | ||
| // The incoming messages will be published to the given target channel using the | ||
| // provided redis.Client. | ||
| func NewPubSubSink(ctx context.Context, redisClient *redis.Client, channel string) *PubSubSink { | ||
| sink := &PubSubSink{ | ||
| ctx: ctx, | ||
| redisClient: redisClient, | ||
| channel: channel, | ||
| in: make(chan interface{}), | ||
| } | ||
|
|
||
| go sink.init() | ||
| return sink | ||
| } | ||
|
|
||
| // init starts the main loop | ||
| func (ps *PubSubSink) init() { | ||
| for msg := range ps.in { | ||
| switch m := msg.(type) { | ||
| case string: | ||
| err := ps.redisClient.Publish(ps.ctx, ps.channel, m).Err() | ||
| if err != nil { | ||
| log.Printf("Error in redisClient.Publish: %s", err) | ||
| } | ||
|
|
||
| default: | ||
| log.Printf("Unsupported message type %v", m) | ||
| } | ||
| } | ||
|
|
||
| log.Printf("Closing Redis Pub/Sub producer") | ||
| ps.redisClient.Close() | ||
| } | ||
|
|
||
| // In returns an input channel for receiving data | ||
| func (ps *PubSubSink) In() chan<- interface{} { | ||
| return ps.in | ||
| } |