update example and tests, drop connectionAdapter

Author: ndyakov

.github/copilot-instructions.md

@@ -311,9 +311,50 @@ func (opt *Options) init() {
 
 ```go
 type NotificationProcessor interface {
-    ProcessPushNotification(ctx context.Context, data []byte) error
-    RegisterHandler(notificationType string, handler NotificationHandler) error
-    Close() error
+    RegisterHandler(pushNotificationName string, handler interface{}, protected bool) error
+    UnregisterHandler(pushNotificationName string) error
+    GetHandler(pushNotificationName string) interface{}
+}
+
+type NotificationHandler interface {
+    HandlePushNotification(ctx context.Context, handlerCtx NotificationHandlerContext, notification []interface{}) error
+}
+```
+
+### Notification Hooks
+
+```go
+type NotificationHook interface {
+    PreHook(ctx context.Context, notificationCtx push.NotificationHandlerContext, notificationType string, notification []interface{}) ([]interface{}, bool)
+    PostHook(ctx context.Context, notificationCtx push.NotificationHandlerContext, notificationType string, notification []interface{}, result error)
+}
+
+// NotificationHandlerContext provides context for notification processing
+type NotificationHandlerContext struct {
+    Client     interface{} // Redis client instance
+    Pool       interface{} // Connection pool
+    Conn       interface{} // Specific connection (*pool.Conn)
+    IsBlocking bool        // Whether notification was on blocking connection
+}
+```
+
+### Hook Implementation Pattern
+
+```go
+func (h *CustomHook) PreHook(ctx context.Context, notificationCtx push.NotificationHandlerContext, notificationType string, notification []interface{}) ([]interface{}, bool) {
+    // Access connection information
+    if conn, ok := notificationCtx.Conn.(*pool.Conn); ok {
+        connID := conn.GetID()
+        // Process with connection context
+    }
+    return notification, true // Continue processing
+}
+
+func (h *CustomHook) PostHook(ctx context.Context, notificationCtx push.NotificationHandlerContext, notificationType string, notification []interface{}, result error) {
+    // Handle processing result
+    if result != nil {
+        // Log or handle error
+    }
 }
 ```
 

adapters.go

@@ -7,7 +7,6 @@ import (
 	"time"
 
 	"github.com/redis/go-redis/v9/internal/interfaces"
-	"github.com/redis/go-redis/v9/internal/pool"
 	"github.com/redis/go-redis/v9/push"
 )
 
@@ -88,43 +87,6 @@ func (oa *optionsAdapter) NewDialer() func(context.Context) (net.Conn, error) {
 	}
 }
 
