[CAE-1088] feat: RESP3 notifications support

internal/pool/conn.go

@@ -77,6 +77,7 @@ func (cn *Conn) WithReader(
 			return err
 		}
 	}
+
 	return fn(cn.rd)
 }
 

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,9 @@ type Options struct {
 	MaxActiveConns  int
 	ConnMaxIdleTime time.Duration
 	ConnMaxLifetime time.Duration
+
+	// Protocol version for optimization (3 = RESP3 with push notifications, 2 = RESP2 without)
+	Protocol int
 }
 
 type lastDialErrorWrap struct {
@@ -228,6 +232,7 @@ func (p *ConnPool) dialConn(ctx context.Context, pooled bool) (*Conn, error) {
 
 	cn := NewConn(netConn)
 	cn.pooled = pooled
+
 	return cn, nil
 }
 
@@ -377,6 +382,16 @@ func (p *ConnPool) popIdle() (*Conn, error) {
 
 func (p *ConnPool) Put(ctx context.Context, cn *Conn) {
 	if cn.rd.Buffered() > 0 {
+		// Check if this might be push notification data
+		if p.cfg.Protocol == 3 {
+			if replyType, err := cn.rd.PeekReplyType(); err == nil && replyType == proto.RespPush {
+				// For push notifications, we allow some buffered data
+				// The client will process these notifications before using the connection
+				internal.Logger.Printf(ctx, "push: connection has buffered data, likely push notifications - will be processed by client")
+				return
+			}
+		}
+		// For non-RESP3 or data that is not a push notification, buffered data is unexpected
 		internal.Logger.Printf(ctx, "Conn has unread data")
 		p.Remove(ctx, cn, BadConnError{})
 		return
@@ -523,8 +538,22 @@ 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, it might be push notifications (RESP3)
+		// However, push notification processing is now handled by the client
+		// before WithReader to ensure proper context is available to handlers
+		if err == errUnexpectedRead && p.cfg.Protocol == 3 {
+			if replyType, err := cn.rd.PeekReplyType(); err == nil && replyType == proto.RespPush {
+				// For RESP3 connections with push notifications, we allow some buffered data
+				// The client will process these notifications before using the connection
+				internal.Logger.Printf(context.Background(), "push: connection has buffered data, likely push notifications - will be processed by client")
+				return true // Connection is healthy, client will handle notifications
+			}
+			return false // Unexpected data, not push notifications, connection is unhealthy
+		} else {
+			return false
+		}
 	}
 
 	cn.SetUsedAt(now)

internal/proto/reader.go

@@ -90,6 +90,27 @@ func (r *Reader) PeekReplyType() (byte, error) {
 	return b[0], nil
 }
 
+func (r *Reader) PeekPushNotificationName() (string, error) {
+	// peek 36 bytes, should be enough to read the push notification name
+	buf, err := r.rd.Peek(36)
+	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) {

options.go

@@ -15,6 +15,7 @@ import (
 
 	"github.com/redis/go-redis/v9/auth"
 	"github.com/redis/go-redis/v9/internal/pool"
+	"github.com/redis/go-redis/v9/push"
 )
 
 // Limiter is the interface of a rate limiter or a circuit breaker.
@@ -216,6 +217,13 @@ 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
+
+	// Push notifications are always enabled for RESP3 connections (Protocol: 3)
+	// and are not available for RESP2 connections. No configuration option is needed.
+
+	// PushNotificationProcessor is the processor for handling push notifications.
+	// If nil, a default processor will be created for RESP3 connections.
+	PushNotificationProcessor push.NotificationProcessor
 }
 
 func (opt *Options) init() {
@@ -592,5 +600,7 @@ func newConnPool(
 		MaxActiveConns:  opt.MaxActiveConns,
 		ConnMaxIdleTime: opt.ConnMaxIdleTime,
 		ConnMaxLifetime: opt.ConnMaxLifetime,
+		// Pass protocol version for push notification optimization
+		Protocol: opt.Protocol,
 	})
 }

osscluster.go

@@ -1623,7 +1623,7 @@ func (c *ClusterClient) processTxPipelineNode(
 }
 
 func (c *ClusterClient) processTxPipelineNodeConn(
-	ctx context.Context, _ *clusterNode, cn *pool.Conn, cmds []Cmder, failedCmds *cmdsMap,
+	ctx context.Context, node *clusterNode, cn *pool.Conn, cmds []Cmder, failedCmds *cmdsMap,
 ) error {
 	if err := cn.WithWriter(c.context(ctx), c.opt.WriteTimeout, func(wr *proto.Writer) error {
 		return writeCmds(wr, cmds)
@@ -1641,7 +1641,7 @@ func (c *ClusterClient) processTxPipelineNodeConn(
 		trimmedCmds := cmds[1 : len(cmds)-1]
 
 		if err := c.txPipelineReadQueued(
-			ctx, rd, statusCmd, trimmedCmds, failedCmds,
+			ctx, node, cn, rd, statusCmd, trimmedCmds, failedCmds,
 		); err != nil {
 			setCmdsErr(cmds, err)
 
@@ -1653,30 +1653,50 @@ func (c *ClusterClient) processTxPipelineNodeConn(
 			return err
 		}
 
-		return pipelineReadCmds(rd, trimmedCmds)
+		return node.Client.pipelineReadCmds(ctx, cn, rd, trimmedCmds)
 	})
 }
 
 func (c *ClusterClient) txPipelineReadQueued(
 	ctx context.Context,
+	node *clusterNode,
+	cn *pool.Conn,
 	rd *proto.Reader,
 	statusCmd *StatusCmd,
 	cmds []Cmder,
 	failedCmds *cmdsMap,
 ) error {
 	// Parse queued replies.
+	// To be sure there are no buffered push notifications, we process them before reading the reply
+	if err := node.Client.processPendingPushNotificationWithReader(ctx, cn, rd); err != nil {
+		// Log the error but don't fail the command execution
+		// Push notification processing errors shouldn't break normal Redis operations
+		internal.Logger.Printf(ctx, "push: error processing pending notifications before reading reply: %v", err)
+	}
 	if err := statusCmd.readReply(rd); err != nil {
 		return err
 	}
 
 	for _, cmd := range cmds {
+		// To be sure there are no buffered push notifications, we process them before reading the reply
+		if err := node.Client.processPendingPushNotificationWithReader(ctx, cn, rd); err != nil {
+			// Log the error but don't fail the command execution
+			// Push notification processing errors shouldn't break normal Redis operations
+			internal.Logger.Printf(ctx, "push: error processing pending notifications before reading reply: %v", err)
+		}
 		err := statusCmd.readReply(rd)
 		if err == nil || c.checkMovedErr(ctx, cmd, err, failedCmds) || isRedisError(err) {
 			continue
 		}
 		return err
 	}
 
+	// To be sure there are no buffered push notifications, we process them before reading the reply
+	if err := node.Client.processPendingPushNotificationWithReader(ctx, cn, rd); err != nil {
+		// Log the error but don't fail the command execution
+		// Push notification processing errors shouldn't break normal Redis operations
+		internal.Logger.Printf(ctx, "push: error processing pending notifications before reading reply: %v", err)
+	}
 	// Parse number of replies.
 	line, err := rd.ReadLine()
 	if err != nil {

pubsub.go

@@ -10,6 +10,7 @@
 	"github.com/redis/go-redis/v9/internal"
 	"github.com/redis/go-redis/v9/internal/pool"
 	"github.com/redis/go-redis/v9/internal/proto"
+	"github.com/redis/go-redis/v9/push"
 )
 
 // PubSub implements Pub/Sub commands as described in
@@ -38,6 +39,9 @@
 	chOnce sync.Once
 	msgCh  *channel
 	allCh  *channel
+
+	// Push notification processor for handling generic push notifications
+	pushProcessor push.NotificationProcessor
 }
 
 func (c *PubSub) init() {
@@ -436,6 +440,12 @@
 	}
 
 	err = cn.WithReader(ctx, timeout, func(rd *proto.Reader) error {
+		// To be sure there are no buffered push notifications, we process them before reading the reply
+		if err := c.processPendingPushNotificationWithReader(ctx, cn, rd); err != nil {
+			// Log the error but don't fail the command execution
+			// Push notification processing errors shouldn't break normal Redis operations
+			internal.Logger.Printf(ctx, "push: error processing pending notifications before reading reply: %v", err)
+		}
 		return c.cmd.readReply(rd)
 	})
 
@@ -532,6 +542,23 @@
 	return c.allCh.allCh
 }
 
+func (c *PubSub) processPendingPushNotificationWithReader(ctx context.Context, cn *pool.Conn, rd *proto.Reader) error {
+	if c.pushProcessor == nil {
+		return nil
+	}
+
+	// Create handler context with client, connection pool, and connection information
+	handlerCtx := c.pushNotificationHandlerContext(cn)
+	return c.pushProcessor.ProcessPendingNotifications(ctx, handlerCtx, rd)
+}
+
+func (c *PubSub) pushNotificationHandlerContext(cn *pool.Conn) push.NotificationHandlerContext {
+	// PubSub doesn't have a client or connection pool, so we pass nil for those
+	// PubSub connections are blocking
+	return push.HandlerContext{}
+	return push.NewNotificationHandlerContext(nil, nil, c, cn, true)
+}
+
 type ChannelOption func(c *channel)
 
 // WithChannelSize specifies the Go chan size that is used to buffer incoming messages.