[wip] resp3 notification handlers

internal/pool/conn.go

@@ -25,6 +25,11 @@ type Conn struct {
 	createdAt time.Time
 
 	onClose func() error
+
+	// Push notification processor for handling push notifications on this connection
+	PushNotificationProcessor interface {
+		ProcessPendingNotifications(ctx context.Context, rd *proto.Reader) error
+	}
 }
 
 func NewConn(netConn net.Conn) *Conn {

internal/pool/pool.go

@@ -9,6 +9,7 @@ import (
 	"time"
 
 	"github.com/redis/go-redis/v9/internal"
+	"github.com/redis/go-redis/v9/internal/proto"
 )
 
 var (
@@ -71,6 +72,11 @@ type Options struct {
 	MaxActiveConns  int
 	ConnMaxIdleTime time.Duration
 	ConnMaxLifetime time.Duration
+
+	// Push notification processor for connections
+	PushNotificationProcessor interface {
+		ProcessPendingNotifications(ctx context.Context, rd *proto.Reader) error
+	}
 }
 
 type lastDialErrorWrap struct {
@@ -228,6 +234,12 @@ func (p *ConnPool) dialConn(ctx context.Context, pooled bool) (*Conn, error) {
 
 	cn := NewConn(netConn)
 	cn.pooled = pooled
+
+	// Set push notification processor if available
+	if p.cfg.PushNotificationProcessor != nil {
+		cn.PushNotificationProcessor = p.cfg.PushNotificationProcessor
+	}
+
 	return cn, nil
 }
 
@@ -377,9 +389,24 @@ func (p *ConnPool) popIdle() (*Conn, error) {
 
 func (p *ConnPool) Put(ctx context.Context, cn *Conn) {
 	if cn.rd.Buffered() > 0 {
-		internal.Logger.Printf(ctx, "Conn has unread data")
-		p.Remove(ctx, cn, BadConnError{})
-		return
+		// Check if this might be push notification data
+		if cn.PushNotificationProcessor != nil {
+			// Try to process pending push notifications before discarding connection
+			err := cn.PushNotificationProcessor.ProcessPendingNotifications(ctx, cn.rd)
+			if err != nil {
+				internal.Logger.Printf(ctx, "push: error processing pending notifications: %v", err)
+			}
+			// Check again if there's still unread data after processing push notifications
+			if cn.rd.Buffered() > 0 {
+				internal.Logger.Printf(ctx, "Conn has unread data after processing push notifications")
+				p.Remove(ctx, cn, BadConnError{})
+				return
+			}
+		} else {
+			internal.Logger.Printf(ctx, "Conn has unread data")
+			p.Remove(ctx, cn, BadConnError{})
+			return
+		}
 	}
 
 	if !cn.pooled {
@@ -523,8 +550,24 @@ func (p *ConnPool) isHealthyConn(cn *Conn) bool {
 		return false
 	}
 
-	if connCheck(cn.netConn) != nil {
-		return false
+	// Check connection health, but be aware of push notifications
+	if err := connCheck(cn.netConn); err != nil {
+		// If there's unexpected data and we have push notification support,
+		// it might be push notifications
+		if err == errUnexpectedRead && cn.PushNotificationProcessor != nil {
+			// Try to process any pending push notifications
+			ctx := context.Background()
+			if procErr := cn.PushNotificationProcessor.ProcessPendingNotifications(ctx, cn.rd); procErr != nil {
+				internal.Logger.Printf(ctx, "push: error processing pending notifications during health check: %v", procErr)
+				return false
+			}
+			// Check again after processing push notifications
+			if connCheck(cn.netConn) != nil {
+				return false
+			}
+		} else {
+			return false
+		}
 	}
 
 	cn.SetUsedAt(now)

options.go

@@ -216,6 +216,21 @@ type Options struct {
 	// UnstableResp3 enables Unstable mode for Redis Search module with RESP3.
 	// When unstable mode is enabled, the client will use RESP3 protocol and only be able to use RawResult
 	UnstableResp3 bool
+
+	// PushNotifications enables general push notification processing.
+	// When enabled, the client will process RESP3 push notifications and
+	// route them to registered handlers.
+	//
+	// For RESP3 connections (Protocol: 3), push notifications are automatically enabled.
+	// To disable push notifications for RESP3, use Protocol: 2 instead.
+	// For RESP2 connections, push notifications are not available.
+	//
+	// default: automatically enabled for RESP3, disabled for RESP2
+	PushNotifications bool
+
+	// PushNotificationProcessor is the processor for handling push notifications.
+	// If nil, a default processor will be created when PushNotifications is enabled.
+	PushNotificationProcessor PushNotificationProcessorInterface
 }
 
 func (opt *Options) init() {
@@ -592,5 +607,7 @@ func newConnPool(
 		MaxActiveConns:  opt.MaxActiveConns,
 		ConnMaxIdleTime: opt.ConnMaxIdleTime,
 		ConnMaxLifetime: opt.ConnMaxLifetime,
+		// Pass push notification processor for connection initialization
+		PushNotificationProcessor: opt.PushNotificationProcessor,
 	})
 }

pubsub.go

@@ -38,12 +38,21 @@ type PubSub struct {
 	chOnce sync.Once
 	msgCh  *channel
 	allCh  *channel
+
+	// Push notification processor for handling generic push notifications
+	pushProcessor PushNotificationProcessorInterface
 }
 
 func (c *PubSub) init() {
 	c.exit = make(chan struct{})
 }
 
+// SetPushNotificationProcessor sets the push notification processor for handling
+// generic push notifications received on this PubSub connection.
+func (c *PubSub) SetPushNotificationProcessor(processor PushNotificationProcessorInterface) {
+	c.pushProcessor = processor
+}
+
 func (c *PubSub) String() string {
 	c.mu.Lock()
 	defer c.mu.Unlock()
@@ -367,6 +376,18 @@ func (p *Pong) String() string {
 	return "Pong"
 }
 
+// PushNotificationMessage represents a generic push notification received on a PubSub connection.
+type PushNotificationMessage struct {
+	// Command is the push notification command (e.g., "MOVING", "CUSTOM_EVENT").
+	Command string
+	// Args are the arguments following the command.
+	Args []interface{}
+}
+
+func (m *PushNotificationMessage) String() string {
+	return fmt.Sprintf("push: %s", m.Command)
+}
+
 func (c *PubSub) newMessage(reply interface{}) (interface{}, error) {
 	switch reply := reply.(type) {
 	case string:
@@ -413,6 +434,19 @@ func (c *PubSub) newMessage(reply interface{}) (interface{}, error) {
 				Payload: reply[1].(string),
 			}, nil
 		default:
+			// Try to handle as generic push notification
+			ctx := c.getContext()
+			registry := c.pushProcessor.GetRegistry()
+			if registry != nil {
+				handled := registry.HandleNotification(ctx, reply)
+				if handled {
+					// Return a special message type to indicate it was handled
+					return &PushNotificationMessage{
+						Command: kind,
+						Args:    reply[1:],
+					}, nil
+				}
+			}
 			return nil, fmt.Errorf("redis: unsupported pubsub message: %q", kind)
 		}
 	default:
@@ -658,6 +692,9 @@ func (c *channel) initMsgChan() {
 				// Ignore.
 			case *Pong:
 				// Ignore.
+			case *PushNotificationMessage:
+				// Ignore push notifications in message-only channel
+				// They are already handled by the push notification processor
 			case *Message:
 				timer.Reset(c.chanSendTimeout)
 				select {
@@ -712,7 +749,7 @@ func (c *channel) initAllChan() {
 			switch msg := msg.(type) {
 			case *Pong:
 				// Ignore.
-			case *Subscription, *Message:
+			case *Subscription, *Message, *PushNotificationMessage:
 				timer.Reset(c.chanSendTimeout)
 				select {
 				case c.allCh <- msg: