wip

Author: ndyakov

hitless/README.md

@@ -2,7 +2,6 @@ package hitless
 
 import (
 	"context"
-	"fmt"
 	"strings"
 	"time"
 
@@ -202,62 +201,3 @@ func (vh *ValidationHook) PreHook(ctx context.Context, notificationType string,
 func (vh *ValidationHook) PostHook(ctx context.Context, notificationType string, notification []interface{}, result error) {
 	// No post-processing needed
 }
-
-// ExampleUsage demonstrates how to use the hooks and notification types with HitlessManager.
-func ExampleUsage() {
-	// This is just an example - in real usage, you'd have actual client and config
-	fmt.Println("Example of using hooks and notification types with HitlessManager:")
-	fmt.Println()
-
-	fmt.Println("1. Using notification type constants:")
-	fmt.Printf("   MOVING: %s\n", NotificationMoving)
-	fmt.Printf("   MIGRATING: %s\n", NotificationMigrating)
-	fmt.Printf("   MIGRATED: %s\n", NotificationMigrated)
-	fmt.Printf("   FAILING_OVER: %s\n", NotificationFailingOver)
-	fmt.Printf("   FAILED_OVER: %s\n", NotificationFailedOver)
-	fmt.Println()
-
-	fmt.Println("2. Using notification type sets:")
-	fmt.Println("   // Register handlers for all notification types")
-	fmt.Println("   manager.RegisterSelectiveHandlers(AllNotificationTypes())")
-	fmt.Println()
-	fmt.Println("   // Register handlers only for MOVING notifications")
-	fmt.Println("   manager.RegisterSelectiveHandlers(MovingOnlyNotifications())")
-	fmt.Println()
-	fmt.Println("   // Register handlers only for migration-related notifications")
-	fmt.Println("   manager.RegisterSelectiveHandlers(MigrationNotifications())")
-	fmt.Println()
-	fmt.Println("   // Register handlers only for failover-related notifications")
-	fmt.Println("   manager.RegisterSelectiveHandlers(FailoverNotifications())")
-	fmt.Println()
-	fmt.Println("   // Register handlers for custom set of notifications")
-	fmt.Println("   customTypes := NewNotificationTypeSet(NotificationMoving, NotificationMigrated)")
-	fmt.Println("   manager.RegisterSelectiveHandlers(customTypes)")
-	fmt.Println()
-
-	fmt.Println("3. Create hooks:")
-	fmt.Println("   metricsHook := NewMetricsHook()")
-	fmt.Println("   rewriteHook := NewEndpointRewriteHook(map[string]string{")
-	fmt.Println("       \"old-redis:6379\": \"new-redis:6379\",")
-	fmt.Println("   })")
-	fmt.Println("   throttleHook := NewThrottleHook(10) // 10 notifications per second")
-	fmt.Println("   validationHook := NewValidationHook(true) // strict mode")
-	fmt.Println()
-
-	fmt.Println("4. Add hooks to manager:")
-	fmt.Println("   manager.AddHook(validationHook)  // Validate first")
-	fmt.Println("   manager.AddHook(throttleHook)    // Then throttle")
-	fmt.Println("   manager.AddHook(rewriteHook)     // Then rewrite")
-	fmt.Println("   manager.AddHook(metricsHook)     // Finally collect metrics")
-	fmt.Println()
-
-	fmt.Println("5. Hooks will be called in order for each notification:")
-	fmt.Println("   - ValidationHook validates the notification")
-	fmt.Println("   - ThrottleHook may skip processing if rate limit exceeded")
-	fmt.Println("   - EndpointRewriteHook may modify the endpoint")
-	fmt.Println("   - MetricsHook collects processing statistics")
-	fmt.Println()
-
-	fmt.Println("6. Remove hooks when no longer needed:")
-	fmt.Println("   manager.RemoveHook(throttleHook)")
-}

hitless/example_config_usage.go

@@ -271,47 +271,6 @@ func (hm *HitlessManager) processPostHooks(ctx context.Context, notificationType
 	}
 }
 
-// LoggingHook is an example hook implementation that logs all notifications.
-type LoggingHook struct {
-	LogLevel int
-}
-
-// PreHook logs the notification before processing and allows modification.
-func (lh *LoggingHook) PreHook(ctx context.Context, notificationType string, notification []interface{}) ([]interface{}, bool) {
-	if lh.LogLevel >= 2 { // Info level
-		internal.Logger.Printf(ctx, "hitless: processing %s notification: %v", notificationType, notification)
-	}
-	return notification, true // Continue processing with unmodified notification
-}
-
-// PostHook logs the result after processing.
-func (lh *LoggingHook) PostHook(ctx context.Context, notificationType string, notification []interface{}, result error) {
-	if result != nil && lh.LogLevel >= 1 { // Warning level
-		internal.Logger.Printf(ctx, "hitless: %s notification processing failed: %v", notificationType, result)
-	} else if lh.LogLevel >= 3 { // Debug level
-		internal.Logger.Printf(ctx, "hitless: %s notification processed successfully", notificationType)
-	}
-}
-
-// FilterHook is an example hook that can filter out certain notifications.
-type FilterHook struct {
-	BlockedTypes map[string]bool
-}
-
-// PreHook filters notifications based on type.
-func (fh *FilterHook) PreHook(ctx context.Context, notificationType string, notification []interface{}) ([]interface{}, bool) {
-	if fh.BlockedTypes[notificationType] {
-		internal.Logger.Printf(ctx, "hitless: filtering out %s notification", notificationType)
-		return notification, false // Skip processing
-	}
-	return notification, true
-}
-
-// PostHook does nothing for filter hook.
-func (fh *FilterHook) PostHook(ctx context.Context, notificationType string, notification []interface{}, result error) {
-	// No post-processing needed for filter hook
-}
-
 // createPoolHook creates a pool hook with this manager already set.
 func (hm *HitlessManager) createPoolHook(baseDialer func(context.Context, string, string) (net.Conn, error)) *PoolHook {
 	if hm.poolHooksRef != nil {

hitless/example_hooks.go

@@ -0,0 +1,48 @@
+package hitless
+
+import (
+	"context"
+
+	"github.com/redis/go-redis/v9/internal"
+)
+
+// LoggingHook is an example hook implementation that logs all notifications.
+type LoggingHook struct {
+	LogLevel int
+}
+
+// PreHook logs the notification before processing and allows modification.
+func (lh *LoggingHook) PreHook(ctx context.Context, notificationType string, notification []interface{}) ([]interface{}, bool) {
+	if lh.LogLevel >= 2 { // Info level
+		internal.Logger.Printf(ctx, "hitless: processing %s notification: %v", notificationType, notification)
+	}
+	return notification, true // Continue processing with unmodified notification
+}
+
+// PostHook logs the result after processing.
+func (lh *LoggingHook) PostHook(ctx context.Context, notificationType string, notification []interface{}, result error) {
+	if result != nil && lh.LogLevel >= 1 { // Warning level
+		internal.Logger.Printf(ctx, "hitless: %s notification processing failed: %v", notificationType, result)
+	} else if lh.LogLevel >= 3 { // Debug level
+		internal.Logger.Printf(ctx, "hitless: %s notification processed successfully", notificationType)
+	}
+}
+
+// FilterHook is an example hook that can filter out certain notifications.
+type FilterHook struct {
+	BlockedTypes map[string]bool
+}
+
+// PreHook filters notifications based on type.
+func (fh *FilterHook) PreHook(ctx context.Context, notificationType string, notification []interface{}) ([]interface{}, bool) {
+	if fh.BlockedTypes[notificationType] {
+		internal.Logger.Printf(ctx, "hitless: filtering out %s notification", notificationType)
+		return notification, false // Skip processing
+	}
+	return notification, true
+}
+
+// PostHook does nothing for filter hook.
+func (fh *FilterHook) PostHook(ctx context.Context, notificationType string, notification []interface{}, result error) {
+	// No post-processing needed for filter hook
+}

hitless/hitless_manager.go

@@ -138,6 +138,7 @@ func (ph *PoolHook) OnGet(ctx context.Context, conn *pool.Conn, _ bool) error {
 	// in a handoff state at the moment.
 
 	// Check if connection is usable (not in a handoff state)
+	// Should not happen since the pool will not return a connection that is not usable.
 	if !conn.IsUsable() {
 		return ErrConnectionMarkedForHandoff
 	}

hitless/hooks.go

@@ -45,7 +45,6 @@ type Conn struct {
 
 	Inited    bool
 	pooled    bool
-	pubsub    bool
 	createdAt time.Time
 	expiresAt time.Time
 
@@ -453,18 +452,6 @@ func (cn *Conn) IncrementAndGetHandoffRetries(n int) int {
 	return cn.incrementHandoffRetries(n)
 }
 
-// Rd returns the connection's reader for protocol-specific processing
-func (cn *Conn) Rd() *proto.Reader {
-	return cn.rd
-}
-
-// Reader returns the connection's proto reader for processing notifications
-// Note: This method should be used carefully as it returns the raw reader.
-// For thread-safe operations, use HasBufferedData() and PeekReplyTypeSafe().
-func (cn *Conn) Reader() *proto.Reader {
-	return cn.rd
-}
-
 // HasBufferedData safely checks if the connection has buffered data.
 // This method is used to avoid data races when checking for push notifications.
 func (cn *Conn) HasBufferedData() bool {
@@ -607,11 +594,3 @@ func (cn *Conn) deadline(ctx context.Context, timeout time.Duration) time.Time {
 
 	return noDeadline
 }
-
-func (cn *Conn) IsPubSub() bool {
-	return cn.pubsub
-}
-
-func (cn *Conn) IsPooled() bool {
-	return cn.pooled
-}

hitless/pool_hook.go

@@ -24,6 +24,8 @@ var (
 	// ErrPoolTimeout timed out waiting to get a connection from the connection pool.
 	ErrPoolTimeout = errors.New("redis: connection pool timeout")
 
+	popAttempts  = 10
+	getAttempts  = 3
 	minTime      = time.Unix(-2208988800, 0) // Jan 1, 1900
 	maxTime      = minTime.Add(1<<63 - 1)
 	noExpiration = maxTime
@@ -356,14 +358,16 @@ func (p *ConnPool) getConn(ctx context.Context) (*Conn, error) {
 		return nil, err
 	}
 
-	tries := 0
 	now := time.Now()
+	attempts := 0
 	for {
-		if tries > 10 {
-			log.Printf("redis: connection pool: failed to get a connection after %d tries", tries)
+
+		if attempts >= getAttempts {
+			log.Printf("redis: connection pool: failed to get an connection accepted by hook after %d attempts", attempts)
 			break
 		}
-		tries++
+		attempts++
+
 		p.connsMu.Lock()
 		cn, err = p.popIdle()
 		p.connsMu.Unlock()
@@ -407,6 +411,7 @@ func (p *ConnPool) getConn(ctx context.Context) (*Conn, error) {
 	if p.hookManager != nil {
 		if err := p.hookManager.ProcessOnGet(ctx, newcn, true); err != nil {
 			// Failed to process connection, discard it
+			log.Printf("redis: connection pool: failed to process new connection by hook: %v", err)
 			_ = p.CloseConn(newcn)
 			return nil, err
 		}
@@ -467,9 +472,8 @@ func (p *ConnPool) popIdle() (*Conn, error) {
 
 	var cn *Conn
 	attempts := 0
-	maxAttempts := len(p.idleConns) + 1 // Prevent infinite loop
 
-	for attempts < maxAttempts {
+	for attempts < popAttempts {
 		if len(p.idleConns) == 0 {
 			return nil, nil
 		}
@@ -503,7 +507,8 @@ func (p *ConnPool) popIdle() (*Conn, error) {
 	}
 
 	// If we exhausted all attempts without finding a usable connection, return nil
-	if attempts >= maxAttempts {
+	if attempts >= popAttempts {
+		log.Printf("redis: connection pool: failed to get an usable connection after %d attempts", popAttempts)
 		return nil, nil
 	}