one more additional check for shutdown

Author: ndyakov

hitless/pool_hook.go

@@ -327,7 +327,16 @@ func (ph *PoolHook) handoffWorker() {
 				return // Exit this worker
 			}
 
-			ph.processHandoffRequest(request)
+			// Check for shutdown before processing
+			select {
+			case <-ph.shutdown:
+				// Clean up the request before exiting
+				ph.pending.Delete(request.ConnID)
+				return
+			default:
+				// Continue with processing
+				ph.processHandoffRequest(request)
+			}
 		case <-ph.shutdown:
 			return
 		}
@@ -505,7 +514,6 @@ func (ph *PoolHook) performConnectionHandoffWithPool(ctx context.Context, conn *
 		// Keep the handoff state for retry
 		return err
 	}
-	// Note: CLIENT MAINT_NOTIFICATIONS is sent during client initialization, not per connection
 	defer func() {
 		if oldConn != nil {
 			oldConn.Close()
@@ -542,7 +550,9 @@ func (ph *PoolHook) createEndpointDialer(endpoint string) func(context.Context)
 		if err != nil {
 			// If no port specified, assume default Redis port
 			host = endpoint
-			port = "6379"
+			if port == "" {
+				port = "6379"
+			}
 		}
 
 		// Use the base dialer to connect to the new endpoint

pubsub.go

@@ -493,6 +493,7 @@ func (c *PubSub) ReceiveTimeout(ctx context.Context, timeout time.Duration) (int
 // Receive returns a message as a Subscription, Message, Pong or error.
 // See PubSub example for details. This is low-level API and in most cases
 // Channel should be used instead.
+// This will block until a message is received.
 func (c *PubSub) Receive(ctx context.Context) (interface{}, error) {
 	return c.ReceiveTimeout(ctx, 0)
 }
@@ -575,7 +576,8 @@ func (c *PubSub) ChannelWithSubscriptions(opts ...ChannelOption) <-chan interfac
 }
 
 func (c *PubSub) processPendingPushNotificationWithReader(ctx context.Context, cn *pool.Conn, rd *proto.Reader) error {
-	if c.pushProcessor == nil {
+	// Only process push notifications for RESP3 connections with a processor
+	if c.opt.Protocol != 3 || c.pushProcessor == nil {
 		return nil
 	}