refactor processor

Author: ndyakov

async_handoff_integration_test.go

@@ -41,17 +41,20 @@ func TestEventDrivenHandoffIntegration(t *testing.T) {
 		}
 
 		// Create processor with event-driven handoff support
-		processor := hitless.NewRedisConnectionProcessor(3, baseDialer, nil, nil)
+		processor := hitless.NewPoolHook(3, baseDialer, nil, nil)
 		defer processor.Shutdown(context.Background())
 
-		// Create a test pool with the processor
+		// Create a test pool with hooks
+		hookManager := pool.NewPoolHookManager()
+		hookManager.AddHook(processor)
+
 		testPool := pool.NewConnPool(&pool.Options{
 			Dialer: func(ctx context.Context) (net.Conn, error) {
 				return &mockNetConn{addr: "original:6379"}, nil
 			},
-			ConnectionProcessor: processor,
-			PoolSize:           5,
-			PoolTimeout:        time.Second,
+			PoolHooks: hookManager,
+			PoolSize:       5,
+			PoolTimeout:    time.Second,
 		})
 		defer testPool.Close()
 
@@ -128,16 +131,20 @@ func TestEventDrivenHandoffIntegration(t *testing.T) {
 			return &mockNetConn{addr: addr}, nil
 		}
 
-		processor := hitless.NewRedisConnectionProcessor(3, baseDialer, nil, nil)
+		processor := hitless.NewPoolHook(3, baseDialer, nil, nil)
 		defer processor.Shutdown(context.Background())
 
+		// Create hooks manager and add processor as hook
+		hookManager := pool.NewPoolHookManager()
+		hookManager.AddHook(processor)
+
 		testPool := pool.NewConnPool(&pool.Options{
 			Dialer: func(ctx context.Context) (net.Conn, error) {
 				return &mockNetConn{addr: "original:6379"}, nil
 			},
-			ConnectionProcessor: processor,
-			PoolSize:           10,
-			PoolTimeout:        time.Second,
+			PoolHooks: hookManager,
+			PoolSize:       10,
+			PoolTimeout:    time.Second,
 		})
 		defer testPool.Close()
 
@@ -193,16 +200,20 @@ func TestEventDrivenHandoffIntegration(t *testing.T) {
 			return nil, &net.OpError{Op: "dial", Err: &net.DNSError{Name: addr}}
 		}
 
-		processor := hitless.NewRedisConnectionProcessor(3, failingDialer, nil, nil)
+		processor := hitless.NewPoolHook(3, failingDialer, nil, nil)
 		defer processor.Shutdown(context.Background())
 
+		// Create hooks manager and add processor as hook
+		hookManager := pool.NewPoolHookManager()
+		hookManager.AddHook(processor)
+
 		testPool := pool.NewConnPool(&pool.Options{
 			Dialer: func(ctx context.Context) (net.Conn, error) {
 				return &mockNetConn{addr: "original:6379"}, nil
 			},
-			ConnectionProcessor: processor,
-			PoolSize:           3,
-			PoolTimeout:        time.Second,
+			PoolHooks: hookManager,
+			PoolSize:       3,
+			PoolTimeout:    time.Second,
 		})
 		defer testPool.Close()
 
@@ -249,15 +260,19 @@ func TestEventDrivenHandoffIntegration(t *testing.T) {
 			return &mockNetConn{addr: addr}, nil
 		}
 
-		processor := hitless.NewRedisConnectionProcessor(3, slowDialer, nil, nil)
+		processor := hitless.NewPoolHook(3, slowDialer, nil, nil)
+
+		// Create hooks manager and add processor as hook
+		hookManager := pool.NewPoolHookManager()
+		hookManager.AddHook(processor)
 
 		testPool := pool.NewConnPool(&pool.Options{
 			Dialer: func(ctx context.Context) (net.Conn, error) {
 				return &mockNetConn{addr: "original:6379"}, nil
 			},
-			ConnectionProcessor: processor,
-			PoolSize:           2,
-			PoolTimeout:        time.Second,
+			PoolHooks: hookManager,
+			PoolSize:       2,
+			PoolTimeout:    time.Second,
 		})
 		defer testPool.Close()
 

