wip

Author: ndyakov

adapters.go

@@ -17,8 +17,8 @@ var ErrInvalidCommand = errors.New("invalid command type")
 // ErrInvalidPool is returned when the pool type is not supported.
 var ErrInvalidPool = errors.New("invalid pool type")
 
-// NewClientAdapter creates a new client adapter for regular Redis clients.
-func NewClientAdapter(client *baseClient) interfaces.ClientInterface {
+// newClientAdapter creates a new client adapter for regular Redis clients.
+func newClientAdapter(client *baseClient) interfaces.ClientInterface {
 	return &clientAdapter{client: client}
 }
 

async_handoff_integration_test.go

@@ -41,7 +41,7 @@ func TestEventDrivenHandoffIntegration(t *testing.T) {
 		}
 
 		// Create processor with event-driven handoff support
-		processor := hitless.NewPoolHook(3, baseDialer, nil, nil)
+		processor := hitless.NewPoolHook(baseDialer, nil, nil)
 		defer processor.Shutdown(context.Background())
 
 		// Create a test pool with hooks
@@ -52,10 +52,12 @@ func TestEventDrivenHandoffIntegration(t *testing.T) {
 			Dialer: func(ctx context.Context) (net.Conn, error) {
 				return &mockNetConn{addr: "original:6379"}, nil
 			},
-			PoolHooks: hookManager,
-			PoolSize:       5,
-			PoolTimeout:    time.Second,
+			PoolSize:    5,
+			PoolTimeout: time.Second,
 		})
+
+		// Add the hook to the pool after creation
+		testPool.AddPoolHook(processor)
 		defer testPool.Close()
 
 		// Set the pool reference in the processor for connection removal on handoff failure
@@ -131,7 +133,7 @@ func TestEventDrivenHandoffIntegration(t *testing.T) {
 			return &mockNetConn{addr: addr}, nil
 		}
 
-		processor := hitless.NewPoolHook(3, baseDialer, nil, nil)
+		processor := hitless.NewPoolHook(baseDialer, nil, nil)
 		defer processor.Shutdown(context.Background())
 
 		// Create hooks manager and add processor as hook
@@ -142,12 +144,15 @@ func TestEventDrivenHandoffIntegration(t *testing.T) {
 			Dialer: func(ctx context.Context) (net.Conn, error) {
 				return &mockNetConn{addr: "original:6379"}, nil
 			},
-			PoolHooks: hookManager,
+			
 			PoolSize:       10,
 			PoolTimeout:    time.Second,
 		})
 		defer testPool.Close()
 
+		// Add the hook to the pool after creation
+		testPool.AddPoolHook(processor)
+
 		// Set the pool reference in the processor
 		processor.SetPool(testPool)
 
@@ -200,7 +205,7 @@ func TestEventDrivenHandoffIntegration(t *testing.T) {
 			return nil, &net.OpError{Op: "dial", Err: &net.DNSError{Name: addr}}
 		}
 
-		processor := hitless.NewPoolHook(3, failingDialer, nil, nil)
+		processor := hitless.NewPoolHook(failingDialer, nil, nil)
 		defer processor.Shutdown(context.Background())
 
 		// Create hooks manager and add processor as hook
@@ -211,12 +216,15 @@ func TestEventDrivenHandoffIntegration(t *testing.T) {
 			Dialer: func(ctx context.Context) (net.Conn, error) {
 				return &mockNetConn{addr: "original:6379"}, nil
 			},
-			PoolHooks: hookManager,
+			
 			PoolSize:       3,
 			PoolTimeout:    time.Second,
 		})
 		defer testPool.Close()
 
+		// Add the hook to the pool after creation
+		testPool.AddPoolHook(processor)
+
 		// Set the pool reference in the processor
 		processor.SetPool(testPool)
 
@@ -260,7 +268,7 @@ func TestEventDrivenHandoffIntegration(t *testing.T) {
 			return &mockNetConn{addr: addr}, nil
 		}
 
-		processor := hitless.NewPoolHook(3, slowDialer, nil, nil)
+		processor := hitless.NewPoolHook(slowDialer, nil, nil)
 
 		// Create hooks manager and add processor as hook
 		hookManager := pool.NewPoolHookManager()
@@ -270,12 +278,15 @@ func TestEventDrivenHandoffIntegration(t *testing.T) {
 			Dialer: func(ctx context.Context) (net.Conn, error) {
 				return &mockNetConn{addr: "original:6379"}, nil
 			},
-			PoolHooks: hookManager,
+			
 			PoolSize:       2,
 			PoolTimeout:    time.Second,
 		})
 		defer testPool.Close()
 
+		// Add the hook to the pool after creation
+		testPool.AddPoolHook(processor)
+
 		// Set the pool reference in the processor
 		processor.SetPool(testPool)
 

commands.go

