fix pubsub redis close bug
When component close function called `r.cancel()` and `r.client.Close()` will be excuted.
The pollNewMessagesLoop loop use client to read from redis which will lead to err. And when err occured this loop will continue.
We need to check context on the top of this loop.
@@ -279,6 +279,11 @@ func (r *redisStreams) processMessage(msg redisMessageWrapper) error {
// by calling `enqueueMessages`.
func (r *redisStreams) pollNewMessagesLoop(stream string, handler func(msg *pubsub.NewMessage) error) {
for {
+ // Return on cancelation
+ if r.ctx.Err() != nil {
+ return
+ }
+
// Read messages
streams, err := r.client.XReadGroup(&redis.XReadGroupArgs{
Group: r.metadata.consumerID,
@@ -297,11 +302,6 @@ func (r *redisStreams) pollNewMessagesLoop(stream string, handler func(msg *pubs
for _, s := range streams {
r.enqueueMessages(s.Stream, handler, s.Messages)
}
-
- // Return on cancelation
- if r.ctx.Err() != nil {
- return
- }
}
}