-// connectionAdapter adapts a Redis connection to interfaces.ConnectionWithRelaxedTimeout
-type connectionAdapter struct {
-	conn *pool.Conn
-}
-
-// Close closes the connection.
-func (ca *connectionAdapter) Close() error {
-	return ca.conn.Close()
-}
-
-// IsUsable returns true if the connection is safe to use for new commands.
-func (ca *connectionAdapter) IsUsable() bool {
-	return ca.conn.IsUsable()
-}
-
-// GetPoolConn returns the underlying pool connection.
-func (ca *connectionAdapter) GetPoolConn() *pool.Conn {
-	return ca.conn
-}
-
-// SetRelaxedTimeout sets relaxed timeouts for this connection during hitless upgrades.
-// These timeouts remain active until explicitly cleared.
-func (ca *connectionAdapter) SetRelaxedTimeout(readTimeout, writeTimeout time.Duration) {
-	ca.conn.SetRelaxedTimeout(readTimeout, writeTimeout)
-}
-
-// SetRelaxedTimeoutWithDeadline sets relaxed timeouts with an expiration deadline.
-// After the deadline, timeouts automatically revert to normal values.
-func (ca *connectionAdapter) SetRelaxedTimeoutWithDeadline(readTimeout, writeTimeout time.Duration, deadline time.Time) {
-	ca.conn.SetRelaxedTimeoutWithDeadline(readTimeout, writeTimeout, deadline)
-}
-
-// ClearRelaxedTimeout clears relaxed timeouts for this connection.
-func (ca *connectionAdapter) ClearRelaxedTimeout() {
-	ca.conn.ClearRelaxedTimeout()
-}
-
 // pushProcessorAdapter adapts a push.NotificationProcessor to implement interfaces.NotificationProcessor.
 type pushProcessorAdapter struct {
 	processor push.NotificationProcessor

example/pubsub/main.go

@@ -5,26 +5,37 @@ import (
 	"fmt"
 	"log"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/redis/go-redis/v9"
 	"github.com/redis/go-redis/v9/hitless"
 )
 
 var ctx = context.Background()
+var cntErrors atomic.Int64
+var cntSuccess atomic.Int64
+var startTime = time.Now()
 
 // This example is not supposed to be run as is. It is just a test to see how pubsub behaves in relation to pool management.
 // It was used to find regressions in pool management in hitless mode.
 // Please don't use it as a reference for how to use pubsub.
 func main() {
+	startTime = time.Now()
 	wg := &sync.WaitGroup{}
 	rdb := redis.NewClient(&redis.Options{
-		Addr: ":6379",
+		Addr: "redis-15176.aws-cluster-1213.cto.redislabs.com:15176",
 		HitlessUpgradeConfig: &redis.HitlessUpgradeConfig{
 			Mode: hitless.MaintNotificationsEnabled,
 		},
 	})
 	_ = rdb.FlushDB(ctx).Err()
+	hitlessManager := rdb.GetHitlessManager()
+	if hitlessManager == nil {
+		panic("hitless manager is nil")
+	}
+	loggingHook := hitless.NewLoggingHook(3)
+	hitlessManager.AddNotificationHook(loggingHook)
 
 	go func() {
 		for {
@@ -62,20 +73,23 @@ func main() {
 	subCtx, cancelSubCtx = context.WithCancel(ctx)
 	for i := 0; i < 10; i++ {
 		if err := rdb.Incr(ctx, "publishers").Err(); err != nil {
-			panic(err)
+			fmt.Println("incr error:", err)
+			cntErrors.Add(1)
 		}
 		wg.Add(1)
 		go floodThePool(pubCtx, rdb, wg)
 	}
 
 	for i := 0; i < 500; i++ {
 		if err := rdb.Incr(ctx, "subscribers").Err(); err != nil {
-			panic(err)
+			fmt.Println("incr error:", err)
+			cntErrors.Add(1)
 		}
+
 		wg.Add(1)
 		go subscribe(subCtx, rdb, "test2", i, wg)
 	}
-	time.Sleep(5 * time.Second)
+	time.Sleep(120 * time.Second)
 	fmt.Println("canceling publishers")
 	cancelPublishers()
 	time.Sleep(10 * time.Second)
@@ -95,6 +109,9 @@ func main() {
 	fmt.Printf("if drained = published*subscribers: %d\n", publishedInt*subscribersInt)
 
 	time.Sleep(2 * time.Second)
+	fmt.Println("errors:", cntErrors.Load())
+	fmt.Println("success:", cntSuccess.Load())
+	fmt.Println("time:", time.Since(startTime))
 }
 
 func floodThePool(ctx context.Context, rdb *redis.Client, wg *sync.WaitGroup) {
@@ -107,14 +124,18 @@ func floodThePool(ctx context.Context, rdb *redis.Client, wg *sync.WaitGroup) {
 		}
 		err := rdb.Publish(ctx, "test2", "hello").Err()
 		if err != nil {
-			// noop
-			//log.Println("publish error:", err)
+			if err.Error() != "context canceled" {
+				log.Println("publish error:", err)
+				cntErrors.Add(1)
+			}
 		}
 
 		err = rdb.Incr(ctx, "published").Err()
 		if err != nil {
-			// noop
-			//log.Println("incr error:", err)
+			if err.Error() != "context canceled" {
+				log.Println("incr error:", err)
+				cntErrors.Add(1)
+			}
 		}
 		time.Sleep(10 * time.Nanosecond)
 	}
@@ -137,7 +158,10 @@ func subscribe(ctx context.Context, rdb *redis.Client, topic string, subscriberI
 			case msg := <-recChan:
 				err := rdb.Incr(ctx, "received").Err()
 				if err != nil {
-					log.Println("incr error:", err)
+					if err.Error() != "context canceled" {
+						log.Printf("%s\n", err.Error())
+						cntErrors.Add(1)
+					}
 				}
 				_ = msg // Use the message to avoid unused variable warning
 			}

hitless/README.md

@@ -49,23 +49,62 @@ Config: &hitless.Config{
 - **Auto-calculated**: `10 × MaxWorkers`, capped by pool size
 - **Always capped**: Queue size never exceeds pool size
 
-## Metrics Hook Example
+## Notification Hooks
 
-A metrics collection hook is available in `example_hooks.go` that demonstrates how to monitor hitless upgrade operations:
+Notification hooks allow you to monitor and customize hitless upgrade operations. The `NotificationHook` interface provides pre and post processing hooks:
+
+```go
+type NotificationHook interface {
+    PreHook(ctx context.Context, notificationCtx push.NotificationHandlerContext, notificationType string, notification []interface{}) ([]interface{}, bool)
+    PostHook(ctx context.Context, notificationCtx push.NotificationHandlerContext, notificationType string, notification []interface{}, result error)
+}
+```
+
+### Example: Metrics Collection Hook
+
+A metrics collection hook is available in `example_hooks.go`:
 
 ```go
 import "github.com/redis/go-redis/v9/hitless"
 
 metricsHook := hitless.NewMetricsHook()
-// Use with your monitoring system
+manager.AddNotificationHook(metricsHook)
+
+// Access metrics
+metrics := metricsHook.GetMetrics()
 ```
 
-The metrics hook tracks:
+### Example: Custom Logging Hook
+
+```go
+type CustomHook struct{}
+
+func (h *CustomHook) PreHook(ctx context.Context, notificationCtx push.NotificationHandlerContext, notificationType string, notification []interface{}) ([]interface{}, bool) {
+    // Log notification with connection details
+    if conn, ok := notificationCtx.Conn.(*pool.Conn); ok {
+        log.Printf("Processing %s on connection %d", notificationType, conn.GetID())
+    }
+    return notification, true // Continue processing
+}
+
+func (h *CustomHook) PostHook(ctx context.Context, notificationCtx push.NotificationHandlerContext, notificationType string, notification []interface{}, result error) {
+    if result != nil {
+        log.Printf("Failed to process %s: %v", notificationType, result)
+    }
+}
+```
+
+The notification context provides access to:
+- **Client**: The Redis client instance
+- **Pool**: The connection pool
+- **Conn**: The specific connection that received the notification
+- **IsBlocking**: Whether the notification was received on a blocking connection
+
+Hooks can track:
 - Handoff success/failure rates
-- Handoff duration
-- Queue depth
-- Worker utilization
-- Connection lifecycle events
+- Processing duration
+- Connection-specific metrics
+- Custom business logic
 
 ## Requirements
 

hitless/example_hooks.go

@@ -3,6 +3,10 @@ package hitless
 import (
 	"context"
 	"time"
+
+	"github.com/redis/go-redis/v9/internal"
+	"github.com/redis/go-redis/v9/internal/pool"
+	"github.com/redis/go-redis/v9/push"
 )
 
 // contextKey is a custom type for context keys to avoid collisions
@@ -29,9 +33,14 @@ func NewMetricsHook() *MetricsHook {
 }
 
 // PreHook records the start time for processing metrics.
-func (mh *MetricsHook) PreHook(ctx context.Context, notificationType string, notification []interface{}) ([]interface{}, bool) {
+func (mh *MetricsHook) PreHook(ctx context.Context, notificationCtx push.NotificationHandlerContext, notificationType string, notification []interface{}) ([]interface{}, bool) {
 	mh.NotificationCounts[notificationType]++
 
+	// Log connection information if available
+	if conn, ok := notificationCtx.Conn.(*pool.Conn); ok {
+		internal.Logger.Printf(ctx, "hitless: metrics hook processing %s notification on connection %d", notificationType, conn.GetID())
+	}
+
 	// Store start time in context for duration calculation
 	startTime := time.Now()
 	_ = context.WithValue(ctx, startTimeKey, startTime) // Context not used further
@@ -40,7 +49,7 @@ func (mh *MetricsHook) PreHook(ctx context.Context, notificationType string, not
 }
 
 // PostHook records processing completion and any errors.
-func (mh *MetricsHook) PostHook(ctx context.Context, notificationType string, notification []interface{}, result error) {
+func (mh *MetricsHook) PostHook(ctx context.Context, notificationCtx push.NotificationHandlerContext, notificationType string, notification []interface{}, result error) {
 	// Calculate processing duration
 	if startTime, ok := ctx.Value(startTimeKey).(time.Time); ok {
 		duration := time.Since(startTime)
@@ -50,6 +59,11 @@ func (mh *MetricsHook) PostHook(ctx context.Context, notificationType string, no
 	// Record errors
 	if result != nil {
 		mh.ErrorCounts[notificationType]++
+
+		// Log error details with connection information
+		if conn, ok := notificationCtx.Conn.(*pool.Conn); ok {
+			internal.Logger.Printf(ctx, "hitless: metrics hook recorded error for %s notification on connection %d: %v", notificationType, conn.GetID(), result)
+		}
 	}
 }
 

hitless/hitless_manager.go

@@ -11,6 +11,7 @@ import (
 	"github.com/redis/go-redis/v9/internal"
 	"github.com/redis/go-redis/v9/internal/interfaces"
 	"github.com/redis/go-redis/v9/internal/pool"
+	"github.com/redis/go-redis/v9/push"
 )
 
 // Push notification type constants for hitless upgrades
@@ -35,8 +36,8 @@ var hitlessNotificationTypes = []string{
 // PreHook can modify the notification and return false to skip processing
 // PostHook is called after successful processing
 type NotificationHook interface {
-	PreHook(ctx context.Context, notificationType string, notification []interface{}) ([]interface{}, bool)
-	PostHook(ctx context.Context, notificationType string, notification []interface{}, result error)
+	PreHook(ctx context.Context, notificationCtx push.NotificationHandlerContext, notificationType string, notification []interface{}) ([]interface{}, bool)
+	PostHook(ctx context.Context, notificationCtx push.NotificationHandlerContext, notificationType string, notification []interface{}, result error)
 }
 
 // MovingOperationKey provides a unique key for tracking MOVING operations
@@ -252,14 +253,14 @@ func (hm *HitlessManager) GetState() State {
 }
 
 // processPreHooks calls all pre-hooks and returns the modified notification and whether to continue processing.
-func (hm *HitlessManager) processPreHooks(ctx context.Context, notificationType string, notification []interface{}) ([]interface{}, bool) {
+func (hm *HitlessManager) processPreHooks(ctx context.Context, notificationCtx push.NotificationHandlerContext, notificationType string, notification []interface{}) ([]interface{}, bool) {
 	hm.hooksMu.RLock()
 	defer hm.hooksMu.RUnlock()
 
 	currentNotification := notification
 
 	for _, hook := range hm.hooks {
-		modifiedNotification, shouldContinue := hook.PreHook(ctx, notificationType, currentNotification)
+		modifiedNotification, shouldContinue := hook.PreHook(ctx, notificationCtx, notificationType, currentNotification)
 		if !shouldContinue {
 			return modifiedNotification, false
 		}
@@ -270,12 +271,12 @@ func (hm *HitlessManager) processPreHooks(ctx context.Context, notificationType
 }
 
 // processPostHooks calls all post-hooks with the processing result.
-func (hm *HitlessManager) processPostHooks(ctx context.Context, notificationType string, notification []interface{}, result error) {
+func (hm *HitlessManager) processPostHooks(ctx context.Context, notificationCtx push.NotificationHandlerContext, notificationType string, notification []interface{}, result error) {
 	hm.hooksMu.RLock()
 	defer hm.hooksMu.RUnlock()
 
 	for _, hook := range hm.hooks {
-		hook.PostHook(ctx, notificationType, notification, result)
+		hook.PostHook(ctx, notificationCtx, notificationType, notification, result)
 	}
 }