@@ -524,6 +524,9 @@ func (c cmdable) ClientInfo(ctx context.Context) *ClientInfoCmd {
 func (c cmdable) ClientMaintNotifications(ctx context.Context, enabled bool, endpointType string) *StatusCmd {
 	args := []interface{}{"client", "maint_notifications"}
 	if enabled {
+		if endpointType == "" {
+			endpointType = "none"
+		}
 		args = append(args, "on", "moving-endpoint-type", endpointType)
 	} else {
 		args = append(args, "off")

hitless/config_test.go

@@ -249,7 +249,7 @@ func TestProcessorWithConfig(t *testing.T) {
 			return &mockNetConn{addr: addr}, nil
 		}
 
-		processor := NewPoolHook(3, baseDialer, config, nil)
+		processor := NewPoolHook(baseDialer, config, nil)
 		defer processor.Shutdown(context.Background())
 
 		// The processor should be created successfully with custom config
@@ -269,7 +269,7 @@ func TestProcessorWithConfig(t *testing.T) {
 			return &mockNetConn{addr: addr}, nil
 		}
 
-		processor := NewPoolHook(3, baseDialer, config, nil)
+		processor := NewPoolHook(baseDialer, config, nil)
 		defer processor.Shutdown(context.Background())
 
 		// Should work with partial config (defaults applied)
@@ -283,7 +283,7 @@ func TestProcessorWithConfig(t *testing.T) {
 			return &mockNetConn{addr: addr}, nil
 		}
 
-		processor := NewPoolHook(3, baseDialer, nil, nil)
+		processor := NewPoolHook(baseDialer, nil, nil)
 		defer processor.Shutdown(context.Background())
 
 		// Should use default config when nil is passed
@@ -308,7 +308,7 @@ func TestIntegrationWithApplyDefaults(t *testing.T) {
 		}
 
 		// Create processor - should apply defaults to missing fields
-		processor := NewPoolHook(3, baseDialer, partialConfig, nil)
+		processor := NewPoolHook(baseDialer, partialConfig, nil)
 		defer processor.Shutdown(context.Background())
 
 		// Processor should be created successfully

hitless/hitless_manager.go

@@ -10,6 +10,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"
 )
 
 // Push notification type constants for hitless upgrades
@@ -56,6 +57,7 @@ type HitlessManager struct {
 	client  interfaces.ClientInterface
 	config  *Config
 	options interfaces.OptionsInterface
+	pool    pool.Pooler
 
 	// MOVING operation tracking - using sync.Map for better concurrent performance
 	activeMovingOps sync.Map // map[MovingOperationKey]*MovingOperation
@@ -65,8 +67,9 @@ type HitlessManager struct {
 	closed               atomic.Bool  // Manager closed state
 
 	// Notification hooks for extensibility
-	hooks   []NotificationHook
-	hooksMu sync.RWMutex // Protects hooks slice
+	hooks        []NotificationHook
+	hooksMu      sync.RWMutex // Protects hooks slice
+	poolHooksRef *PoolHook
 }
 
 // MovingOperation tracks an active MOVING operation.
@@ -78,13 +81,14 @@ type MovingOperation struct {
 }
 
 // NewHitlessManager creates a new simplified hitless manager.
-func NewHitlessManager(client interfaces.ClientInterface, config *Config) (*HitlessManager, error) {
+func NewHitlessManager(client interfaces.ClientInterface, pool pool.Pooler, config *Config) (*HitlessManager, error) {
 	if client == nil {
 		return nil, ErrInvalidClient
 	}
 
 	hm := &HitlessManager{
 		client:  client,
+		pool:    pool,
 		options: client.GetOptions(),
 		config:  config.Clone(),
 		hooks:   make([]NotificationHook, 0),
@@ -98,9 +102,10 @@ func NewHitlessManager(client interfaces.ClientInterface, config *Config) (*Hitl
 	return hm, nil
 }
 
-// CreatePoolHook creates a pool hook with a custom dialer.
-func (hm *HitlessManager) CreatePoolHook(baseDialer func(context.Context, string, string) (net.Conn, error)) *PoolHook {
-	return hm.createPoolHook(baseDialer)
+// GetPoolHook creates a pool hook with a custom dialer.
+func (hm *HitlessManager) InitPoolHook(baseDialer func(context.Context, string, string) (net.Conn, error)) {
+	poolHook := hm.createPoolHook(baseDialer)
+	hm.pool.AddPoolHook(poolHook)
 }
 
 // setupPushNotifications sets up push notification handling by registering with the client's processor.
@@ -208,6 +213,16 @@ func (hm *HitlessManager) Close() error {
 		return nil // Already closed
 	}
 
+	// Shutdown the pool hook
+	err := hm.poolHooksRef.Shutdown(context.Background())
+	if err != nil {
+		// was not able to close pool hook, keep closed state false
+		hm.closed.Store(false)
+		return err
+	}
+	// Remove the pool hook from the pool
+	hm.pool.RemovePoolHook(hm.poolHooksRef)
+
 	// Clear all active operations
 	hm.activeMovingOps.Range(func(key, value interface{}) bool {
 		hm.activeMovingOps.Delete(key)
@@ -299,13 +314,17 @@ func (fh *FilterHook) PostHook(ctx context.Context, notificationType string, not
 
 // 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 {
+		return hm.poolHooksRef
+	}
 	// Get pool size from client options for better worker defaults
 	poolSize := 0
 	if hm.options != nil {
 		poolSize = hm.options.GetPoolSize()
 	}
 
-	hook := NewPoolHookWithPoolSize(baseDialer, hm.config, hm, poolSize)
+	hm.poolHooksRef = NewPoolHookWithPoolSize(baseDialer, hm.config, hm, poolSize)
+	hm.poolHooksRef.SetPool(hm.pool)
 
-	return hook
+	return hm.poolHooksRef
 }