1
0
mirror of https://github.com/redis/go-redis.git synced 2025-07-28 06:42:00 +03:00

feat: add pub/sub message filtering to push notification processor

- Add isPubSubMessage() function to identify pub/sub message types
- Filter out pub/sub messages in ProcessPendingNotifications
- Allow pub/sub system to handle its own messages without interference
- Process only cluster/system push notifications (MOVING, MIGRATING, etc.)
- Add comprehensive test coverage for filtering logic

Pub/sub message types filtered:
- message (regular pub/sub)
- pmessage (pattern pub/sub)
- subscribe/unsubscribe (subscription management)
- psubscribe/punsubscribe (pattern subscription management)
- smessage (sharded pub/sub, Redis 7.0+)

Benefits:
- Clear separation of concerns between pub/sub and push notifications
- Prevents interference between the two messaging systems
- Ensures pub/sub messages reach their intended handlers
- Eliminates message loss due to incorrect interception
- Improved system reliability and performance
- Better resource utilization and message flow

Implementation:
- Efficient O(1) switch statement for message type lookup
- Case-sensitive matching for precise filtering
- Early return to skip unnecessary processing
- Maintains processing of other notifications in same batch
- Applied to all processing points (WithReader, Pool.Put, isHealthyConn)

Test coverage:
- TestIsPubSubMessage - Function correctness and edge cases
- TestPubSubFiltering - End-to-end integration testing
- Mixed message scenarios and handler verification
This commit is contained in:
Nedyalko Dyakov
2025-06-27 23:20:25 +03:00
parent b6e712b41a
commit f66518cf3a
4 changed files with 199 additions and 24 deletions

View File

@ -90,6 +90,27 @@ func (r *Reader) PeekReplyType() (byte, error) {
return b[0], nil
}
func (r *Reader) PeekPushNotificationName() (string, error) {
// peek 32 bytes, should be enough to read the push notification name
buf, err := r.rd.Peek(32)
if err != nil {
return "", err
}
if buf[0] != RespPush {
return "", fmt.Errorf("redis: can't parse push notification: %q", buf)
}
// remove push notification type and length
nextLine := buf[2:]
for i := 1; i < len(buf); i++ {
if buf[i] == '\r' && buf[i+1] == '\n' {
nextLine = buf[i+2:]
break
}
}
// return notification name or error
return r.readStringReply(nextLine)
}
// ReadLine Return a valid reply, it will check the protocol or redis error,
// and discard the attribute type.
func (r *Reader) ReadLine() ([]byte, error) {