Skip to content

Commit 0cea7bc

Browse files
ndyakovndyakov
committed
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
1 parent b6e712b commit 0cea7bc

File tree

4 files changed

+176
-24
lines changed

4 files changed

+176
-24
lines changed

‎internal/proto/reader.go

Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -90,6 +90,27 @@ func (r *Reader) PeekReplyType() (byte, error) {
9090
return b[0], nil
9191
}
9292

93+
func (r *Reader) PeekPushNotificationName() (string, error) {
94+
// peek 32 bytes, should be enough to read the push notification name
95+
buf, err := r.rd.Peek(32)
96+
if err != nil {
97+
return "", err
98+
}
99+
if buf[0] != RespPush {
100+
return "", fmt.Errorf("redis: can't parse push notification: %q", buf)
101+
}
102+
// remove push notification type and length
103+
nextLine := buf[2:]
104+
for i := 1; i < len(buf); i++ {
105+
if buf[i] == '\r' && buf[i+1] == '\n' {
106+
nextLine = buf[i+2:]
107+
break
108+
}
109+
}
110+
// return notification name or error
111+
return r.readStringReply(nextLine)
112+
}
113+
93114
// ReadLine Return a valid reply, it will check the protocol or redis error,
94115
// and discard the attribute type.
95116
func (r *Reader) ReadLine() ([]byte, error) {

‎internal/pushnotif/processor.go

Lines changed: 28 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -38,8 +38,6 @@ func (p *Processor) UnregisterHandler(pushNotificationName string) error {
3838
return p.registry.UnregisterHandler(pushNotificationName)
3939
}
4040

41-
42-
4341
// ProcessPendingNotifications checks for and processes any pending push notifications.
4442
func (p *Processor) ProcessPendingNotifications(ctx context.Context, rd *proto.Reader) error {
4543
// Check for nil reader
@@ -66,6 +64,17 @@ func (p *Processor) ProcessPendingNotifications(ctx context.Context, rd *proto.R
6664
break
6765
}
6866

67+
notificationName, err := rd.PeekPushNotificationName()
68+
if err != nil {
69+
// Error reading - continue to next iteration
70+
break
71+
}
72+
73+
// Skip pub/sub messages - they should be handled by the pub/sub system
74+
if isPubSubMessage(notificationName) {
75+
break
76+
}
77+
6978
// Try to read the push notification
7079
reply, err := rd.ReadReply()
7180
if err != nil {
@@ -94,6 +103,23 @@ func (p *Processor) ProcessPendingNotifications(ctx context.Context, rd *proto.R
94103
return nil
95104
}
96105

106+
// isPubSubMessage checks if a notification type is a pub/sub message that should be ignored
107+
// by the push notification processor and handled by the pub/sub system instead.
108+
func isPubSubMessage(notificationType string) bool {
109+
switch notificationType {
110+
case "message", // Regular pub/sub message
111+
"pmessage", // Pattern pub/sub message
112+
"subscribe", // Subscription confirmation
113+
"unsubscribe", // Unsubscription confirmation
114+
"psubscribe", // Pattern subscription confirmation
115+
"punsubscribe", // Pattern unsubscription confirmation
116+
"smessage": // Sharded pub/sub message (Redis 7.0+)
117+
return true
118+
default:
119+
return false
120+
}
121+
}
122+
97123
// VoidProcessor discards all push notifications without processing them.
98124
type VoidProcessor struct{}
99125

@@ -119,8 +145,6 @@ func (v *VoidProcessor) UnregisterHandler(pushNotificationName string) error {
119145
return fmt.Errorf("cannot unregister push notification handler '%s': push notifications are disabled (using void processor)", pushNotificationName)
120146
}
121147

122-
123-
124148
// ProcessPendingNotifications for VoidProcessor does nothing since push notifications
125149
// are only available in RESP3 and this processor is used when they're disabled.
126150
// This avoids unnecessary buffer scanning overhead.

‎internal/pushnotif/pushnotif_test.go

Lines changed: 125 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@ import (
66
"strings"
77
"testing"
88

9+
"github.com/redis/go-redis/v9/internal"
910
"github.com/redis/go-redis/v9/internal/proto"
1011
)
1112

@@ -40,6 +41,7 @@ func (h *TestHandler) Reset() {
4041
// TestReaderInterface defines the interface needed for testing
4142
type TestReaderInterface interface {
4243
PeekReplyType() (byte, error)
44+
PeekPushNotificationName() (string, error)
4345
ReadReply() (interface{}, error)
4446
}
4547

@@ -119,10 +121,22 @@ func testProcessPendingNotifications(processor *Processor, ctx context.Context,
119121
break
120122
}
121123

124+
notificationName, err := reader.PeekPushNotificationName()
125+
if err != nil {
126+
// Error reading - continue to next iteration
127+
break
128+
}
129+
130+
// Skip pub/sub messages - they should be handled by the pub/sub system
131+
if isPubSubMessage(notificationName) {
132+
break
133+
}
134+
122135
// Read the push notification
123136
reply, err := reader.ReadReply()
124137
if err != nil {
125138
// Error reading - continue to next iteration
139+
internal.Logger.Printf(ctx, "push: error reading push notification: %v", err)
126140
continue
127141
}
128142

@@ -420,7 +434,7 @@ func TestProcessor(t *testing.T) {
420434
// Test with mock reader - push notification with ReadReply error
421435
mockReader = NewMockReader()
422436
mockReader.AddPeekReplyType(proto.RespPush, nil)
423-
mockReader.AddReadReply(nil, io.ErrUnexpectedEOF) // ReadReply fails
437+
mockReader.AddReadReply(nil, io.ErrUnexpectedEOF) // ReadReply fails
424438
mockReader.AddPeekReplyType(proto.RespString, io.EOF) // No more push notifications
425439
err = testProcessPendingNotifications(processor, ctx, mockReader)
426440
if err != nil {
@@ -430,7 +444,7 @@ func TestProcessor(t *testing.T) {
430444
// Test with mock reader - push notification with invalid reply type
431445
mockReader = NewMockReader()
432446
mockReader.AddPeekReplyType(proto.RespPush, nil)
433-
mockReader.AddReadReply("not-a-slice", nil) // Invalid reply type
447+
mockReader.AddReadReply("not-a-slice", nil) // Invalid reply type
434448
mockReader.AddPeekReplyType(proto.RespString, io.EOF) // No more push notifications
435449
err = testProcessPendingNotifications(processor, ctx, mockReader)
436450
if err != nil {
@@ -620,4 +634,112 @@ func TestVoidProcessor(t *testing.T) {
620634
t.Errorf("VoidProcessor ProcessPendingNotifications should never error, got: %v", err)
621635
}
622636
})
623-
}
637+
}
638+
639+
// TestIsPubSubMessage tests the isPubSubMessage function
640+
func TestIsPubSubMessage(t *testing.T) {
641+
t.Run("PubSubMessages", func(t *testing.T) {
642+
pubSubMessages := []string{
643+
"message", // Regular pub/sub message
644+
"pmessage", // Pattern pub/sub message
645+
"subscribe", // Subscription confirmation
646+
"unsubscribe", // Unsubscription confirmation
647+
"psubscribe", // Pattern subscription confirmation
648+
"punsubscribe", // Pattern unsubscription confirmation
649+
"smessage", // Sharded pub/sub message (Redis 7.0+)
650+
}
651+
652+
for _, msgType := range pubSubMessages {
653+
if !isPubSubMessage(msgType) {
654+
t.Errorf("isPubSubMessage(%q) should return true", msgType)
655+
}
656+
}
657+
})
658+
659+
t.Run("NonPubSubMessages", func(t *testing.T) {
660+
nonPubSubMessages := []string{
661+
"MOVING", // Cluster slot migration
662+
"MIGRATING", // Cluster slot migration
663+
"MIGRATED", // Cluster slot migration
664+
"FAILING_OVER", // Cluster failover
665+
"FAILED_OVER", // Cluster failover
666+
"unknown", // Unknown message type
667+
"", // Empty string
668+
"MESSAGE", // Case sensitive - should not match
669+
"PMESSAGE", // Case sensitive - should not match
670+
}
671+
672+
for _, msgType := range nonPubSubMessages {
673+
if isPubSubMessage(msgType) {
674+
t.Errorf("isPubSubMessage(%q) should return false", msgType)
675+
}
676+
}
677+
})
678+
}
679+
680+
// TestPubSubFiltering tests that pub/sub messages are filtered out during processing
681+
func TestPubSubFiltering(t *testing.T) {
682+
t.Run("PubSubMessagesIgnored", func(t *testing.T) {
683+
processor := NewProcessor()
684+
handler := NewTestHandler("test", true)
685+
ctx := context.Background()
686+
687+
// Register a handler for a non-pub/sub notification
688+
err := processor.RegisterHandler("MOVING", handler, false)
689+
if err != nil {
690+
t.Fatalf("Failed to register handler: %v", err)
691+
}
692+
693+
// Test with mock reader - pub/sub message should be ignored
694+
mockReader := NewMockReader()
695+
mockReader.AddPeekReplyType(proto.RespPush, nil)
696+
pubSubNotification := []interface{}{"message", "channel", "data"}
697+
mockReader.AddReadReply(pubSubNotification, nil)
698+
mockReader.AddPeekReplyType(proto.RespString, io.EOF) // No more push notifications
699+
700+
handler.Reset()
701+
err = testProcessPendingNotifications(processor, ctx, mockReader)
702+
if err != nil {
703+
t.Errorf("ProcessPendingNotifications should handle pub/sub messages gracefully, got: %v", err)
704+
}
705+
706+
// Check that handler was NOT called for pub/sub message
707+
handled := handler.GetHandledNotifications()
708+
if len(handled) != 0 {
709+
t.Errorf("Expected 0 handled notifications for pub/sub message, got: %d", len(handled))
710+
}
711+
})
712+
713+
t.Run("NonPubSubMessagesProcessed", func(t *testing.T) {
714+
processor := NewProcessor()
715+
handler := NewTestHandler("test", true)
716+
ctx := context.Background()
717+
718+
// Register a handler for a non-pub/sub notification
719+
err := processor.RegisterHandler("MOVING", handler, false)
720+
if err != nil {
721+
t.Fatalf("Failed to register handler: %v", err)
722+
}
723+
724+
// Test with mock reader - non-pub/sub message should be processed
725+
mockReader := NewMockReader()
726+
mockReader.AddPeekReplyType(proto.RespPush, nil)
727+
clusterNotification := []interface{}{"MOVING", "slot", "12345"}
728+
mockReader.AddReadReply(clusterNotification, nil)
729+
mockReader.AddPeekReplyType(proto.RespString, io.EOF) // No more push notifications
730+
731+
handler.Reset()
732+
err = testProcessPendingNotifications(processor, ctx, mockReader)
733+
if err != nil {
734+
t.Errorf("ProcessPendingNotifications should handle cluster notifications, got: %v", err)
735+
}
736+
737+
// Check that handler WAS called for cluster notification
738+
handled := handler.GetHandledNotifications()
739+
if len(handled) != 1 {
740+
t.Errorf("Expected 1 handled notification for cluster message, got: %d", len(handled))
741+
} else if len(handled[0]) != 3 || handled[0][0] != "MOVING" {
742+
t.Errorf("Expected MOVING notification, got: %v", handled[0])
743+
}
744+
})
745+
}

‎push_notifications.go

Lines changed: 2 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -39,20 +39,14 @@ func (r *PushNotificationRegistry) UnregisterHandler(pushNotificationName string
3939

4040
// GetHandler returns the handler for a specific push notification name.
4141
func (r *PushNotificationRegistry) GetHandler(pushNotificationName string) PushNotificationHandler {
42-
handler := r.registry.GetHandler(pushNotificationName)
43-
if handler == nil {
44-
return nil
45-
}
46-
return handler
42+
return r.registry.GetHandler(pushNotificationName)
4743
}
4844

4945
// GetRegisteredPushNotificationNames returns a list of all registered push notification names.
5046
func (r *PushNotificationRegistry) GetRegisteredPushNotificationNames() []string {
5147
return r.registry.GetRegisteredPushNotificationNames()
5248
}
5349

54-
55-
5650
// PushNotificationProcessor handles push notifications with a registry of handlers.
5751
type PushNotificationProcessor struct {
5852
processor *pushnotif.Processor
@@ -67,12 +61,7 @@ func NewPushNotificationProcessor() *PushNotificationProcessor {
6761

6862
// GetHandler returns the handler for a specific push notification name.
6963
func (p *PushNotificationProcessor) GetHandler(pushNotificationName string) PushNotificationHandler {
70-
handler := p.processor.GetHandler(pushNotificationName)
71-
if handler == nil {
72-
return nil
73-
}
74-
// The handler is already a PushNotificationHandler since we store it directly
75-
return handler.(PushNotificationHandler)
64+
return p.processor.GetHandler(pushNotificationName)
7665
}
7766

7867
// RegisterHandler registers a handler for a specific push notification name.
@@ -90,8 +79,6 @@ func (p *PushNotificationProcessor) ProcessPendingNotifications(ctx context.Cont
9079
return p.processor.ProcessPendingNotifications(ctx, rd)
9180
}
9281

93-
94-
9582
// VoidPushNotificationProcessor discards all push notifications without processing them.
9683
type VoidPushNotificationProcessor struct {
9784
processor *pushnotif.VoidProcessor
@@ -119,8 +106,6 @@ func (v *VoidPushNotificationProcessor) ProcessPendingNotifications(ctx context.
119106
return v.processor.ProcessPendingNotifications(ctx, rd)
120107
}
121108

122-
123-
124109
// Redis Cluster push notification names
125110
const (
126111
PushNotificationMoving = "MOVING"

0 commit comments

Comments
 (0)