Increase keepalive timeout. Add reconnection delay.

outbound/config.go

@@ -168,6 +168,7 @@ func WithWorkerFactory(wf WorkerFactory) PoolOption {
 
 type WorkerConfig struct {
 	KeepaliveTimeout time.Duration
+	ReconnectDelay   time.Duration
 	MaxQueueSize     int
 	LoggingEnabled   bool
 	LogLevel         *slog.Level
@@ -188,7 +189,8 @@ func NewWorkerConfig(options ...WorkerOption) (*WorkerConfig, error) {
 
 func GetDefaultWorkerConfig() *WorkerConfig {
 	return &WorkerConfig{
-		KeepaliveTimeout: 20 * time.Second,
+		KeepaliveTimeout: 60 * time.Second,
+		ReconnectDelay:   2 * time.Second,
 		MaxQueueSize:     0, // disabled by default
 		LoggingEnabled:   true,
 		LogLevel:         nil,
@@ -232,6 +234,13 @@ func validateKeepaliveTimeout(value time.Duration) error {
 	return nil
 }
 
+func validateReconnectDelay(value time.Duration) error {
+	if value < 0 {
+		return InvalidReconnectDelay
+	}
+	return nil
+}
+
 // When KeepaliveTimeout is set to zero, keepalive timeouts are disabled.
 func WithKeepaliveTimeout(value time.Duration) WorkerOption {
 	return func(c *WorkerConfig) error {
@@ -244,6 +253,17 @@ func WithKeepaliveTimeout(value time.Duration) WorkerOption {
 	}
 }
 
+func WithReconnectDelay(value time.Duration) WorkerOption {
+	return func(c *WorkerConfig) error {
+		err := validateReconnectDelay(value)
+		if err != nil {
+			return err
+		}
+		c.ReconnectDelay = value
+		return nil
+	}
+}
+
 // When MaxQueueSize is set to zero, queue limits are disabled.
 func WithMaxQueueSize(value int) WorkerOption {
 	return func(c *WorkerConfig) error {

outbound/errors.go

@@ -6,6 +6,7 @@ import "fmt"
 var (
 	// Config errors
 	InvalidKeepaliveTimeout = errors.New("keepalive timeout cannot be negative")
+	InvalidReconnectDelay   = errors.New("reconnect delay cannot be negative")
 	InvalidMaxQueueSize     = errors.New("maximum queue size cannot be negative")
 	InvalidBufferSize       = errors.New("buffer size must be greater than zero")
 

outbound/worker.go

@@ -100,6 +100,7 @@ func (w *DefaultWorker) Start(pool PoolPlugin) {
 			dial:           dial,
 			keepalive:      keepalive,
 			newConn:        newConn,
+			reconnectDelay: w.config.ReconnectDelay,
 			logger:         w.logger,
 		}
 		session.Start(w.ctx, pool)
@@ -153,6 +154,8 @@ type Session struct {
 	keepalive <-chan struct{}
 	newConn   <-chan *transport.Connection
 
+	reconnectDelay time.Duration
+
 	logger *slog.Logger
 }
 
@@ -238,6 +241,7 @@ func (s *Session) Start(
 		}
 
 		// refresh session
+		time.Sleep(s.reconnectDelay)
 	}
 
 }