wip

Author: ndyakov

adapters.go

@@ -52,6 +52,11 @@ func (oa *optionsAdapter) GetWriteTimeout() time.Duration {
 	return oa.options.WriteTimeout
 }
 
+// GetNetwork returns the network type.
+func (oa *optionsAdapter) GetNetwork() string {
+	return oa.options.Network
+}
+
 // GetAddr returns the connection address.
 func (oa *optionsAdapter) GetAddr() string {
 	return oa.options.Addr

async_handoff_integration_test.go

@@ -41,7 +41,7 @@ func TestEventDrivenHandoffIntegration(t *testing.T) {
 		}
 
 		// Create processor with event-driven handoff support
-		processor := hitless.NewPoolHook(baseDialer, nil, nil)
+		processor := hitless.NewPoolHook(baseDialer, "tcp", nil, nil)
 		defer processor.Shutdown(context.Background())
 
 		// Create a test pool with hooks
@@ -133,7 +133,7 @@ func TestEventDrivenHandoffIntegration(t *testing.T) {
 			return &mockNetConn{addr: addr}, nil
 		}
 
-		processor := hitless.NewPoolHook(baseDialer, nil, nil)
+		processor := hitless.NewPoolHook(baseDialer, "tcp", nil, nil)
 		defer processor.Shutdown(context.Background())
 
 		// Create hooks manager and add processor as hook
@@ -205,7 +205,7 @@ func TestEventDrivenHandoffIntegration(t *testing.T) {
 			return nil, &net.OpError{Op: "dial", Err: &net.DNSError{Name: addr}}
 		}
 
-		processor := hitless.NewPoolHook(failingDialer, nil, nil)
+		processor := hitless.NewPoolHook(failingDialer, "tcp", nil, nil)
 		defer processor.Shutdown(context.Background())
 
 		// Create hooks manager and add processor as hook
@@ -268,7 +268,7 @@ func TestEventDrivenHandoffIntegration(t *testing.T) {
 			return &mockNetConn{addr: addr}, nil
 		}
 
-		processor := hitless.NewPoolHook(slowDialer, nil, nil)
+		processor := hitless.NewPoolHook(slowDialer, "tcp", nil, nil)
 
 		// Create hooks manager and add processor as hook
 		hookManager := pool.NewPoolHookManager()

hitless/config_test.go

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

hitless/hitless_manager.go

@@ -286,7 +286,7 @@ func (hm *HitlessManager) createPoolHook(baseDialer func(context.Context, string
 		poolSize = hm.options.GetPoolSize()
 	}
 
-	hm.poolHooksRef = NewPoolHookWithPoolSize(baseDialer, hm.config, hm, poolSize)
+	hm.poolHooksRef = NewPoolHookWithPoolSize(baseDialer, hm.options.GetNetwork(), hm.config, hm, poolSize)
 	hm.poolHooksRef.SetPool(hm.pool)
 
 	return hm.poolHooksRef

hitless/hitless_manager_test.go

@@ -64,6 +64,10 @@ func (mo *MockOptions) GetPoolSize() int {
 	return 10
 }
 
+func (mo *MockOptions) GetNetwork() string {
+	return "tcp"
+}
+
 func (mo *MockOptions) NewDialer() func(context.Context) (net.Conn, error) {
 	return func(ctx context.Context) (net.Conn, error) {
 		return nil, nil

hitless/pool_hook.go

@@ -14,6 +14,7 @@ import (
 
 // HitlessManagerInterface defines the interface for completing handoff operations
 type HitlessManagerInterface interface {
+	TrackMovingOperationWithConnID(ctx context.Context, newEndpoint string, deadline time.Time, seqID int64, connID uint64) error
 	UntrackOperationWithConnID(seqID int64, connID uint64)
 }
 
@@ -31,8 +32,12 @@ type HandoffRequest struct {
 // with hitless upgrade support.
 type PoolHook struct {
 	// Base dialer for creating connections to new endpoints during handoffs
+	// args are network and address
 	baseDialer func(context.Context, string, string) (net.Conn, error)
 
+	// Network type (e.g., "tcp", "unix")
+	network string
+
 	// Event-driven handoff support
 	handoffQueue chan HandoffRequest // Queue for handoff requests
 	shutdown     chan struct{}       // Shutdown signal
@@ -55,7 +60,7 @@ type PoolHook struct {
 	// Simple state tracking
 	pending sync.Map // map[uint64]int64 (connID -> seqID)
 
-	// Configuration for the processor
+	// Configuration for the hitless upgrade
 	config *Config
 
 	// Hitless manager for operation completion tracking
@@ -66,19 +71,19 @@ type PoolHook struct {
 }
 
 // NewPoolHook creates a new pool hook
-func NewPoolHook(baseDialer func(context.Context, string, string) (net.Conn, error), config *Config, hitlessManager HitlessManagerInterface) *PoolHook {
-	return NewPoolHookWithPoolSize(baseDialer, config, hitlessManager, 0)
+func NewPoolHook(baseDialer func(context.Context, string, string) (net.Conn, error), network string, config *Config, hitlessManager HitlessManagerInterface) *PoolHook {
+	return NewPoolHookWithPoolSize(baseDialer, network, config, hitlessManager, 0)
 }
 
 // NewPoolHookWithPoolSize creates a new pool hook with pool size for better worker defaults
-func NewPoolHookWithPoolSize(baseDialer func(context.Context, string, string) (net.Conn, error), config *Config, hitlessManager HitlessManagerInterface, poolSize int) *PoolHook {
+func NewPoolHookWithPoolSize(baseDialer func(context.Context, string, string) (net.Conn, error), network string, config *Config, hitlessManager HitlessManagerInterface, poolSize int) *PoolHook {
 	// Apply defaults to any missing configuration fields, using pool size for worker calculations
 	config = config.ApplyDefaultsWithPoolSize(poolSize)
 
 	ph := &PoolHook{
 		// baseDialer is used to create connections to new endpoints during handoffs
 		baseDialer: baseDialer,
-		// Note: CLIENT MAINT_NOTIFICATIONS is handled during client initialization
+		network:    network,
 		// handoffQueue is a buffered channel for queuing handoff requests
 		handoffQueue: make(chan HandoffRequest, config.HandoffQueueSize),
 		// shutdown is a channel for signaling shutdown
@@ -551,7 +556,7 @@ func (ph *PoolHook) createEndpointDialer(endpoint string) func(context.Context)
 		}
 
 		// Use the base dialer to connect to the new endpoint
-		return ph.baseDialer(ctx, "tcp", net.JoinHostPort(host, port))
+		return ph.baseDialer(ctx, ph.network, net.JoinHostPort(host, port))
 	}
 }