hitless/config_test.go

@@ -249,7 +249,7 @@ func TestProcessorWithConfig(t *testing.T) {
 			return &mockNetConn{addr: addr}, nil
 		}
 
-		processor := NewRedisConnectionProcessor(3, baseDialer, config, nil)
+		processor := NewPoolHook(3, 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 := NewRedisConnectionProcessor(3, baseDialer, config, nil)
+		processor := NewPoolHook(3, 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 := NewRedisConnectionProcessor(3, baseDialer, nil, nil)
+		processor := NewPoolHook(3, 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 := NewRedisConnectionProcessor(3, baseDialer, partialConfig, nil)
+		processor := NewPoolHook(3, baseDialer, partialConfig, nil)
 		defer processor.Shutdown(context.Background())
 
 		// Processor should be created successfully

hitless/hitless_manager.go

@@ -98,27 +98,9 @@ func NewHitlessManager(client interfaces.ClientInterface, config *Config) (*Hitl
 	return hm, nil
 }
 
-// AddHook adds a notification hook to the manager.
-// Hooks are called in the order they were added.
-func (hm *HitlessManager) AddHook(hook NotificationHook) {
-	hm.hooksMu.Lock()
-	defer hm.hooksMu.Unlock()
-	hm.hooks = append(hm.hooks, hook)
-}
-
-// RemoveHook removes a notification hook from the manager.
-func (hm *HitlessManager) RemoveHook(hook NotificationHook) {
-	hm.hooksMu.Lock()
-	defer hm.hooksMu.Unlock()
-
-	for i, h := range hm.hooks {
-		if h == hook {
-			// Remove hook by swapping with last element and truncating
-			hm.hooks[i] = hm.hooks[len(hm.hooks)-1]
-			hm.hooks = hm.hooks[:len(hm.hooks)-1]
-			break
-		}
-	}
+// 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)
 }
 
 // setupPushNotifications sets up push notification handling by registering with the client's processor.
@@ -246,11 +228,6 @@ func (hm *HitlessManager) GetState() State {
 	return StateIdle
 }
 
-// GetConfig returns the hitless manager configuration.
-func (hm *HitlessManager) GetConfig() *Config {
-	return hm.config.Clone()
-}
-
 // 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) {
 	hm.hooksMu.RLock()
@@ -320,16 +297,15 @@ func (fh *FilterHook) PostHook(ctx context.Context, notificationType string, not
 	// No post-processing needed for filter hook
 }
 
-// CreateConnectionProcessor creates a connection processor with this manager already set.
-// Returns the processor as the shared interface type.
-func (hm *HitlessManager) CreateConnectionProcessor(protocol int, baseDialer func(context.Context, string, string) (net.Conn, error)) *RedisConnectionProcessor {
+// createPoolHook creates a pool hook with this manager already set.
+func (hm *HitlessManager) createPoolHook(baseDialer func(context.Context, string, string) (net.Conn, error)) *PoolHook {
 	// Get pool size from client options for better worker defaults
 	poolSize := 0
 	if hm.options != nil {
 		poolSize = hm.options.GetPoolSize()
 	}
 
-	processor := NewRedisConnectionProcessorWithPoolSize(protocol, baseDialer, hm.config, hm, poolSize)
+	hook := NewPoolHookWithPoolSize(baseDialer, hm.config, hm, poolSize)
 
-	return processor
+	return hook
 }

hitless/hitless_manager_test.go

@@ -3,7 +3,6 @@ package hitless
 import (
 	"context"
 	"net"
-	"sync/atomic"
 	"testing"
 	"time"
 
@@ -71,32 +70,6 @@ func (mo *MockOptions) NewDialer() func(context.Context) (net.Conn, error) {
 	}
 }
 
-// TestHook for testing hook functionality
-type TestHook struct {
-	PreHookCalled      atomic.Int32
-	PostHookCalled     atomic.Int32
-	ModifyNotification bool
-	BlockProcessing    bool
-}
-
-func (th *TestHook) PreHook(ctx context.Context, notificationType string, notification []interface{}) ([]interface{}, bool) {
-	th.PreHookCalled.Add(1)
-
-	if th.ModifyNotification && notificationType == NotificationMoving && len(notification) > 3 {
-		// Modify the endpoint in the notification
-		modifiedNotification := make([]interface{}, len(notification))
-		copy(modifiedNotification, notification)
-		modifiedNotification[3] = "modified-endpoint:6379"
-		return modifiedNotification, !th.BlockProcessing
-	}
-
-	return notification, !th.BlockProcessing
-}
-
-func (th *TestHook) PostHook(ctx context.Context, notificationType string, notification []interface{}, result error) {
-	th.PostHookCalled.Add(1)
-}
-
 func TestHitlessManagerRefactoring(t *testing.T) {
 	t.Run("AtomicStateTracking", func(t *testing.T) {
 		config := DefaultConfig()
@@ -201,73 +174,6 @@ func TestHitlessManagerRefactoring(t *testing.T) {
 		}
 	})
 
-	t.Run("HookFunctionality", func(t *testing.T) {
-		config := DefaultConfig()
-		client := &MockClient{options: &MockOptions{}}
-
-		manager, err := NewHitlessManager(client, config)
-		if err != nil {
-			t.Fatalf("Failed to create hitless manager: %v", err)
-		}
-		defer manager.Close()
-
-		// Create test hooks
-		hook1 := &TestHook{}
-		hook2 := &TestHook{ModifyNotification: true}
-		hook3 := &TestHook{BlockProcessing: true}
-
-		// Add hooks
-		manager.AddHook(hook1)
-		manager.AddHook(hook2)
-
-		ctx := context.Background()
-		notification := []interface{}{NotificationMoving, "12345", "30", "original-endpoint:6379"}
-
-		// Test pre-hooks
-		modifiedNotification, shouldContinue := manager.processPreHooks(ctx, NotificationMoving, notification)
-		if !shouldContinue {
-			t.Error("Expected processing to continue")
-		}
-
-		if hook1.PreHookCalled.Load() != 1 {
-			t.Errorf("Expected hook1 PreHook to be called once, got %d", hook1.PreHookCalled.Load())
-		}
-
-		if hook2.PreHookCalled.Load() != 1 {
-			t.Errorf("Expected hook2 PreHook to be called once, got %d", hook2.PreHookCalled.Load())
-		}
-
-		// Check if notification was modified by hook2
-		if len(modifiedNotification) < 4 || modifiedNotification[3] != "modified-endpoint:6379" {
-			t.Errorf("Expected notification to be modified, got %v", modifiedNotification)
-		}
-
-		// Test post-hooks
-		manager.processPostHooks(ctx, NotificationMoving, modifiedNotification, nil)
-
-		if hook1.PostHookCalled.Load() != 1 {
-			t.Errorf("Expected hook1 PostHook to be called once, got %d", hook1.PostHookCalled.Load())
-		}
-
-		if hook2.PostHookCalled.Load() != 1 {
-			t.Errorf("Expected hook2 PostHook to be called once, got %d", hook2.PostHookCalled.Load())
-		}
-
-		// Test blocking hook
-		manager.AddHook(hook3)
-		_, shouldContinue = manager.processPreHooks(ctx, NotificationMoving, notification)
-		if shouldContinue {
-			t.Error("Expected processing to be blocked by hook3")
-		}
-
-		// Test hook removal
-		manager.RemoveHook(hook3)
-		_, shouldContinue = manager.processPreHooks(ctx, NotificationMoving, notification)
-		if !shouldContinue {
-			t.Error("Expected processing to continue after removing blocking hook")
-		}
-	})
-
 	t.Run("DuplicateOperationHandling", func(t *testing.T) {
 		config := DefaultConfig()
 		client := &MockClient{options: &MockOptions{}}