diff --git a/helper_test.go b/helper_test.go index 5774cab..fbd8f94 100644 --- a/helper_test.go +++ b/helper_test.go @@ -19,7 +19,7 @@ func setupTestConnection(t *testing.T) ( socket, incoming, outgoing = honeybeetest.SetupTestSocket(t) var err error - conn, err = transport.NewConnectionFromSocket(context.Background(), socket, nil, nil) + conn, err = transport.NewConnection(context.Background(), socket, nil, nil) assert.NoError(t, err) return } diff --git a/transport/config.go b/transport/config.go deleted file mode 100644 index 8bef51b..0000000 --- a/transport/config.go +++ /dev/null @@ -1,292 +0,0 @@ -package transport - -import ( - "git.wisehodl.dev/jay/go-honeybee/types" - "net/http" - "time" -) - -// ---------------------------------------------------------------------------- -// Connection Config -// ---------------------------------------------------------------------------- - -// Types - -type CloseHandler func(code int, text string) error - -type ConnectionConfig struct { - CloseHandler CloseHandler - RequestHeader http.Header - WriteTimeout time.Duration - PingInterval time.Duration - IncomingBufferSize int - ErrorsBufferSize int - Retry RetryConfig - Dialer types.Dialer -} - -type RetryConfig struct { - Disabled bool - MaxRetries int - InitialDelay time.Duration - MaxDelay time.Duration - JitterFactor float64 -} - -type ConnectionOption func(*ConnectionConfig) error - -// Constructors - -func NewConnectionConfig(options ...ConnectionOption) (*ConnectionConfig, error) { - conf := GetDefaultConnectionConfig() - if err := applyConnectionOptions(conf, options...); err != nil { - return nil, err - } - if err := ValidateConnectionConfig(conf); err != nil { - return nil, err - } - return conf, nil -} - -func GetDefaultConnectionConfig() *ConnectionConfig { - header := http.Header{} - header.Set("User-Agent", "honeybee/0.1.0") - return &ConnectionConfig{ - CloseHandler: nil, - RequestHeader: header, - WriteTimeout: 30 * time.Second, - PingInterval: 20 * time.Second, - IncomingBufferSize: 100, - ErrorsBufferSize: 10, - Retry: RetryConfig{ - MaxRetries: 0, // Infinite retries - InitialDelay: 1 * time.Second, - MaxDelay: 60 * time.Second, - JitterFactor: 0.2, - }, - } -} - -func (c ConnectionConfig) Clone() ConnectionConfig { - if c.RequestHeader != nil { - c.RequestHeader = c.RequestHeader.Clone() - } - return c -} - -func applyConnectionOptions(config *ConnectionConfig, options ...ConnectionOption) error { - for _, option := range options { - if err := option(config); err != nil { - return err - } - } - return nil -} - -// Validation - -func ValidateConnectionConfig(config *ConnectionConfig) error { - err := validateWriteTimeout(config.WriteTimeout) - if err != nil { - return err - } - - if !config.Retry.Disabled { - err = validateMaxRetries(config.Retry.MaxRetries) - if err != nil { - return err - } - - err = validateInitialDelay(config.Retry.InitialDelay) - if err != nil { - return err - } - - err = validateMaxDelay(config.Retry.MaxDelay) - if err != nil { - return err - } - - err = validateJitterFactor(config.Retry.JitterFactor) - if err != nil { - return err - } - - if config.Retry.InitialDelay > config.Retry.MaxDelay { - return NewConfigError(InvalidDelays) - } - } - - return nil -} - -func validateWriteTimeout(value time.Duration) error { - if value < 0 { - return InvalidWriteTimeout - } - return nil -} - -func validatePingInterval(value time.Duration) error { - if value < 0 { - return InvalidPingInterval - } - return nil -} - -func validateBufferSize(value int) error { - if value < 1 { - return InvalidBufferSize - } - return nil -} - -func validateMaxRetries(value int) error { - if value < 0 { - return InvalidRetryMaxRetries - } - return nil -} - -func validateInitialDelay(value time.Duration) error { - if value <= 0 { - return InvalidRetryInitialDelay - } - return nil -} - -func validateMaxDelay(value time.Duration) error { - if value <= 0 { - return InvalidRetryMaxDelay - } - return nil -} - -func validateJitterFactor(value float64) error { - if value < 0.0 || value > 1.0 { - return InvalidRetryJitterFactor - } - return nil -} - -// Options - -func WithCloseHandler(handler CloseHandler) ConnectionOption { - return func(c *ConnectionConfig) error { - c.CloseHandler = handler - return nil - } -} - -func WithRequestHeader(header http.Header) ConnectionOption { - return func(c *ConnectionConfig) error { - c.RequestHeader = header.Clone() - return nil - } -} - -// When WriteTimeout is set to zero, read timeouts are disabled. -func WithWriteTimeout(value time.Duration) ConnectionOption { - return func(c *ConnectionConfig) error { - err := validateWriteTimeout(value) - if err != nil { - return err - } - c.WriteTimeout = value - return nil - } -} - -// When PingInterval is set to zero, ping frames are disabled. -func WithPingInterval(value time.Duration) ConnectionOption { - return func(c *ConnectionConfig) error { - err := validatePingInterval(value) - if err != nil { - return err - } - c.PingInterval = value - return nil - } -} - -func WithIncomingBufferSize(value int) ConnectionOption { - return func(c *ConnectionConfig) error { - if err := validateBufferSize(value); err != nil { - return err - } - c.IncomingBufferSize = value - return nil - } -} - -func WithErrorsBufferSize(value int) ConnectionOption { - return func(c *ConnectionConfig) error { - if err := validateBufferSize(value); err != nil { - return err - } - c.ErrorsBufferSize = value - return nil - } -} - -func WithConnectionDialer(d types.Dialer) ConnectionOption { - return func(c *ConnectionConfig) error { - c.Dialer = d - return nil - } -} - -func WithRetryDisabled() ConnectionOption { - return func(c *ConnectionConfig) error { - c.Retry.Disabled = true - return nil - } -} - -func WithRetryMaxRetries(value int) ConnectionOption { - return func(c *ConnectionConfig) error { - err := validateMaxRetries(value) - if err != nil { - return err - } - - c.Retry.MaxRetries = value - return nil - } -} - -func WithRetryInitialDelay(value time.Duration) ConnectionOption { - return func(c *ConnectionConfig) error { - err := validateInitialDelay(value) - if err != nil { - return err - } - - c.Retry.InitialDelay = value - return nil - } -} - -func WithRetryMaxDelay(value time.Duration) ConnectionOption { - return func(c *ConnectionConfig) error { - err := validateMaxDelay(value) - if err != nil { - return err - } - - c.Retry.MaxDelay = value - return nil - } -} - -func WithRetryJitterFactor(value float64) ConnectionOption { - return func(c *ConnectionConfig) error { - err := validateJitterFactor(value) - if err != nil { - return err - } - - c.Retry.JitterFactor = value - return nil - } -} diff --git a/transport/config_test.go b/transport/config_test.go deleted file mode 100644 index 11e5828..0000000 --- a/transport/config_test.go +++ /dev/null @@ -1,283 +0,0 @@ -package transport - -import ( - "git.wisehodl.dev/jay/go-honeybee/honeybeetest" - "github.com/stretchr/testify/assert" - "net/http" - "testing" - "time" -) - -// Connection Config Tests - -func TestNewConnectionConfig(t *testing.T) { - _, err := NewConnectionConfig() - - assert.NoError(t, err) - // errors propagate - _, err = NewConnectionConfig(WithRetryMaxRetries(-1)) - assert.Error(t, err) - - _, err = NewConnectionConfig(WithRetryInitialDelay(10), WithRetryMaxDelay(1)) - assert.Error(t, err) -} - -// Default Tests - -func TestDefaultConnectionConfig(t *testing.T) { - header := http.Header{} - header.Set("User-Agent", "honeybee/0.1.0") - conf := GetDefaultConnectionConfig() - - assert.Equal(t, conf, &ConnectionConfig{ - CloseHandler: nil, - RequestHeader: header, - WriteTimeout: 30 * time.Second, - PingInterval: 20 * time.Second, - IncomingBufferSize: 100, - ErrorsBufferSize: 10, - Retry: RetryConfig{ - MaxRetries: 0, - InitialDelay: 1 * time.Second, - MaxDelay: 60 * time.Second, - JitterFactor: 0.2, - }, - }) -} - -// Builder Tests - -func TestApplyConnectionOptions(t *testing.T) { - conf := &ConnectionConfig{} - err := applyConnectionOptions( - conf, - WithIncomingBufferSize(256), - WithErrorsBufferSize(100), - WithRetryMaxRetries(0), - WithRetryInitialDelay(3*time.Second), - WithRetryJitterFactor(0.5), - ) - - assert.NoError(t, err) - assert.Equal(t, 256, conf.IncomingBufferSize) - assert.Equal(t, 100, conf.ErrorsBufferSize) - assert.Equal(t, 0, conf.Retry.MaxRetries) - assert.Equal(t, 3*time.Second, conf.Retry.InitialDelay) - assert.Equal(t, 0.5, conf.Retry.JitterFactor) - - // errors propagate - err = applyConnectionOptions( - conf, - WithRetryMaxRetries(-10), - ) - - assert.ErrorIs(t, err, InvalidRetryMaxRetries) -} - -// Option Tests - -func TestWithCloseHandler(t *testing.T) { - conf := &ConnectionConfig{} - opt := WithCloseHandler(func(code int, text string) error { return nil }) - err := applyConnectionOptions(conf, opt) - assert.NoError(t, err) - assert.Nil(t, conf.CloseHandler(0, "")) -} - -func TestWithWriteTimeout(t *testing.T) { - conf := &ConnectionConfig{} - opt := WithWriteTimeout(30) - err := applyConnectionOptions(conf, opt) - assert.NoError(t, err) - assert.Equal(t, conf.WriteTimeout, time.Duration(30)) - - // zero allowed - conf = &ConnectionConfig{} - opt = WithWriteTimeout(0) - err = applyConnectionOptions(conf, opt) - assert.NoError(t, err) - assert.Equal(t, conf.WriteTimeout, time.Duration(0)) - - // negative disallowed - conf = &ConnectionConfig{} - opt = WithWriteTimeout(-30) - err = applyConnectionOptions(conf, opt) - assert.ErrorIs(t, err, InvalidWriteTimeout) - assert.ErrorContains(t, err, "write timeout cannot be negative") -} - -func TestWithRetry(t *testing.T) { - t.Run("without retry", func(t *testing.T) { - conf := GetDefaultConnectionConfig() - opt := WithRetryDisabled() - err := applyConnectionOptions(conf, opt) - assert.NoError(t, err) - assert.True(t, conf.Retry.Disabled) - }) - - t.Run("with attempts", func(t *testing.T) { - conf := &ConnectionConfig{} - opt := WithRetryMaxRetries(3) - err := applyConnectionOptions(conf, opt) - assert.NoError(t, err) - assert.Equal(t, 3, conf.Retry.MaxRetries) - - // zero allowed - opt = WithRetryMaxRetries(0) - err = applyConnectionOptions(conf, opt) - assert.NoError(t, err) - - // negative disallowed - opt = WithRetryMaxRetries(-10) - err = applyConnectionOptions(conf, opt) - assert.ErrorIs(t, err, InvalidRetryMaxRetries) - assert.ErrorContains(t, err, "max retry count cannot be negative") - }) - - t.Run("with initial delay", func(t *testing.T) { - conf := &ConnectionConfig{} - opt := WithRetryInitialDelay(10 * time.Second) - err := applyConnectionOptions(conf, opt) - assert.NoError(t, err) - assert.Equal(t, 10*time.Second, conf.Retry.InitialDelay) - - // zero disallowed - opt = WithRetryInitialDelay(0 * time.Second) - err = applyConnectionOptions(conf, opt) - assert.ErrorIs(t, err, InvalidRetryInitialDelay) - assert.ErrorContains(t, err, "initial delay must be positive") - - // negative disallowed - opt = WithRetryInitialDelay(-10 * time.Second) - err = applyConnectionOptions(conf, opt) - assert.ErrorIs(t, err, InvalidRetryInitialDelay) - }) - - t.Run("with max delay", func(t *testing.T) { - conf := &ConnectionConfig{} - opt := WithRetryMaxDelay(10 * time.Second) - err := applyConnectionOptions(conf, opt) - assert.NoError(t, err) - assert.Equal(t, 10*time.Second, conf.Retry.MaxDelay) - - // zero disallowed - opt = WithRetryMaxDelay(0 * time.Second) - err = applyConnectionOptions(conf, opt) - assert.ErrorIs(t, err, InvalidRetryMaxDelay) - assert.ErrorContains(t, err, "max delay must be positive") - - // negative disallowed - opt = WithRetryMaxDelay(-10 * time.Second) - err = applyConnectionOptions(conf, opt) - assert.ErrorIs(t, err, InvalidRetryMaxDelay) - }) - - t.Run("with jitter factor", func(t *testing.T) { - conf := &ConnectionConfig{} - - opt := WithRetryJitterFactor(0.2) - err := applyConnectionOptions(conf, opt) - assert.NoError(t, err) - assert.Equal(t, 0.2, conf.Retry.JitterFactor) - - // negative disallowed - opt = WithRetryJitterFactor(-1) - err = applyConnectionOptions(conf, opt) - assert.ErrorIs(t, err, InvalidRetryJitterFactor) - assert.ErrorContains(t, err, "jitter factor must be between 0.0 and 1.0") - - // >1 disallowed - opt = WithRetryJitterFactor(1.1) - err = applyConnectionOptions(conf, opt) - assert.ErrorIs(t, err, InvalidRetryJitterFactor) - }) -} - -// Validation Tests - -func TestValidateConnectionConfig(t *testing.T) { - cases := []struct { - name string - conf ConnectionConfig - wantErr error - wantErrText string - }{ - { - name: "valid empty", - conf: ConnectionConfig{Retry: RetryConfig{Disabled: true}}, - }, - { - name: "valid defaults", - conf: *GetDefaultConnectionConfig(), - }, - { - name: "valid complete", - conf: ConnectionConfig{ - CloseHandler: (func(code int, text string) error { return nil }), - WriteTimeout: time.Duration(30), - Retry: RetryConfig{ - MaxRetries: 0, - InitialDelay: 2 * time.Second, - MaxDelay: 10 * time.Second, - JitterFactor: 0.2, - }, - }, - }, - { - name: "invalid - initial delay > max delay", - conf: ConnectionConfig{ - Retry: RetryConfig{ - InitialDelay: 10 * time.Second, - MaxDelay: 1 * time.Second, - }, - }, - wantErrText: "initial delay may not exceed maximum delay", - }, - } - - for _, tc := range cases { - t.Run(tc.name, func(t *testing.T) { - err := ValidateConnectionConfig(&tc.conf) - - if tc.wantErr != nil || tc.wantErrText != "" { - if tc.wantErr != nil { - assert.ErrorIs(t, err, tc.wantErr) - } - - if tc.wantErrText != "" { - assert.ErrorContains(t, err, tc.wantErrText) - } - return - } - - assert.NoError(t, err) - }) - } -} - -func TestConnectionConfigClone(t *testing.T) { - header := http.Header{} - header.Set("X-Test", "val") - orig := ConnectionConfig{ - RequestHeader: header, - WriteTimeout: 5 * time.Second, - Retry: RetryConfig{Disabled: true}, - } - - cloned := orig.Clone() - - // values match - assert.Equal(t, orig.WriteTimeout, cloned.WriteTimeout) - assert.Equal(t, "val", cloned.RequestHeader.Get("X-Test")) - - // header is a distinct copy - cloned.RequestHeader.Set("X-Test", "mutated") - assert.Equal(t, "val", orig.RequestHeader.Get("X-Test")) -} - -func TestWithConnectionDialer(t *testing.T) { - mock := &honeybeetest.MockDialer{} - conf, err := NewConnectionConfig(WithConnectionDialer(mock)) - assert.NoError(t, err) - assert.Equal(t, mock, conf.Dialer) -} diff --git a/transport/connection.go b/transport/connection.go index 9fbd65d..ed6b60d 100644 --- a/transport/connection.go +++ b/transport/connection.go @@ -6,7 +6,6 @@ import ( "fmt" "log/slog" "math/rand" - "net/url" "sync" "sync/atomic" "time" @@ -20,30 +19,6 @@ import ( // Types // ---------------------------------------------------------------------------- -type ConnectionState int - -const ( - StateDisconnected ConnectionState = iota - StateConnecting - StateConnected - StateClosed -) - -func (s ConnectionState) String() string { - switch s { - case StateDisconnected: - return "disconnected" - case StateConnecting: - return "connecting" - case StateConnected: - return "connected" - case StateClosed: - return "closed" - default: - return "unknown" - } -} - type ConnectionStats struct { ChanIncoming int ChanErrors int @@ -53,6 +28,86 @@ type ConnectionStats struct { TotalHeartbeats uint64 } +// ---------------------------------------------------------------------------- +// Config +// ---------------------------------------------------------------------------- + +type ConnectionConfig struct { + WriteTimeout time.Duration + PingInterval time.Duration + IncomingBufferSize int + ErrorsBufferSize int +} + +func NewConnectionConfig(opts ...ConnectionOption) (*ConnectionConfig, error) { + conf := &ConnectionConfig{ + WriteTimeout: 30 * time.Second, + PingInterval: 20 * time.Second, + IncomingBufferSize: 100, + ErrorsBufferSize: 10, + } + for _, o := range opts { + o(conf) + } + + err := ValidateConnectionConfig(*conf) + if err != nil { + return nil, err + } + + return conf, nil +} + +type ConnectionOption func(*ConnectionConfig) + +// When WriteTimeout is set to zero, read timeouts are disabled. +func WithWriteTimeout(value time.Duration) ConnectionOption { + return func(c *ConnectionConfig) { + c.WriteTimeout = value + } +} + +// When PingInterval is set to zero, ping frames are disabled. +func WithPingInterval(value time.Duration) ConnectionOption { + return func(c *ConnectionConfig) { + c.PingInterval = value + } +} + +func WithIncomingBufferSize(value int) ConnectionOption { + return func(c *ConnectionConfig) { + c.IncomingBufferSize = value + } +} + +func WithErrorsBufferSize(value int) ConnectionOption { + return func(c *ConnectionConfig) { + c.ErrorsBufferSize = value + } +} + +func ValidateConnectionConfig(c ConnectionConfig) error { + if c.WriteTimeout < 0 { + return fmt.Errorf("invalid write timeout: %v", c.WriteTimeout) + } + + if c.PingInterval < 0 { + return fmt.Errorf("invalid ping interval: %v", c.PingInterval) + } + + if c.IncomingBufferSize < 1 { + return fmt.Errorf("invalid incoming buffer size: %d", + c.IncomingBufferSize) + } + + if c.ErrorsBufferSize < 1 { + return fmt.Errorf("invalid errors buffer size: %d", + c.ErrorsBufferSize) + } + + return nil +} + // ---------------------------------------------------------------------------- // Connection // ---------------------------------------------------------------------------- @@ -62,8 +117,6 @@ type ConnectionStats struct { // -------------------------/ type Connection struct { - url *url.URL - dialer types.Dialer socket types.Socket config ConnectionConfig logger *slog.Logger @@ -77,8 +130,6 @@ type Connection struct { outgoingCount *atomic.Uint64 heartbeatCount *atomic.Uint64 - state ConnectionState - wg sync.WaitGroup closed bool mu sync.RWMutex @@ -87,16 +138,18 @@ type Connection struct { cleanupOnce sync.Once } -func NewConnection(ctx context.Context, urlStr string, config *ConnectionConfig, handler slog.Handler) (*Connection, error) { +func NewConnection( + ctx context.Context, socket types.Socket, config *ConnectionConfig, handler slog.Handler, +) (*Connection, error) { + if socket == nil { + return nil, NewConnectionError(ErrNilSocket) + } + if config == nil { - config = GetDefaultConnectionConfig() + config, _ = NewConnectionConfig() } - if err := ValidateConnectionConfig(config); err != nil { - return nil, err - } - - url, err := ParseURL(urlStr) + err := ValidateConnectionConfig(*config) if err != nil { return nil, err } @@ -107,71 +160,15 @@ func NewConnection(ctx context.Context, urlStr string, config *ConnectionConfig, ctx = component.MustExtend(ctx, "connection") } - // Clone config to ensure full ownership of all fields. - cc := config.Clone() - if cc.Dialer == nil { - cc.Dialer = NewDialer() - } - conn := &Connection{ - url: url, - dialer: cc.Dialer, - socket: nil, - config: cc, - incoming: make(chan []byte, cc.IncomingBufferSize), - heartbeat: make(chan struct{}, 1), - errors: make(chan error, cc.ErrorsBufferSize), - incomingCount: &atomic.Uint64{}, - outgoingCount: &atomic.Uint64{}, - heartbeatCount: &atomic.Uint64{}, - state: StateDisconnected, - done: make(chan struct{}), - } - - if handler != nil { - comp := component.FromContext(ctx) - conn.logger = slog.New(handler).With(slog.Any("component", comp)) - } - - return conn, nil -} - -func NewConnectionFromSocket( - ctx context.Context, socket types.Socket, config *ConnectionConfig, handler slog.Handler, -) (*Connection, error) { - if socket == nil { - return nil, NewConnectionError(ErrNilSocket) - } - - if config == nil { - config = GetDefaultConnectionConfig() - } - - if err := ValidateConnectionConfig(config); err != nil { - return nil, err - } - - if component.FromContext(ctx) == nil { - ctx = component.MustNew(ctx, "honeybee", "connection") - } else { - ctx = component.MustExtend(ctx, "connection") - } - - // Clone config to ensure full ownership of all fields. - cc := config.Clone() - - conn := &Connection{ - url: nil, - dialer: nil, socket: socket, - config: cc, - incoming: make(chan []byte, cc.IncomingBufferSize), + config: *config, + incoming: make(chan []byte, config.IncomingBufferSize), heartbeat: make(chan struct{}, 1), - errors: make(chan error, cc.ErrorsBufferSize), + errors: make(chan error, config.ErrorsBufferSize), incomingCount: &atomic.Uint64{}, outgoingCount: &atomic.Uint64{}, heartbeatCount: &atomic.Uint64{}, - state: StateConnected, done: make(chan struct{}), } @@ -180,11 +177,6 @@ func NewConnectionFromSocket( conn.logger = slog.New(handler).With(slog.Any("component", comp)) } - // initialize - if config.CloseHandler != nil { - socket.SetCloseHandler(config.CloseHandler) - } - conn.setupPongHandler() if conn.config.PingInterval > 0 { @@ -200,65 +192,6 @@ func NewConnectionFromSocket( // Methods // -------------------------/ -func (c *Connection) Connect(ctx context.Context, onDialError func(error)) error { - c.mu.Lock() - defer c.mu.Unlock() - - if c.socket != nil { - return NewConnectionError(ErrSocketExists) - } - - if c.closed { - return NewConnectionError(ErrConnectionClosed) - } - - // begin connecting - if c.logger != nil { - c.logger.Debug("connecting") - } - - c.state = StateConnecting - - // obtain socket - retryMgr := NewRetryManager(c.config.Retry) - socket, _, err := AcquireSocket( - ctx, retryMgr, c.dialer, c.url.String(), c.config.RequestHeader, c.logger, onDialError) - - if err != nil { - // socket acquisition failed - c.state = StateDisconnected - if c.logger != nil { - c.logger.Warn("connection failed", "error", err) - } - return NewConnectionError(err) - } - - // got socket - c.socket = socket - - // initialize - if c.config.CloseHandler != nil { - c.socket.SetCloseHandler(c.config.CloseHandler) - } - - c.setupPongHandler() - - if c.config.PingInterval > 0 { - c.wg.Go(c.startPinger) - } - - c.wg.Go(c.startReader) - - // connected - c.state = StateConnected - - if c.logger != nil { - c.logger.Debug("connected") - } - - return nil -} - func (c *Connection) Send(data []byte) error { c.writeMu.Lock() defer c.writeMu.Unlock() @@ -304,12 +237,6 @@ func (c *Connection) Errors() <-chan error { return c.errors } -func (c *Connection) State() ConnectionState { - c.mu.RLock() - defer c.mu.RUnlock() - return c.state -} - func (c *Connection) Stats() ConnectionStats { return ConnectionStats{ ChanIncoming: len(c.incoming), @@ -459,7 +386,6 @@ func (c *Connection) shutdownExternal() { return } c.closed = true - c.state = StateClosed c.mu.Unlock() // perform shutdown @@ -479,7 +405,6 @@ func (c *Connection) shutdownInternal() { return } c.closed = true - c.state = StateClosed c.mu.Unlock() // perform shutdown diff --git a/transport/connection_close_test.go b/transport/connection_close_test.go deleted file mode 100644 index ce6e3af..0000000 --- a/transport/connection_close_test.go +++ /dev/null @@ -1,103 +0,0 @@ -package transport - -import ( - "bytes" - "context" - "fmt" - "git.wisehodl.dev/jay/go-honeybee/honeybeetest" - "github.com/gorilla/websocket" - "github.com/stretchr/testify/assert" - "testing" -) - -func TestDisconnectedConnectionClose(t *testing.T) { - t.Run("close succeeds on disconnected connection", func(t *testing.T) { - conn, err := NewConnection(context.Background(), "ws://test", nil, nil) - assert.NoError(t, err) - assert.Equal(t, StateDisconnected, conn.State()) - - conn.Close() - assert.Equal(t, StateClosed, conn.State()) - }) - - t.Run("close is idempotent", func(t *testing.T) { - conn, err := NewConnection(context.Background(), "ws://test", nil, nil) - assert.NoError(t, err) - - conn.Close() - conn.Close() - assert.Equal(t, StateClosed, conn.State()) - }) - - t.Run("close with nil socket", func(t *testing.T) { - conn, err := NewConnection(context.Background(), "ws://test", nil, nil) - assert.NoError(t, err) - assert.Nil(t, conn.socket) - - conn.Close() - assert.Equal(t, StateClosed, conn.State()) - }) - - t.Run("socket close error does not propagate", func(t *testing.T) { - expectedErr := fmt.Errorf("socket close failed") - mockSocket := honeybeetest.NewMockSocket() - mockSocket.CloseFunc = func() error { - return expectedErr - } - - conn, err := NewConnection(context.Background(), "ws://test", nil, nil) - assert.NoError(t, err) - conn.socket = mockSocket - - conn.Close() - assert.Equal(t, StateClosed, conn.State()) - }) - - t.Run("channels close after close", func(t *testing.T) { - conn, err := NewConnection(context.Background(), "ws://test", nil, nil) - assert.NoError(t, err) - - conn.Close() - - assert.True(t, conn.closed) - _, ok := <-conn.incoming - assert.False(t, ok) - _, ok = <-conn.errors - assert.False(t, ok) - }) - - t.Run("send fails after close", func(t *testing.T) { - conn, err := NewConnection(context.Background(), "ws://test", nil, nil) - assert.NoError(t, err) - - conn.Close() - - err = conn.Send([]byte("test")) - assert.Error(t, err) - assert.ErrorIs(t, err, ErrConnectionClosed) - }) - -} - -func TestConnectedConnectionClose(t *testing.T) { - t.Run("blocked on ReadMessage, unblocks on closed", func(t *testing.T) { - conn, _, incomingData, _ := setupTestConnection(t) - - // Send a message to ensure reader loop is blocking - canary := []byte("canary") - incomingData <- honeybeetest.MockIncomingData{ - MsgType: websocket.TextMessage, Data: canary} - - honeybeetest.Eventually(t, func() bool { - select { - case msg := <-conn.Incoming(): - return bytes.Equal(msg, canary) - default: - return false - } - }, "expected canary message") - - conn.Close() - assert.Equal(t, StateClosed, conn.State()) - }) -} diff --git a/transport/connection_goroutine_test.go b/transport/connection_goroutine_test.go deleted file mode 100644 index e1ea04b..0000000 --- a/transport/connection_goroutine_test.go +++ /dev/null @@ -1,77 +0,0 @@ -package transport - -import ( - "context" - "git.wisehodl.dev/jay/go-honeybee/honeybeetest" - "github.com/gorilla/websocket" - "github.com/stretchr/testify/assert" - "io" - "testing" -) - -func TestStartReader(t *testing.T) { - t.Run("text messages route to incoming channel", func(t *testing.T) { - conn, _, incomingData, _ := setupTestConnection(t) - defer conn.Close() - - testData := []byte("hello") - incomingData <- honeybeetest.MockIncomingData{ - MsgType: websocket.TextMessage, - Data: testData, - Err: nil, - } - - honeybeetest.ExpectIncoming(t, conn.Incoming(), testData) - }) - - t.Run("binary messages route to incoming channel", func(t *testing.T) { - conn, _, incomingData, _ := setupTestConnection(t) - defer conn.Close() - - testData := []byte{0x00, 0x01, 0x02} - incomingData <- honeybeetest.MockIncomingData{ - MsgType: websocket.BinaryMessage, - Data: testData, - Err: nil, - } - - honeybeetest.ExpectIncoming(t, conn.Incoming(), testData) - }) - - t.Run("multiple messages processed sequentially", func(t *testing.T) { - conn, _, incomingData, _ := setupTestConnection(t) - defer conn.Close() - - messages := [][]byte{[]byte("first"), []byte("second"), []byte("third")} - for _, msg := range messages { - incomingData <- honeybeetest.MockIncomingData{ - MsgType: websocket.TextMessage, Data: msg, Err: nil} - } - - for _, expected := range messages { - honeybeetest.ExpectIncoming(t, conn.Incoming(), expected) - } - }) - - t.Run("reader exits on socket read error", func(t *testing.T) { - mockSocket := honeybeetest.NewMockSocket() - - mockSocket.CloseFunc = func() error { - mockSocket.Once.Do(func() { - close(mockSocket.Closed) - }) - return nil - } - - mockSocket.ReadMessageFunc = func() (int, []byte, error) { - return 0, nil, io.EOF - } - - conn, err := NewConnectionFromSocket(context.Background(), mockSocket, nil, nil) - assert.NoError(t, err) - - honeybeetest.Eventually(t, func() bool { - return conn.State() == StateClosed - }, "expected closed state") - }) -} diff --git a/transport/connection_send_test.go b/transport/connection_send_test.go deleted file mode 100644 index 1af00cf..0000000 --- a/transport/connection_send_test.go +++ /dev/null @@ -1,240 +0,0 @@ -package transport - -import ( - "context" - "fmt" - "git.wisehodl.dev/jay/go-honeybee/honeybeetest" - "github.com/gorilla/websocket" - "github.com/stretchr/testify/assert" - "io" - "sync" - "testing" - "time" -) - -func TestConnectionSend(t *testing.T) { - t.Run("writes message to socket", func(t *testing.T) { - conn, _, _, outgoingData := setupTestConnection(t) - defer conn.Close() - - testData := []byte("test message") - err := conn.Send(testData) - assert.NoError(t, err) - - honeybeetest.ExpectWrite(t, outgoingData, websocket.TextMessage, testData) - }) - - t.Run("writes multiple message to socket", func(t *testing.T) { - conn, _, _, outgoingData := setupTestConnection(t) - defer conn.Close() - - messages := [][]byte{[]byte("first"), []byte("second"), []byte("third")} - for _, msg := range messages { - err := conn.Send(msg) - assert.NoError(t, err) - } - - for _, expected := range messages { - honeybeetest.ExpectWrite(t, outgoingData, websocket.TextMessage, expected) - } - }) - - t.Run("concurrent sends write messages to socket", func(t *testing.T) { - conn, _, _, outgoingData := setupTestConnection(t) - defer conn.Close() - - mu := sync.Mutex{} - messages := []string{} - done := make(chan struct{}) - - go func() { - for { - select { - case msg := <-outgoingData: - mu.Lock() - messages = append(messages, string(msg.Data)) - mu.Unlock() - case <-done: - return - } - } - }() - - defer close(done) - - var wg sync.WaitGroup - for i := range 5 { - wg.Add(1) - go func(id int) { - defer wg.Done() - for j := range 10 { - data := fmt.Appendf(nil, "msg-%d-%d", id, j) - for { - // send and retry until success - err := conn.Send(data) - if err != nil { - continue - } else { - break - } - } - } - }(i) - } - - wg.Wait() - - honeybeetest.Eventually(t, func() bool { - mu.Lock() - defer mu.Unlock() - return len(messages) == 50 - }, "should have received 50 messages") - - }) - - t.Run("send fails when connection is closed", func(t *testing.T) { - conn, _, _, _ := setupTestConnection(t) - conn.Close() - - testData := []byte("test message") - err := conn.Send(testData) - assert.ErrorIs(t, err, ErrConnectionClosed) - }) - - t.Run("write timeout disabled when zero", func(t *testing.T) { - config := &ConnectionConfig{WriteTimeout: 0, Retry: RetryConfig{Disabled: true}} - - outgoingData := make(chan honeybeetest.MockOutgoingData, 10) - mockSocket := honeybeetest.NewMockSocket() - - mockSocket.CloseFunc = func() error { - mockSocket.Once.Do(func() { - close(mockSocket.Closed) - }) - return nil - } - - deadlineCalled := make(chan struct{}, 1) - mockSocket.SetWriteDeadlineFunc = func(t time.Time) error { - deadlineCalled <- struct{}{} - return nil - } - - mockSocket.WriteMessageFunc = func(msgType int, data []byte) error { - select { - case outgoingData <- honeybeetest.MockOutgoingData{ - MsgType: msgType, Data: data}: - case <-mockSocket.Closed: - return io.EOF - } - return nil - } - - conn, err := NewConnectionFromSocket(context.Background(), mockSocket, config, nil) - assert.NoError(t, err) - defer conn.Close() - - err = conn.Send([]byte("test")) - assert.NoError(t, err) - - honeybeetest.Never(t, func() bool { - select { - case <-deadlineCalled: - return true - default: - return false - } - }, "SetWriteDeadline should not be called when timeout is zero") - }) - - t.Run("write timeout sets deadline when positive", func(t *testing.T) { - config := &ConnectionConfig{WriteTimeout: 30 * time.Millisecond, Retry: RetryConfig{Disabled: true}} - - outgoingData := make(chan honeybeetest.MockOutgoingData, 10) - mockSocket := honeybeetest.NewMockSocket() - - mockSocket.CloseFunc = func() error { - mockSocket.Once.Do(func() { - close(mockSocket.Closed) - }) - return nil - } - - deadlineCalled := make(chan struct{}, 1) - mockSocket.SetWriteDeadlineFunc = func(t time.Time) error { - deadlineCalled <- struct{}{} - return nil - } - - mockSocket.WriteMessageFunc = func(msgType int, data []byte) error { - select { - case outgoingData <- honeybeetest.MockOutgoingData{ - MsgType: msgType, Data: data}: - case <-mockSocket.Closed: - return io.EOF - } - return nil - } - - conn, err := NewConnectionFromSocket(context.Background(), mockSocket, config, nil) - assert.NoError(t, err) - defer conn.Close() - - err = conn.Send([]byte("test")) - assert.NoError(t, err) - - honeybeetest.Eventually(t, func() bool { - select { - case <-deadlineCalled: - return true - default: - return false - } - }, "SetWriteDeadline should be called when timeout is positive") - }) - - t.Run("send fails on deadline error", func(t *testing.T) { - config := &ConnectionConfig{WriteTimeout: 1 * time.Millisecond, Retry: RetryConfig{Disabled: true}} - - mockSocket := honeybeetest.NewMockSocket() - - mockSocket.CloseFunc = func() error { - mockSocket.Once.Do(func() { - close(mockSocket.Closed) - }) - return nil - } - - mockSocket.SetWriteDeadlineFunc = func(t time.Time) error { - return fmt.Errorf("test error") - } - - conn, err := NewConnectionFromSocket(context.Background(), mockSocket, config, nil) - assert.NoError(t, err) - defer conn.Close() - - err = conn.Send([]byte("test")) - assert.ErrorIs(t, err, ErrFailedWriteDeadline) - - honeybeetest.Never(t, func() bool { - return conn.State() == StateClosed - }, "write error does not close connection") - }) - - t.Run("send fails on socket write error", func(t *testing.T) { - mockSocket := honeybeetest.NewMockSocket() - - writeErr := fmt.Errorf("test error") - mockSocket.WriteMessageFunc = func(msgType int, data []byte) error { - return writeErr - } - - conn, err := NewConnectionFromSocket(context.Background(), mockSocket, nil, nil) - assert.NoError(t, err) - defer conn.Close() - - err = conn.Send([]byte("test")) - assert.ErrorIs(t, err, ErrWriteFailed) - assert.ErrorContains(t, err, "test error") - }) -} diff --git a/transport/connection_test.go b/transport/connection_test.go index bc7d8c6..1c66f78 100644 --- a/transport/connection_test.go +++ b/transport/connection_test.go @@ -10,134 +10,37 @@ import ( "github.com/gorilla/websocket" "github.com/stretchr/testify/assert" "io" - "net/http" "sync" "sync/atomic" "testing" "time" ) -// Connection state tests +// ---------------------------------------------------------------------------- +// Helpers +// ---------------------------------------------------------------------------- -func TestConnectionStateString(t *testing.T) { - cases := []struct { - state ConnectionState - want string - }{ - {StateDisconnected, "disconnected"}, - {StateConnecting, "connecting"}, - {StateConnected, "connected"}, - {StateClosed, "closed"}, - {ConnectionState(99), "unknown"}, - } +func setupTestConnection(t *testing.T) ( + conn *Connection, + socket *honeybeetest.MockSocket, + incoming chan honeybeetest.MockIncomingData, + outgoing chan honeybeetest.MockOutgoingData, +) { + t.Helper() - for _, tc := range cases { - t.Run(tc.want, func(t *testing.T) { - assert.Equal(t, tc.want, tc.state.String()) - }) - } + socket, incoming, outgoing = honeybeetest.SetupTestSocket(t) + + var err error + conn, err = NewConnection(context.Background(), socket, nil, nil) + assert.NoError(t, err) + return } -func TestConnectionState(t *testing.T) { - // Test initial state - conn, _ := NewConnection(context.Background(), "ws://test", nil, nil) - assert.Equal(t, StateDisconnected, conn.State()) - - // Test state after FromSocket (should be Connected) - conn2, _ := NewConnectionFromSocket(context.Background(), honeybeetest.NewMockSocket(), nil, nil) - assert.Equal(t, StateConnected, conn2.State()) - - // Test state after close - conn.Close() - assert.Equal(t, StateClosed, conn.State()) -} - -// Connection constructor tests +// ---------------------------------------------------------------------------- +// Constructor +// ---------------------------------------------------------------------------- func TestNewConnection(t *testing.T) { - cases := []struct { - name string - url string - config *ConnectionConfig - wantErr bool - wantErrText string - }{ - { - name: "valid url, nil config", - url: "ws://example.com", - config: nil, - }, - { - name: "valid url, valid config", - url: "wss://relay.example.com:8080/path", - config: func() *ConnectionConfig { - c, _ := NewConnectionConfig(WithWriteTimeout(30*time.Second), WithRetryDisabled()) - return c - }(), - }, - { - name: "invalid url", - url: "http://example.com", - config: nil, - wantErr: true, - wantErrText: "URL must use ws:// or wss:// scheme", - }, - { - name: "invalid config", - url: "ws://example.com", - config: &ConnectionConfig{ - Retry: RetryConfig{ - InitialDelay: 10 * time.Second, - MaxDelay: 1 * time.Second, - }, - }, - wantErr: true, - wantErrText: "initial delay may not exceed maximum delay", - }, - } - - for _, tc := range cases { - t.Run(tc.name, func(t *testing.T) { - conn, err := NewConnection(context.Background(), tc.url, tc.config, nil) - - if tc.wantErr { - assert.Error(t, err) - if tc.wantErrText != "" { - assert.ErrorContains(t, err, tc.wantErrText) - } - assert.Nil(t, conn) - return - } - - assert.NoError(t, err) - assert.NotNil(t, conn) - - // Verify struct fields - assert.NotNil(t, conn.url) - assert.NotNil(t, conn.dialer) - assert.Nil(t, conn.socket) - assert.NotNil(t, conn.config) - assert.NotNil(t, conn.incoming) - assert.NotNil(t, conn.errors) - assert.NotNil(t, conn.done) - assert.Equal(t, StateDisconnected, conn.state) - assert.False(t, conn.closed) - - // Verify default config is used if nil is passed - if tc.config == nil { - expected := *GetDefaultConnectionConfig() - expected.Dialer = conn.config.Dialer // dialer resolved at construction - assert.Equal(t, expected, conn.config) - } else { - expected := *tc.config - expected.Dialer = conn.config.Dialer - assert.Equal(t, expected, conn.config) - } - }) - } -} - -func TestNewConnectionFromSocket(t *testing.T) { cases := []struct { name string socket types.Socket @@ -161,30 +64,7 @@ func TestNewConnectionFromSocket(t *testing.T) { name: "valid socket with valid config", socket: honeybeetest.NewMockSocket(), config: func() *ConnectionConfig { - c, _ := NewConnectionConfig(WithWriteTimeout(30*time.Second), WithRetryDisabled()) - return c - }(), - }, - { - name: "invalid config", - socket: honeybeetest.NewMockSocket(), - config: &ConnectionConfig{ - Retry: RetryConfig{ - InitialDelay: 10 * time.Second, - MaxDelay: 1 * time.Second, - }, - }, - wantErr: true, - wantErrText: "initial delay may not exceed maximum delay", - }, - { - name: "close handler set when provided", - socket: honeybeetest.NewMockSocket(), - config: func() *ConnectionConfig { - c, _ := NewConnectionConfig( - WithCloseHandler(func(code int, text string) error { return nil }), - WithRetryDisabled(), - ) + c, _ := NewConnectionConfig(WithWriteTimeout(30 * time.Second)) return c }(), }, @@ -192,22 +72,8 @@ func TestNewConnectionFromSocket(t *testing.T) { for _, tc := range cases { t.Run(tc.name, func(t *testing.T) { - // track if SetCloseHandler was called - closeHandlerSet := false - if tc.socket != nil { - mockSocket := tc.socket.(*honeybeetest.MockSocket) - originalSetCloseHandler := mockSocket.SetCloseHandlerFunc - - // wrapper around the original handler function - mockSocket.SetCloseHandlerFunc = func(h func(int, string) error) { - closeHandlerSet = true - if originalSetCloseHandler != nil { - originalSetCloseHandler(h) - } - } - } - - conn, err := NewConnectionFromSocket(context.Background(), tc.socket, tc.config, nil) + conn, err := NewConnection( + context.Background(), tc.socket, tc.config, nil) if tc.wantErr { assert.Error(t, err) @@ -222,344 +88,32 @@ func TestNewConnectionFromSocket(t *testing.T) { assert.NotNil(t, conn) // Verify fields initialized correctly - assert.Nil(t, conn.url) - assert.Nil(t, conn.dialer) assert.Equal(t, tc.socket, conn.socket) assert.NotNil(t, conn.config) assert.NotNil(t, conn.incoming) assert.NotNil(t, conn.errors) assert.NotNil(t, conn.done) - assert.Equal(t, StateConnected, conn.state) assert.False(t, conn.closed) // Verify default config is used if nil is passed. - // CloseHandler is a func; exclude it from the struct comparison - // (identity is verified separately via closeHandlerSet). gotCfg := conn.config - gotCfg.CloseHandler = nil if tc.config == nil { - expected := *GetDefaultConnectionConfig() - expected.CloseHandler = nil - assert.Equal(t, expected, gotCfg) + expected, _ := NewConnectionConfig() + assert.Equal(t, *expected, gotCfg) } else { expected := *tc.config - expected.CloseHandler = nil assert.Equal(t, expected, gotCfg) } - - // Verify close handler was set if provided - if tc.config != nil && tc.config.CloseHandler != nil { - assert.True(t, closeHandlerSet, "CloseHandler should be set on socket") - } }) } } -func TestConnect(t *testing.T) { - t.Run("connect fails when socket already present", func(t *testing.T) { - conn, err := NewConnection(context.Background(), "ws://test", nil, nil) - assert.NoError(t, err) - - conn.socket = honeybeetest.NewMockSocket() - - err = conn.Connect(context.Background(), nil) - assert.Error(t, err) - assert.ErrorIs(t, err, ErrSocketExists) - assert.Equal(t, StateDisconnected, conn.State()) - }) - - t.Run("connect fails when connection closed", func(t *testing.T) { - conn, err := NewConnection(context.Background(), "ws://test", nil, nil) - assert.NoError(t, err) - - conn.Close() - - err = conn.Connect(context.Background(), nil) - assert.Error(t, err) - assert.ErrorIs(t, err, ErrConnectionClosed) - assert.Equal(t, StateClosed, conn.State()) - }) - - t.Run("connect succeeds and starts goroutines", func(t *testing.T) { - outgoingData := make(chan honeybeetest.MockOutgoingData, 10) - - mockSocket := honeybeetest.NewMockSocket() - mockSocket.WriteMessageFunc = func(msgType int, data []byte) error { - outgoingData <- honeybeetest.MockOutgoingData{MsgType: msgType, Data: data} - return nil - } - - mockDialer := &honeybeetest.MockDialer{ - DialContextFunc: func(context.Context, string, http.Header) (types.Socket, *http.Response, error) { - return mockSocket, nil, nil - }, - } - conf, _ := NewConnectionConfig(WithRetryDisabled(), WithConnectionDialer(mockDialer)) - conn, err := NewConnection(context.Background(), "ws://test", conf, nil) - assert.NoError(t, err) - - err = conn.Connect(context.Background(), nil) - assert.NoError(t, err) - assert.Equal(t, StateConnected, conn.State()) - - testData := []byte("test") - conn.Send(testData) - - honeybeetest.Eventually(t, func() bool { - select { - case msg := <-outgoingData: - return bytes.Equal(msg.Data, testData) - default: - return false - } - }, "expected message") - - conn.Close() - }) - - t.Run("connect retries on dial failure", func(t *testing.T) { - attemptCount := 0 - mockDialer := &honeybeetest.MockDialer{ - DialContextFunc: func(context.Context, string, http.Header) (types.Socket, *http.Response, error) { - attemptCount++ - if attemptCount < 3 { - return nil, nil, fmt.Errorf("dial failed") - } - return honeybeetest.NewMockSocket(), nil, nil - }, - } - config, _ := NewConnectionConfig( - WithRetryMaxRetries(2), - WithRetryInitialDelay(1*time.Millisecond), - WithRetryMaxDelay(5*time.Millisecond), - WithRetryJitterFactor(0.0), - WithConnectionDialer(mockDialer), - ) - conn, err := NewConnection(context.Background(), "ws://test", config, nil) - assert.NoError(t, err) - - err = conn.Connect(context.Background(), nil) - assert.NoError(t, err) - assert.Equal(t, 3, attemptCount) - assert.Equal(t, StateConnected, conn.State()) - - conn.Close() - }) - - t.Run("connect fails after max retries", func(t *testing.T) { - mockDialer := &honeybeetest.MockDialer{ - DialContextFunc: func(context.Context, string, http.Header) (types.Socket, *http.Response, error) { - return nil, nil, fmt.Errorf("dial failed") - }, - } - config, _ := NewConnectionConfig( - WithRetryMaxRetries(2), - WithRetryInitialDelay(1*time.Millisecond), - WithRetryMaxDelay(5*time.Millisecond), - WithRetryJitterFactor(0.0), - WithConnectionDialer(mockDialer), - ) - conn, err := NewConnection(context.Background(), "ws://test", config, nil) - assert.NoError(t, err) - - err = conn.Connect(context.Background(), nil) - assert.Error(t, err) - assert.ErrorContains(t, err, "dial failed") - assert.Equal(t, StateDisconnected, conn.State()) - }) - - t.Run("state transitions during connect", func(t *testing.T) { - stateDuringDial := StateDisconnected - // conn captured after construction; closure safe because dialer runs during Connect - var conn *Connection - mockDialer := &honeybeetest.MockDialer{ - DialContextFunc: func(context.Context, string, http.Header) (types.Socket, *http.Response, error) { - stateDuringDial = conn.state - return honeybeetest.NewMockSocket(), nil, nil - }, - } - var err error - conf, _ := NewConnectionConfig(WithRetryDisabled(), WithConnectionDialer(mockDialer)) - conn, err = NewConnection(context.Background(), "ws://test", conf, nil) - assert.NoError(t, err) - assert.Equal(t, StateDisconnected, conn.State()) - - conn.Connect(context.Background(), nil) - - assert.Equal(t, StateConnecting, stateDuringDial) - assert.Equal(t, StateConnected, conn.State()) - - conn.Close() - }) - - t.Run("close handler configured when provided", func(t *testing.T) { - handlerSet := false - mockSocket := honeybeetest.NewMockSocket() - mockSocket.SetCloseHandlerFunc = func(h func(int, string) error) { - handlerSet = true - } - mockDialer := &honeybeetest.MockDialer{ - DialContextFunc: func(context.Context, string, http.Header) (types.Socket, *http.Response, error) { - return mockSocket, nil, nil - }, - } - config, _ := NewConnectionConfig( - WithCloseHandler(func(code int, text string) error { return nil }), - WithRetryDisabled(), - WithConnectionDialer(mockDialer), - ) - conn, err := NewConnection(context.Background(), "ws://test", config, nil) - assert.NoError(t, err) - - conn.Connect(context.Background(), nil) - - assert.True(t, handlerSet, "close handler should be set on socket") - - conn.Close() - }) - - t.Run("passes headers when configured", func(t *testing.T) { - header := http.Header{"X-Custom": []string{"val"}} - dialCalled := false - mockDialer := &honeybeetest.MockDialer{ - DialContextFunc: func(ctx context.Context, url string, h http.Header) (types.Socket, *http.Response, error) { - assert.Equal(t, "val", h.Get("X-Custom")) - dialCalled = true - return honeybeetest.NewMockSocket(), nil, nil - }, - } - conf, _ := NewConnectionConfig(WithRequestHeader(header), WithConnectionDialer(mockDialer)) - conn, _ := NewConnection(context.Background(), "ws://test", conf, nil) - - err := conn.Connect(context.Background(), nil) - - assert.NoError(t, err) - assert.True(t, dialCalled) - }) - - t.Run("onDialError callback fires on each failed attempt", func(t *testing.T) { - var mu sync.Mutex - var capturedErrors []error - - attemptCount := 0 - dialErr := fmt.Errorf("dial failed") - mockDialer := &honeybeetest.MockDialer{ - DialContextFunc: func(context.Context, string, http.Header) (types.Socket, *http.Response, error) { - attemptCount++ - if attemptCount < 3 { - return nil, nil, dialErr - } - return honeybeetest.NewMockSocket(), nil, nil - }, - } - config, _ := NewConnectionConfig( - WithRetryMaxRetries(3), - WithRetryInitialDelay(1*time.Millisecond), - WithRetryMaxDelay(5*time.Millisecond), - WithRetryJitterFactor(0.0), - WithConnectionDialer(mockDialer), - ) - conn, err := NewConnection(context.Background(), "ws://test", config, nil) - assert.NoError(t, err) - - onDialError := func(e error) { - mu.Lock() - defer mu.Unlock() - capturedErrors = append(capturedErrors, e) - } - - err = conn.Connect(context.Background(), onDialError) - assert.NoError(t, err) - defer conn.Close() - - mu.Lock() - defer mu.Unlock() - assert.Equal(t, []error{dialErr, dialErr}, capturedErrors) - }) - - t.Run("onDialError not called when first dial succeeds", func(t *testing.T) { - callbackCalled := false - mockDialer := &honeybeetest.MockDialer{ - DialContextFunc: func(context.Context, string, http.Header) (types.Socket, *http.Response, error) { - return honeybeetest.NewMockSocket(), nil, nil - }, - } - config, _ := NewConnectionConfig(WithRetryDisabled(), WithConnectionDialer(mockDialer)) - conn, err := NewConnection(context.Background(), "ws://test", config, nil) - assert.NoError(t, err) - - err = conn.Connect(context.Background(), func(error) { callbackCalled = true }) - assert.NoError(t, err) - defer conn.Close() - - assert.False(t, callbackCalled) - }) - - t.Run("nil onDialError does not panic on failed dial", func(t *testing.T) { - mockDialer := &honeybeetest.MockDialer{ - DialContextFunc: func(context.Context, string, http.Header) (types.Socket, *http.Response, error) { - return nil, nil, fmt.Errorf("dial failed") - }, - } - config, _ := NewConnectionConfig(WithRetryDisabled(), WithConnectionDialer(mockDialer)) - conn, err := NewConnection(context.Background(), "ws://test", config, nil) - assert.NoError(t, err) - - assert.NotPanics(t, func() { - conn.Connect(context.Background(), nil) - }) - }) -} - -func TestConnectContextCancellation(t *testing.T) { - t.Run("context cancelled during connect returns before retries exhaust", func(t *testing.T) { - dialCount := atomic.Int32{} - mockDialer := &honeybeetest.MockDialer{ - DialContextFunc: func(ctx context.Context, _ string, _ http.Header) (types.Socket, *http.Response, error) { - dialCount.Add(1) - return nil, nil, fmt.Errorf("dial failed") - }, - } - config, _ := NewConnectionConfig( - WithRetryMaxRetries(100), - WithRetryInitialDelay(500*time.Millisecond), - WithRetryMaxDelay(1*time.Second), - WithRetryJitterFactor(0.0), - WithConnectionDialer(mockDialer), - ) - conn, err := NewConnection(context.Background(), "ws://test", config, nil) - assert.NoError(t, err) - - ctx, cancel := context.WithCancel(context.Background()) - - done := make(chan error, 1) - go func() { - done <- conn.Connect(ctx, nil) - }() - - // wait for first dial - honeybeetest.Eventually(t, func() bool { - return dialCount.Load() >= 1 - }, "expected dial") - cancel() - - select { - case err := <-done: - assert.ErrorIs(t, err, context.Canceled) - - // number of dials is fewer than max retry count - assert.Less(t, dialCount.Load(), int32(100)) - case <-time.After(honeybeetest.TestTimeout): - t.Fatal("Connect did not return after context cancellation") - } - }) -} - -// Connection method tests +// ---------------------------------------------------------------------------- +// Accessors +// ---------------------------------------------------------------------------- func TestConnectionIncoming(t *testing.T) { - conn, err := NewConnection(context.Background(), "ws://test", nil, nil) - assert.NoError(t, err) + conn, _, _, _ := setupTestConnection(t) incoming := conn.Incoming() assert.NotNil(t, incoming) @@ -581,7 +135,7 @@ func TestConnectionErrors(t *testing.T) { } } - conn, err := NewConnectionFromSocket(context.Background(), mockSocket, nil, nil) + conn, err := NewConnection(context.Background(), mockSocket, nil, nil) assert.NoError(t, err) defer conn.Close() @@ -604,7 +158,7 @@ func TestConnectionErrors(t *testing.T) { } } - conn, err := NewConnectionFromSocket(context.Background(), mockSocket, nil, nil) + conn, err := NewConnection(context.Background(), mockSocket, nil, nil) assert.NoError(t, err) defer conn.Close() @@ -624,7 +178,7 @@ func TestConnectionErrors(t *testing.T) { return 0, nil, io.EOF } - conn, err := NewConnectionFromSocket(context.Background(), mockSocket, nil, nil) + conn, err := NewConnection(context.Background(), mockSocket, nil, nil) assert.NoError(t, err) defer conn.Close() @@ -656,7 +210,7 @@ func TestConnectionHeartbeat(t *testing.T) { ) assert.NoError(t, err) - conn, _ := NewConnectionFromSocket(context.Background(), socket, conf, nil) + conn, _ := NewConnection(context.Background(), socket, conf, nil) defer conn.Close() honeybeetest.Eventually(t, @@ -669,7 +223,7 @@ func TestConnectionHeartbeat(t *testing.T) { socket, _, _ := honeybeetest.SetupTestSocket(t) socket.SetPongHandlerFunc = func(h func(string) error) { handler = h } - conn, _ := NewConnectionFromSocket(context.Background(), socket, nil, nil) + conn, _ := NewConnection(context.Background(), socket, nil, nil) defer conn.Close() honeybeetest.Eventually(t, func() bool { @@ -690,20 +244,366 @@ func TestConnectionHeartbeat(t *testing.T) { }) } -// Test helpers +// ---------------------------------------------------------------------------- +// Send +// ---------------------------------------------------------------------------- -func setupTestConnection(t *testing.T) ( - conn *Connection, - socket *honeybeetest.MockSocket, - incoming chan honeybeetest.MockIncomingData, - outgoing chan honeybeetest.MockOutgoingData, -) { - t.Helper() +func TestConnectionSend(t *testing.T) { + t.Run("writes message to socket", func(t *testing.T) { + conn, _, _, outgoingData := setupTestConnection(t) + defer conn.Close() - socket, incoming, outgoing = honeybeetest.SetupTestSocket(t) + testData := []byte("test message") + err := conn.Send(testData) + assert.NoError(t, err) - var err error - conn, err = NewConnectionFromSocket(context.Background(), socket, nil, nil) - assert.NoError(t, err) - return + honeybeetest.ExpectWrite(t, outgoingData, websocket.TextMessage, testData) + }) + + t.Run("writes multiple message to socket", func(t *testing.T) { + conn, _, _, outgoingData := setupTestConnection(t) + defer conn.Close() + + messages := [][]byte{[]byte("first"), []byte("second"), []byte("third")} + for _, msg := range messages { + err := conn.Send(msg) + assert.NoError(t, err) + } + + for _, expected := range messages { + honeybeetest.ExpectWrite(t, outgoingData, websocket.TextMessage, expected) + } + }) + + t.Run("concurrent sends write messages to socket", func(t *testing.T) { + conn, _, _, outgoingData := setupTestConnection(t) + defer conn.Close() + + mu := sync.Mutex{} + messages := []string{} + done := make(chan struct{}) + + go func() { + for { + select { + case msg := <-outgoingData: + mu.Lock() + messages = append(messages, string(msg.Data)) + mu.Unlock() + case <-done: + return + } + } + }() + + defer close(done) + + var wg sync.WaitGroup + for i := range 5 { + wg.Add(1) + go func(id int) { + defer wg.Done() + for j := range 10 { + data := fmt.Appendf(nil, "msg-%d-%d", id, j) + for { + // send and retry until success + err := conn.Send(data) + if err != nil { + continue + } else { + break + } + } + } + }(i) + } + + wg.Wait() + + honeybeetest.Eventually(t, func() bool { + mu.Lock() + defer mu.Unlock() + return len(messages) == 50 + }, "should have received 50 messages") + + }) + + t.Run("send fails when connection is closed", func(t *testing.T) { + conn, _, _, _ := setupTestConnection(t) + conn.Close() + + testData := []byte("test message") + err := conn.Send(testData) + assert.ErrorIs(t, err, ErrConnectionClosed) + }) + + t.Run("write timeout disabled when zero", func(t *testing.T) { + config, _ := NewConnectionConfig( + WithWriteTimeout(0), + ) + + outgoingData := make(chan honeybeetest.MockOutgoingData, 10) + mockSocket := honeybeetest.NewMockSocket() + + mockSocket.CloseFunc = func() error { + mockSocket.Once.Do(func() { + close(mockSocket.Closed) + }) + return nil + } + + deadlineCalled := make(chan struct{}, 1) + mockSocket.SetWriteDeadlineFunc = func(t time.Time) error { + deadlineCalled <- struct{}{} + return nil + } + + mockSocket.WriteMessageFunc = func(msgType int, data []byte) error { + select { + case outgoingData <- honeybeetest.MockOutgoingData{ + MsgType: msgType, Data: data}: + case <-mockSocket.Closed: + return io.EOF + } + return nil + } + + conn, err := NewConnection(context.Background(), mockSocket, config, nil) + assert.NoError(t, err) + defer conn.Close() + + err = conn.Send([]byte("test")) + assert.NoError(t, err) + + honeybeetest.Never(t, func() bool { + select { + case <-deadlineCalled: + return true + default: + return false + } + }, "SetWriteDeadline should not be called when timeout is zero") + }) + + t.Run("write timeout sets deadline when positive", func(t *testing.T) { + config, _ := NewConnectionConfig() + + outgoingData := make(chan honeybeetest.MockOutgoingData, 10) + mockSocket := honeybeetest.NewMockSocket() + + mockSocket.CloseFunc = func() error { + mockSocket.Once.Do(func() { + close(mockSocket.Closed) + }) + return nil + } + + deadlineCalled := make(chan struct{}, 1) + mockSocket.SetWriteDeadlineFunc = func(t time.Time) error { + deadlineCalled <- struct{}{} + return nil + } + + mockSocket.WriteMessageFunc = func(msgType int, data []byte) error { + select { + case outgoingData <- honeybeetest.MockOutgoingData{ + MsgType: msgType, Data: data}: + case <-mockSocket.Closed: + return io.EOF + } + return nil + } + + conn, err := NewConnection(context.Background(), mockSocket, config, nil) + assert.NoError(t, err) + defer conn.Close() + + err = conn.Send([]byte("test")) + assert.NoError(t, err) + + honeybeetest.Eventually(t, func() bool { + select { + case <-deadlineCalled: + return true + default: + return false + } + }, "SetWriteDeadline should be called when timeout is positive") + }) + + t.Run("send fails on deadline error", func(t *testing.T) { + config, _ := NewConnectionConfig(WithWriteTimeout(1 * time.Millisecond)) + + mockSocket := honeybeetest.NewMockSocket() + + mockSocket.CloseFunc = func() error { + mockSocket.Once.Do(func() { + close(mockSocket.Closed) + }) + return nil + } + + mockSocket.SetWriteDeadlineFunc = func(t time.Time) error { + return fmt.Errorf("test error") + } + + conn, err := NewConnection(context.Background(), mockSocket, config, nil) + assert.NoError(t, err) + defer conn.Close() + + err = conn.Send([]byte("test")) + assert.ErrorIs(t, err, ErrFailedWriteDeadline) + + honeybeetest.Never(t, func() bool { + _, ok := <-conn.errors + return !ok + }, "write error does not close connection") + }) + + t.Run("send fails on socket write error", func(t *testing.T) { + mockSocket := honeybeetest.NewMockSocket() + + writeErr := fmt.Errorf("test error") + mockSocket.WriteMessageFunc = func(msgType int, data []byte) error { + return writeErr + } + + conn, err := NewConnection(context.Background(), mockSocket, nil, nil) + assert.NoError(t, err) + defer conn.Close() + + err = conn.Send([]byte("test")) + assert.ErrorIs(t, err, ErrWriteFailed) + assert.ErrorContains(t, err, "test error") + }) +} + +// ---------------------------------------------------------------------------- +// Reader +// ---------------------------------------------------------------------------- + +func TestStartReader(t *testing.T) { + t.Run("text messages route to incoming channel", func(t *testing.T) { + conn, _, incomingData, _ := setupTestConnection(t) + defer conn.Close() + + testData := []byte("hello") + incomingData <- honeybeetest.MockIncomingData{ + MsgType: websocket.TextMessage, + Data: testData, + Err: nil, + } + + honeybeetest.ExpectIncoming(t, conn.Incoming(), testData) + }) + + t.Run("binary messages route to incoming channel", func(t *testing.T) { + conn, _, incomingData, _ := setupTestConnection(t) + defer conn.Close() + + testData := []byte{0x00, 0x01, 0x02} + incomingData <- honeybeetest.MockIncomingData{ + MsgType: websocket.BinaryMessage, + Data: testData, + Err: nil, + } + + honeybeetest.ExpectIncoming(t, conn.Incoming(), testData) + }) + + t.Run("multiple messages processed sequentially", func(t *testing.T) { + conn, _, incomingData, _ := setupTestConnection(t) + defer conn.Close() + + messages := [][]byte{[]byte("first"), []byte("second"), []byte("third")} + for _, msg := range messages { + incomingData <- honeybeetest.MockIncomingData{ + MsgType: websocket.TextMessage, Data: msg, Err: nil} + } + + for _, expected := range messages { + honeybeetest.ExpectIncoming(t, conn.Incoming(), expected) + } + }) + + t.Run("reader exits on socket read error", func(t *testing.T) { + mockSocket := honeybeetest.NewMockSocket() + + mockSocket.CloseFunc = func() error { + mockSocket.Once.Do(func() { + close(mockSocket.Closed) + }) + return nil + } + + mockSocket.ReadMessageFunc = func() (int, []byte, error) { + return 0, nil, io.EOF + } + + conn, err := NewConnection(context.Background(), mockSocket, nil, nil) + assert.NoError(t, err) + + honeybeetest.Eventually(t, func() bool { + _, ok := <-conn.errors + return !ok + }, "expected channel closure") + }) +} + +// ---------------------------------------------------------------------------- +// Close +// ---------------------------------------------------------------------------- + +func TestDisconnectedConnectionClose(t *testing.T) { + t.Run("close is idempotent", func(t *testing.T) { + conn, _, _, _ := setupTestConnection(t) + + conn.Close() + conn.Close() + }) + + t.Run("channels close after close", func(t *testing.T) { + conn, _, _, _ := setupTestConnection(t) + + conn.Close() + + assert.True(t, conn.closed) + _, ok := <-conn.incoming + assert.False(t, ok) + _, ok = <-conn.errors + assert.False(t, ok) + }) + + t.Run("send fails after close", func(t *testing.T) { + conn, _, _, _ := setupTestConnection(t) + + conn.Close() + + err := conn.Send([]byte("test")) + assert.Error(t, err) + assert.ErrorIs(t, err, ErrConnectionClosed) + }) + +} + +func TestConnectedConnectionClose(t *testing.T) { + t.Run("blocked on ReadMessage, unblocks on closed", func(t *testing.T) { + conn, _, incomingData, _ := setupTestConnection(t) + + // Send a message to ensure reader loop is blocking + canary := []byte("canary") + incomingData <- honeybeetest.MockIncomingData{ + MsgType: websocket.TextMessage, Data: canary} + + honeybeetest.Eventually(t, func() bool { + select { + case msg := <-conn.Incoming(): + return bytes.Equal(msg, canary) + default: + return false + } + }, "expected canary message") + + conn.Close() + }) } diff --git a/transport/errors.go b/transport/errors.go index 58c972d..f822681 100644 --- a/transport/errors.go +++ b/transport/errors.go @@ -8,9 +8,10 @@ var ( InvalidProtocol = errors.New("URL must use ws:// or wss:// scheme") // Configuration Errors - InvalidWriteTimeout = errors.New("write timeout cannot be negative") - InvalidPingInterval = errors.New("ping interval cannot be negative") - InvalidBufferSize = errors.New("buffer size must be greater than zero") + InvalidWriteTimeout = errors.New("write timeout cannot be negative") + InvalidPingInterval = errors.New("ping interval cannot be negative") + InvalidBufferSize = errors.New("buffer size must be greater than zero") + InvalidRetryMaxRetries = errors.New("max retry count cannot be negative") InvalidRetryInitialDelay = errors.New("initial delay must be positive") InvalidRetryMaxDelay = errors.New("max delay must be positive") @@ -18,9 +19,7 @@ var ( InvalidDelays = errors.New("initial delay may not exceed maximum delay") // Socket Errors - ErrNilRetryManager = errors.New("retry manager cannot be nil") - ErrNilDialer = errors.New("dialer cannot be nil") - ErrEmptyURL = errors.New("URL cannot be empty") + ErrNilDialFunc = errors.New("dial func cannot be nil") // Connection Errors ErrConnectionClosed = errors.New("connection closed") diff --git a/transport/logging_test.go b/transport/logging_test.go deleted file mode 100644 index 8234f18..0000000 --- a/transport/logging_test.go +++ /dev/null @@ -1,355 +0,0 @@ -package transport - -import ( - "context" - "fmt" - "io" - "log/slog" - "net/http" - "testing" - "time" - // slog used for ExpectedLog level constants - - "git.wisehodl.dev/jay/go-honeybee/honeybeetest" - "git.wisehodl.dev/jay/go-honeybee/types" - "github.com/gorilla/websocket" - "github.com/stretchr/testify/assert" -) - -// Helpers - -func log(level slog.Level, msg string, attrs map[string]any) honeybeetest.ExpectedLog { - return honeybeetest.ExpectedLog{Level: level, Msg: msg, Attrs: attrs} -} - -// Tests - -func TestConnectLogging(t *testing.T) { - t.Run("success", func(t *testing.T) { - mockHandler := honeybeetest.NewMockSlogHandler() - - mockSocket := honeybeetest.NewMockSocket() - mockDialer := &honeybeetest.MockDialer{ - DialContextFunc: func(context.Context, string, http.Header) (types.Socket, *http.Response, error) { - return mockSocket, nil, nil - }, - } - conf, _ := NewConnectionConfig(WithRetryDisabled(), WithConnectionDialer(mockDialer)) - conn, err := NewConnection(context.Background(), "ws://test", conf, mockHandler) - assert.NoError(t, err) - - err = conn.Connect(context.Background(), nil) - assert.NoError(t, err) - defer conn.Close() - - records := mockHandler.GetRecords() - - expected := []honeybeetest.ExpectedLog{ - log(slog.LevelDebug, "connecting", map[string]any{}), - log(slog.LevelDebug, "dialing", map[string]any{"attempt": 1}), - log(slog.LevelDebug, "dial successful", map[string]any{"attempt": 1}), - log(slog.LevelDebug, "connected", map[string]any{}), - } - - honeybeetest.AssertLogSequence(t, records, expected) - }) - - t.Run("max retries failure", func(t *testing.T) { - mockHandler := honeybeetest.NewMockSlogHandler() - - dialErr := fmt.Errorf("dial error") - mockDialer := &honeybeetest.MockDialer{ - DialContextFunc: func(context.Context, string, http.Header) (types.Socket, *http.Response, error) { - return nil, nil, dialErr - }, - } - config, _ := NewConnectionConfig( - WithRetryMaxRetries(2), - WithRetryInitialDelay(1*time.Millisecond), - WithRetryMaxDelay(5*time.Millisecond), - WithRetryJitterFactor(0.0), - WithConnectionDialer(mockDialer), - ) - - conn, err := NewConnection(context.Background(), "ws://test", config, mockHandler) - assert.NoError(t, err) - - err = conn.Connect(context.Background(), nil) - assert.Error(t, err) - - records := mockHandler.GetRecords() - - expected := []honeybeetest.ExpectedLog{ - log(slog.LevelDebug, "connecting", map[string]any{}), - log(slog.LevelDebug, "dialing", map[string]any{"attempt": 1}), - log(slog.LevelWarn, "dial failed, retrying", map[string]any{"attempt": 1, "error": dialErr}), - log(slog.LevelDebug, "dialing", map[string]any{"attempt": 2}), - log(slog.LevelWarn, "dial failed, retrying", map[string]any{"attempt": 2, "error": dialErr}), - log(slog.LevelDebug, "dialing", map[string]any{"attempt": 3}), - log(slog.LevelDebug, "dial failed, max retries reached", map[string]any{"attempt": 3, "error": dialErr}), - log(slog.LevelWarn, "connection failed", map[string]any{"error": dialErr}), - } - - honeybeetest.AssertLogSequence(t, records, expected) - }) - - t.Run("success after retry", func(t *testing.T) { - mockHandler := honeybeetest.NewMockSlogHandler() - - attemptCount := 0 - dialErr := fmt.Errorf("dial error") - mockDialer := &honeybeetest.MockDialer{ - DialContextFunc: func(context.Context, string, http.Header) (types.Socket, *http.Response, error) { - attemptCount++ - if attemptCount < 3 { - return nil, nil, dialErr - } - return honeybeetest.NewMockSocket(), nil, nil - }, - } - config, _ := NewConnectionConfig( - WithRetryMaxRetries(3), - WithRetryInitialDelay(1*time.Millisecond), - WithRetryMaxDelay(5*time.Millisecond), - WithRetryJitterFactor(0.0), - WithConnectionDialer(mockDialer), - ) - - conn, err := NewConnection(context.Background(), "ws://test", config, mockHandler) - assert.NoError(t, err) - - err = conn.Connect(context.Background(), nil) - assert.NoError(t, err) - defer conn.Close() - - records := mockHandler.GetRecords() - - expected := []honeybeetest.ExpectedLog{ - log(slog.LevelDebug, "connecting", map[string]any{}), - log(slog.LevelDebug, "dialing", map[string]any{"attempt": 1}), - log(slog.LevelWarn, "dial failed, retrying", map[string]any{"attempt": 1, "error": dialErr}), - log(slog.LevelDebug, "dialing", map[string]any{"attempt": 2}), - log(slog.LevelWarn, "dial failed, retrying", map[string]any{"attempt": 2, "error": dialErr}), - log(slog.LevelDebug, "dialing", map[string]any{"attempt": 3}), - log(slog.LevelDebug, "dial successful", map[string]any{"attempt": 3}), - log(slog.LevelDebug, "connected", map[string]any{}), - } - - honeybeetest.AssertLogSequence(t, records, expected) - }) -} - -func TestCloseLogging(t *testing.T) { - t.Run("normal close", func(t *testing.T) { - mockHandler := honeybeetest.NewMockSlogHandler() - - mockSocket := honeybeetest.NewMockSocket() - conn, err := NewConnectionFromSocket(context.Background(), mockSocket, nil, mockHandler) - assert.NoError(t, err) - - conn.Close() - - honeybeetest.Eventually(t, func() bool { - return honeybeetest.FindLogRecord( - mockHandler.GetRecords(), slog.LevelDebug, "closed") != nil - }, "expected log") - - records := mockHandler.GetRecords() - - expected := []honeybeetest.ExpectedLog{ - log(slog.LevelDebug, "closing", map[string]any{}), - log(slog.LevelDebug, "closed", map[string]any{}), - } - - honeybeetest.AssertLogSequence(t, records, expected) - }) - - t.Run("close with socket error", func(t *testing.T) { - mockHandler := honeybeetest.NewMockSlogHandler() - - closeErr := fmt.Errorf("close error") - mockSocket := honeybeetest.NewMockSocket() - mockSocket.CloseFunc = func() error { - return closeErr - } - - conn, err := NewConnectionFromSocket(context.Background(), mockSocket, nil, mockHandler) - assert.NoError(t, err) - - conn.Close() - - honeybeetest.Eventually(t, func() bool { - return honeybeetest.FindLogRecord( - mockHandler.GetRecords(), slog.LevelError, "socket close failed") != nil - }, "expected log") - - records := mockHandler.GetRecords() - - expected := []honeybeetest.ExpectedLog{ - log(slog.LevelDebug, "closing", map[string]any{}), - log(slog.LevelError, "socket close failed", map[string]any{"error": closeErr}), - } - - honeybeetest.AssertLogSequence(t, records, expected) - }) -} - -func TestReaderLogging(t *testing.T) { - t.Run("clean close by peer", func(t *testing.T) { - mockHandler := honeybeetest.NewMockSlogHandler() - - mockSocket := honeybeetest.NewMockSocket() - mockSocket.ReadMessageFunc = func() (int, []byte, error) { - return 0, nil, &websocket.CloseError{ - Code: websocket.CloseNormalClosure, - Text: "goodbye", - } - } - - conn, err := NewConnectionFromSocket(context.Background(), mockSocket, nil, mockHandler) - assert.NoError(t, err) - defer conn.Close() - - honeybeetest.Eventually(t, func() bool { - return honeybeetest.FindLogRecord( - mockHandler.GetRecords(), slog.LevelDebug, "connection closed by peer") != nil - }, "expected log") - - record := honeybeetest.FindLogRecord(mockHandler.GetRecords(), slog.LevelDebug, "connection closed by peer") - assert.NotNil(t, record) - honeybeetest.AssertAttributePresent(t, *record, "code", websocket.CloseNormalClosure) - honeybeetest.AssertAttributePresent(t, *record, "text", "goodbye") - - }) - - t.Run("unexpected close", func(t *testing.T) { - mockHandler := honeybeetest.NewMockSlogHandler() - - mockSocket := honeybeetest.NewMockSocket() - mockSocket.ReadMessageFunc = func() (int, []byte, error) { - return 0, nil, &websocket.CloseError{ - Code: websocket.CloseProtocolError, - Text: "bad protocol", - } - } - - conn, err := NewConnectionFromSocket(context.Background(), mockSocket, nil, mockHandler) - assert.NoError(t, err) - defer conn.Close() - - honeybeetest.Eventually(t, func() bool { - return honeybeetest.FindLogRecord( - mockHandler.GetRecords(), slog.LevelWarn, "unexpected close") != nil - }, "expected log") - - record := honeybeetest.FindLogRecord(mockHandler.GetRecords(), slog.LevelWarn, "unexpected close") - assert.NotNil(t, record) - honeybeetest.AssertAttributePresent(t, *record, "code", websocket.CloseProtocolError) - honeybeetest.AssertAttributePresent(t, *record, "text", "bad protocol") - - }) - - t.Run("read error", func(t *testing.T) { - mockHandler := honeybeetest.NewMockSlogHandler() - - mockSocket := honeybeetest.NewMockSocket() - mockSocket.ReadMessageFunc = func() (int, []byte, error) { - return 0, nil, io.EOF - } - - conn, err := NewConnectionFromSocket(context.Background(), mockSocket, nil, mockHandler) - assert.NoError(t, err) - defer conn.Close() - - honeybeetest.Eventually(t, func() bool { - return honeybeetest.FindLogRecord( - mockHandler.GetRecords(), slog.LevelError, "read error") != nil - }, "expected log") - }) -} - -func TestWriterLogging(t *testing.T) { - t.Run("write deadline error", func(t *testing.T) { - mockHandler := honeybeetest.NewMockSlogHandler() - - config, _ := NewConnectionConfig(WithWriteTimeout(1*time.Millisecond), WithRetryDisabled()) - - deadlineErr := fmt.Errorf("deadline error") - mockSocket := honeybeetest.NewMockSocket() - mockSocket.SetWriteDeadlineFunc = func(time.Time) error { - return deadlineErr - } - - conn, err := NewConnectionFromSocket(context.Background(), mockSocket, config, mockHandler) - assert.NoError(t, err) - - err = conn.Send([]byte("test")) - assert.ErrorContains(t, err, "failed to set write deadline: deadline error") - - honeybeetest.Eventually(t, func() bool { - return honeybeetest.FindLogRecord( - mockHandler.GetRecords(), slog.LevelError, "write deadline error") != nil - }, "expected log") - - records := mockHandler.GetRecords() - - record := honeybeetest.FindLogRecord(records, slog.LevelError, "write deadline error") - assert.NotNil(t, record) - honeybeetest.AssertAttributePresent(t, *record, "error", deadlineErr) - - conn.Close() - }) - - t.Run("write message error", func(t *testing.T) { - mockHandler := honeybeetest.NewMockSlogHandler() - - writeErr := fmt.Errorf("write error") - mockSocket := honeybeetest.NewMockSocket() - mockSocket.WriteMessageFunc = func(int, []byte) error { - return writeErr - } - - conn, err := NewConnectionFromSocket(context.Background(), mockSocket, nil, mockHandler) - assert.NoError(t, err) - - err = conn.Send([]byte("test")) - assert.ErrorContains(t, err, "write error") - - honeybeetest.Eventually(t, func() bool { - return honeybeetest.FindLogRecord( - mockHandler.GetRecords(), slog.LevelError, "write error") != nil - }, "expected log") - - records := mockHandler.GetRecords() - - record := honeybeetest.FindLogRecord(records, slog.LevelError, "write error") - assert.NotNil(t, record) - honeybeetest.AssertAttributePresent(t, *record, "error", writeErr) - - conn.Close() - }) -} - -func TestLoggingDisabled(t *testing.T) { - t.Run("nil logger produces no logs", func(t *testing.T) { - mockHandler := honeybeetest.NewMockSlogHandler() - - mockSocket := honeybeetest.NewMockSocket() - mockDialer := &honeybeetest.MockDialer{ - DialContextFunc: func(context.Context, string, http.Header) (types.Socket, *http.Response, error) { - return mockSocket, nil, nil - }, - } - conf, _ := NewConnectionConfig(WithRetryDisabled(), WithConnectionDialer(mockDialer)) - conn, err := NewConnection(context.Background(), "ws://test", conf, nil) - assert.NoError(t, err) - - err = conn.Connect(context.Background(), nil) - assert.NoError(t, err) - - conn.Close() - - records := mockHandler.GetRecords() - assert.Empty(t, records) - }) -} diff --git a/transport/retry.go b/transport/retry.go index ecfd7a9..d72d963 100644 --- a/transport/retry.go +++ b/transport/retry.go @@ -1,11 +1,97 @@ package transport import ( + "fmt" "math" "math/rand" "time" ) +// ---------------------------------------------------------------------------- +// Config +// ---------------------------------------------------------------------------- + +type RetryConfig struct { + MaxRetries int + InitialDelay time.Duration + MaxDelay time.Duration + JitterFactor float64 +} + +func NewRetryConfig(opts ...RetryOption) (RetryConfig, error) { + conf := RetryConfig{ + MaxRetries: 0, // Infinite retries + InitialDelay: 1 * time.Second, + MaxDelay: 60 * time.Second, + JitterFactor: 0.2, + } + for _, o := range opts { + o(&conf) + } + + err := ValidateRetryConfig(conf) + if err != nil { + return RetryConfig{}, err + } + + return conf, nil +} + +type RetryOption func(*RetryConfig) + +func WithMaxRetries(value int) RetryOption { + return func(c *RetryConfig) { + c.MaxRetries = value + } +} + +func WithInitialDelay(value time.Duration) RetryOption { + return func(c *RetryConfig) { + c.InitialDelay = value + } +} + +func WithMaxDelay(value time.Duration) RetryOption { + return func(c *RetryConfig) { + c.MaxDelay = value + } +} + +func WithJitterFactor(value float64) RetryOption { + return func(c *RetryConfig) { + c.JitterFactor = value + } +} + +func ValidateRetryConfig(c RetryConfig) error { + if c.MaxRetries < 0 { + return fmt.Errorf("invalid max retry count: %d", c.MaxRetries) + } + + if c.InitialDelay <= 0 { + return fmt.Errorf("invalid initial delay: %v", c.InitialDelay) + } + + if c.MaxDelay <= 0 { + return fmt.Errorf("invalid max delay: %v", c.MaxDelay) + } + + if c.JitterFactor < 0.0 || c.JitterFactor > 1.0 { + return fmt.Errorf("invalid jitter factor: %f", c.JitterFactor) + } + + if c.InitialDelay > c.MaxDelay { + return fmt.Errorf("Initial delay %v cannot exceed max delay %v", + c.InitialDelay, c.MaxDelay) + } + + return nil +} + +// ---------------------------------------------------------------------------- +// Retry manager +// ---------------------------------------------------------------------------- + type RetryManager struct { config RetryConfig retryCount int @@ -16,8 +102,7 @@ func NewRetryManager(config RetryConfig) *RetryManager { // saturationCount: retry count at which base delay meets or exceeds MaxDelay. // Conservative by two to preserve jitter variance near the boundary. saturation := 0 - if !config.Disabled && - config.InitialDelay > 0 && + if config.InitialDelay > 0 && config.InitialDelay <= config.MaxDelay { ratio := float64(config.MaxDelay) / float64(config.InitialDelay) saturation = int(math.Ceil(math.Log2(ratio))) + 2 @@ -31,10 +116,6 @@ func NewRetryManager(config RetryConfig) *RetryManager { } func (r *RetryManager) ShouldRetry() bool { - if r.config.Disabled { - return false - } - if r.config.MaxRetries > 0 && r.retryCount >= r.config.MaxRetries { return false } @@ -43,10 +124,6 @@ func (r *RetryManager) ShouldRetry() bool { } func (r *RetryManager) CalculateDelay() time.Duration { - if r.config.Disabled { - return time.Second - } - // First attempt: immediate retry if r.retryCount == 0 { return 0 diff --git a/transport/retry_test.go b/transport/retry_test.go index 0203642..3dd3dd6 100644 --- a/transport/retry_test.go +++ b/transport/retry_test.go @@ -7,23 +7,17 @@ import ( ) func TestNewRetryManager(t *testing.T) { - config := RetryConfig{ - MaxRetries: 0, - } + config, _ := NewRetryConfig(WithMaxRetries(0)) mgr := NewRetryManager(config) assert.Equal(t, config, mgr.config) assert.Equal(t, 0, mgr.retryCount) - - // Should accept a disabled config - mgr = NewRetryManager(RetryConfig{Disabled: true}) - assert.True(t, mgr.config.Disabled) - assert.Equal(t, 0, mgr.retryCount) } func TestRecordRetry(t *testing.T) { - mgr := NewRetryManager(RetryConfig{Disabled: true}) + config, _ := NewRetryConfig(WithMaxRetries(0)) + mgr := NewRetryManager(config) assert.Equal(t, mgr.retryCount, 0) mgr.RecordRetry() @@ -34,44 +28,31 @@ func TestRecordRetry(t *testing.T) { } func TestShouldRetry(t *testing.T) { - // never retry if config is disabled - mgr := NewRetryManager(RetryConfig{Disabled: true}) - assert.False(t, mgr.ShouldRetry()) - // always retry if max attempt count is zero - mgr = &RetryManager{ - config: RetryConfig{ - MaxRetries: 0, - }, + config, _ := NewRetryConfig(WithMaxRetries(0)) + mgr := &RetryManager{ + config: config, retryCount: 1000, } assert.True(t, mgr.ShouldRetry()) // retry if below max attempt count + config, _ = NewRetryConfig(WithMaxRetries(10)) mgr = &RetryManager{ - config: RetryConfig{ - MaxRetries: 10, - }, + config: config, retryCount: 5, } assert.True(t, mgr.ShouldRetry()) // do not retry if above max attempt count + config, _ = NewRetryConfig(WithMaxRetries(10)) mgr = &RetryManager{ - config: RetryConfig{ - MaxRetries: 10, - }, + config: config, retryCount: 11, } assert.False(t, mgr.ShouldRetry()) } -func TestCalculateDelayDisabled(t *testing.T) { - // default delay if retry is disabled - mgr := NewRetryManager(RetryConfig{Disabled: true}) - assert.Equal(t, time.Second, mgr.CalculateDelay()) -} - func TestCalculateDelayWithoutJitter(t *testing.T) { mgr := NewRetryManager(RetryConfig{ MaxRetries: 0, diff --git a/transport/socket.go b/transport/socket.go index 1304316..d94679a 100644 --- a/transport/socket.go +++ b/transport/socket.go @@ -8,9 +8,14 @@ import ( "time" "git.wisehodl.dev/jay/go-honeybee/types" + "git.wisehodl.dev/jay/go-mana-component" "github.com/gorilla/websocket" ) +// ---------------------------------------------------------------------------- +// Types +// ---------------------------------------------------------------------------- + func NewDialer() types.Dialer { return NewGorillaDialer() } @@ -42,65 +47,79 @@ func (d *GorillaDialer) DialContext( return conn, resp, err } +// ---------------------------------------------------------------------------- +// Retry Dialer +// ---------------------------------------------------------------------------- + func AcquireSocket( ctx context.Context, - retryMgr *RetryManager, - dialer types.Dialer, - url string, - header http.Header, - logger *slog.Logger, - onDialError func(error), -) (types.Socket, *http.Response, error) { - select { - case <-ctx.Done(): - return nil, nil, ctx.Err() - default: + mgr *RetryManager, + dialFn func(ctx context.Context) (types.Socket, error), + errCh chan<- error, + handler slog.Handler, +) (types.Socket, error) { + if component.FromContext(ctx) == nil { + ctx = component.MustNew(ctx, "honeybee", "dialer") + } else { + ctx = component.MustExtend(ctx, "dialer") } - if retryMgr == nil { - return nil, nil, NewConnectionError(ErrNilRetryManager) + var logger *slog.Logger + if handler != nil { + comp := component.FromContext(ctx) + logger = slog.New(handler).With("component", comp) } - if dialer == nil { - return nil, nil, NewConnectionError(ErrNilDialer) - } - if url == "" { - return nil, nil, NewConnectionError(ErrEmptyURL) + + if dialFn == nil { + return nil, ErrNilDialFunc } for { if logger != nil { - logger.Debug("dialing", "attempt", retryMgr.RetryCount()+1) + logger.Debug("dialing", "attempt", mgr.RetryCount()+1) } // dial - socket, resp, err := dialer.DialContext(ctx, url, header) + socket, err := dialFn(ctx) if err == nil { if logger != nil { - logger.Debug("dial successful", "attempt", retryMgr.RetryCount()+1) + logger.Debug("dial successful", "attempt", mgr.RetryCount()+1) } - return socket, resp, nil + return socket, nil + } + + if errCh != nil { + select { + case <-ctx.Done(): + return nil, ctx.Err() + case errCh <- err: + } + } + + if mgr == nil { + if logger != nil { + logger.Debug("dial failed, retry disabled", "error", err) + } + return nil, err } // dial failed, retry - if onDialError != nil { - onDialError(err) - } - if !retryMgr.ShouldRetry() { + if !mgr.ShouldRetry() { // retry policy expired if logger != nil { logger.Debug("dial failed, max retries reached", "error", err, - "attempt", retryMgr.RetryCount()+1) + "attempt", mgr.RetryCount()+1) } - return nil, nil, err + return nil, err } - delay := retryMgr.CalculateDelay() + delay := mgr.CalculateDelay() if logger != nil { logger.Warn("dial failed, retrying", "error", err, - "attempt", retryMgr.RetryCount()+1, + "attempt", mgr.RetryCount()+1, "next_delay", delay) } @@ -108,8 +127,8 @@ func AcquireSocket( select { case <-time.After(delay): case <-ctx.Done(): - return nil, nil, ctx.Err() + return nil, ctx.Err() } - retryMgr.RecordRetry() + mgr.RecordRetry() } } diff --git a/transport/socket_test.go b/transport/socket_test.go index c0bcf5b..4d59323 100644 --- a/transport/socket_test.go +++ b/transport/socket_test.go @@ -7,8 +7,6 @@ import ( "git.wisehodl.dev/jay/go-honeybee/honeybeetest" "git.wisehodl.dev/jay/go-honeybee/types" "github.com/stretchr/testify/assert" - "net/http" - "sync" "sync/atomic" "testing" "time" @@ -46,15 +44,22 @@ func TestAcquireSocket(t *testing.T) { wantErr: false, }, { - name: "two failures, success", - mockRuns: []error{errors.New("1"), errors.New("2"), nil}, + name: "two failures, success", + mockRuns: []error{ + errors.New("1"), + errors.New("2"), + nil}, maxRetries: 0, wantRetryCount: 2, wantErr: false, }, { - name: "three failures, failure", - mockRuns: []error{errors.New("1"), errors.New("2"), errors.New("3"), errors.New("4")}, + name: "three failures, failure", + mockRuns: []error{ + errors.New("1"), + errors.New("2"), + errors.New("3"), + errors.New("4")}, maxRetries: 3, wantRetryCount: 3, wantErr: true, @@ -64,16 +69,13 @@ func TestAcquireSocket(t *testing.T) { for _, tc := range cases { t.Run(tc.name, func(t *testing.T) { attemptIndex := atomic.Int32{} - mockDialer := &honeybeetest.MockDialer{ - DialContextFunc: func(context.Context, string, http.Header, - ) (types.Socket, *http.Response, error) { - err := tc.mockRuns[attemptIndex.Load()] - attemptIndex.Add(1) - if err != nil { - return nil, nil, err - } - return honeybeetest.NewMockSocket(), nil, nil - }, + dial := func(ctx context.Context) (types.Socket, error) { + err := tc.mockRuns[attemptIndex.Load()] + attemptIndex.Add(1) + if err != nil { + return nil, err + } + return honeybeetest.NewMockSocket(), nil } retryMgr := NewRetryManager(RetryConfig{ @@ -83,8 +85,8 @@ func TestAcquireSocket(t *testing.T) { JitterFactor: 0.0, }) - socket, _, err := AcquireSocket( - context.Background(), retryMgr, mockDialer, "ws://test", nil, nil, nil) + socket, err := AcquireSocket( + context.Background(), retryMgr, dial, nil, nil) assert.Equal(t, tc.wantRetryCount, retryMgr.RetryCount()) if tc.wantErr { @@ -96,249 +98,207 @@ func TestAcquireSocket(t *testing.T) { } }) } + + t.Run("no retry, no errors channel", func(t *testing.T) { + dial := func(ctx context.Context) (types.Socket, error) { + return nil, errors.New("dial failed") + } + _, err := AcquireSocket(context.Background(), nil, dial, nil, nil) + assert.Error(t, err) + }) +} + +func TestAcquireSocketNoRetry(t *testing.T) { + t.Run("success", func(t *testing.T) { + dial := func(ctx context.Context) (types.Socket, error) { + return honeybeetest.NewMockSocket(), nil + } + socket, err := AcquireSocket(context.Background(), nil, dial, nil, nil) + assert.NotNil(t, socket) + assert.NoError(t, err) + }) + + t.Run("error", func(t *testing.T) { + dial := func(ctx context.Context) (types.Socket, error) { + return nil, errors.New("dial failed") + } + _, err := AcquireSocket(context.Background(), nil, dial, nil, nil) + assert.Error(t, err) + }) } func TestAcquireSocketGuards(t *testing.T) { - validDialer := &honeybeetest.MockDialer{ - DialContextFunc: func(context.Context, string, http.Header, - ) (types.Socket, *http.Response, error) { - return honeybeetest.NewMockSocket(), nil, nil - }, - } - validRetryConfig := GetDefaultConnectionConfig().Retry + validRetryConfig, _ := NewRetryConfig() validRetryMgr := NewRetryManager(validRetryConfig) cases := []struct { name string retryMgr *RetryManager - dialer types.Dialer - url string - wantErr string + dialFn func(ctx context.Context) (types.Socket, error) + wantErr error }{ { - name: "nil retry manager", - retryMgr: nil, - dialer: validDialer, - url: "ws://test", - wantErr: "retry manager cannot be nil", - }, - { - name: "nil dialer", + name: "nil dial func", retryMgr: validRetryMgr, - dialer: nil, - url: "ws://test", - wantErr: "dialer cannot be nil", - }, - { - name: "empty URL", - retryMgr: validRetryMgr, - dialer: validDialer, - url: "", - wantErr: "URL cannot be empty", + wantErr: ErrNilDialFunc, }, } for _, tc := range cases { t.Run(tc.name, func(t *testing.T) { - socket, resp, err := AcquireSocket( - context.Background(), tc.retryMgr, tc.dialer, tc.url, nil, nil, nil) + socket, err := AcquireSocket( + context.Background(), tc.retryMgr, tc.dialFn, nil, nil) assert.Error(t, err) - assert.ErrorContains(t, err, tc.wantErr) + assert.ErrorIs(t, err, tc.wantErr) assert.Nil(t, socket) - assert.Nil(t, resp) }) } } func TestAcquireSocketContextCancellation(t *testing.T) { - t.Run("already-canceled context returns immediately without dialing", - func(t *testing.T) { - dialCalled := atomic.Bool{} - mockDialer := &honeybeetest.MockDialer{ - DialContextFunc: func(ctx context.Context, _ string, _ http.Header) (types.Socket, *http.Response, error) { - dialCalled.Store(true) - return honeybeetest.NewMockSocket(), nil, nil - }, + t.Run("context cancelled during sleep returns before next attempt", func(t *testing.T) { + dialCount := atomic.Int32{} + dial := func(ctx context.Context) (types.Socket, error) { + dialCount.Add(1) + select { + case <-ctx.Done(): + return nil, ctx.Err() + default: } + return nil, fmt.Errorf("dial failed") + } - ctx, cancel := context.WithCancel(context.Background()) + ctx, cancel := context.WithCancel(context.Background()) - // cancel before acquiring socket - cancel() + retryCfg, _ := NewRetryConfig( + WithMaxRetries(10), + WithInitialDelay(time.Second), + WithMaxDelay(time.Second), + WithJitterFactor(0.0), + ) + retryMgr := NewRetryManager(retryCfg) - retryMgr := NewRetryManager(GetDefaultConnectionConfig().Retry) - _, _, err := AcquireSocket(ctx, retryMgr, mockDialer, "ws://test", nil, nil, nil) + done := make(chan error, 1) + go func() { + _, err := AcquireSocket(ctx, retryMgr, dial, nil, nil) + done <- err + }() + // wait for first two dials to complete, then cancel during sleep + honeybeetest.Eventually(t, func() bool { + return dialCount.Load() > 1 + }, "expected dials") + cancel() + + select { + case err := <-done: assert.ErrorIs(t, err, context.Canceled) - assert.False(t, dialCalled.Load()) - }) - t.Run("context cancelled during sleep returns before next attempt", - func(t *testing.T) { - dialCount := atomic.Int32{} - mockDialer := &honeybeetest.MockDialer{ - DialContextFunc: func(_ context.Context, _ string, _ http.Header) (types.Socket, *http.Response, error) { - dialCount.Add(1) - return nil, nil, fmt.Errorf("dial failed") - }, - } - - ctx, cancel := context.WithCancel(context.Background()) - - retryMgr := NewRetryManager(RetryConfig{ - MaxRetries: 10, - InitialDelay: 1 * time.Second, - MaxDelay: 1 * time.Second, - JitterFactor: 0.0, - }) - - done := make(chan error, 1) - go func() { - _, _, err := AcquireSocket(ctx, retryMgr, mockDialer, "ws://test", nil, nil, nil) - done <- err - }() - - // wait for first two dials to complete, then cancel during sleep - honeybeetest.Eventually(t, func() bool { - return dialCount.Load() > 1 - }, "expected dials") - cancel() + // dial count is 2 because the first retry is always immediate + assert.Equal(t, int32(2), dialCount.Load()) + case <-time.After(honeybeetest.TestTimeout): + t.Fatal("AcquireSocket did not return after context cancellation") + } + }) + t.Run("context cancelled during in-progress dial unblocks and returns", func(t *testing.T) { + dial := func(ctx context.Context) (types.Socket, error) { + // block until context is cancelled select { - case err := <-done: - assert.ErrorIs(t, err, context.Canceled) - - // dial count is 2 because the first retry is always immediate - assert.Equal(t, int32(2), dialCount.Load()) - case <-time.After(honeybeetest.TestTimeout): - t.Fatal("AcquireSocket did not return after context cancellation") + case <-ctx.Done(): + return nil, ctx.Err() } - }) + } - t.Run("context cancelled during in-progress dial unblocks and returns", - func(t *testing.T) { - mockDialer := &honeybeetest.MockDialer{ - DialContextFunc: func(ctx context.Context, _ string, _ http.Header) (types.Socket, *http.Response, error) { - // block until context is cancelled - select { - case <-ctx.Done(): - return nil, nil, ctx.Err() - } - }, - } + ctx, cancel := context.WithCancel(context.Background()) - ctx, cancel := context.WithCancel(context.Background()) + retryCfg, _ := NewRetryConfig() + retryMgr := NewRetryManager(retryCfg) + done := make(chan error, 1) + go func() { + _, err := AcquireSocket(ctx, retryMgr, dial, nil, nil) + done <- err + }() - retryMgr := NewRetryManager(GetDefaultConnectionConfig().Retry) - done := make(chan error, 1) - go func() { - _, _, err := AcquireSocket(ctx, retryMgr, mockDialer, "ws://test", nil, nil, nil) - done <- err - }() + // wait for dialer to block + time.Sleep(20 * time.Millisecond) + cancel() - // wait for dialer to block - time.Sleep(20 * time.Millisecond) - cancel() + select { + case err := <-done: + assert.ErrorIs(t, err, context.Canceled) + case <-time.After(honeybeetest.TestTimeout): + t.Fatal("AcquireSocket did not return after context cancellation") + } - select { - case err := <-done: - assert.ErrorIs(t, err, context.Canceled) - case <-time.After(honeybeetest.TestTimeout): - t.Fatal("AcquireSocket did not return after context cancellation") - } - - }) + }) } -func TestAcquireSocketOnDialError(t *testing.T) { - t.Run("callback fires once per failed attempt with exact error", func(t *testing.T) { - var mu sync.Mutex - var capturedErrors []error - +func TestAcquireSocketDialErrors(t *testing.T) { + t.Run("receive one error per failed dial", func(t *testing.T) { dialErr1 := errors.New("attempt 1 failed") dialErr2 := errors.New("attempt 2 failed") attemptIndex := atomic.Int32{} - mockDialer := &honeybeetest.MockDialer{ - DialContextFunc: func(context.Context, string, http.Header) (types.Socket, *http.Response, error) { - i := attemptIndex.Add(1) - switch i { - case 1: - return nil, nil, dialErr1 - case 2: - return nil, nil, dialErr2 - default: - return honeybeetest.NewMockSocket(), nil, nil - } - }, + dial := func(ctx context.Context) (types.Socket, error) { + i := attemptIndex.Add(1) + switch i { + case 1: + return nil, dialErr1 + case 2: + return nil, dialErr2 + default: + return honeybeetest.NewMockSocket(), nil + } } - onDialError := func(err error) { - mu.Lock() - defer mu.Unlock() - capturedErrors = append(capturedErrors, err) - } + errCh := make(chan error, 2) - retryMgr := NewRetryManager(RetryConfig{ - MaxRetries: 3, - InitialDelay: 1 * time.Millisecond, - MaxDelay: 5 * time.Millisecond, - JitterFactor: 0.0, - }) - _, _, err := AcquireSocket(context.Background(), retryMgr, mockDialer, "ws://test", nil, nil, onDialError) + retryCfg, _ := NewRetryConfig( + WithMaxRetries(3), + WithInitialDelay(1*time.Millisecond), + WithMaxDelay(5*time.Millisecond), + WithJitterFactor(0.0), + ) + retryMgr := NewRetryManager(retryCfg) - assert.NoError(t, err) - mu.Lock() - defer mu.Unlock() - assert.Equal(t, []error{dialErr1, dialErr2}, capturedErrors) + go func() { + _, err := AcquireSocket(context.Background(), retryMgr, dial, errCh, nil) + assert.NoError(t, err) + }() + + var gotErrors []error + honeybeetest.Eventually(t, func() bool { + select { + case err := <-errCh: + gotErrors = append(gotErrors, err) + default: + } + return len(gotErrors) == 2 + }, "expected errors") + + assert.Equal(t, dialErr1, gotErrors[0]) + assert.Equal(t, dialErr2, gotErrors[1]) }) - t.Run("callback not called on successful first dial", func(t *testing.T) { - callbackCalled := atomic.Bool{} - mockDialer := &honeybeetest.MockDialer{ - DialContextFunc: func(context.Context, string, http.Header) (types.Socket, *http.Response, error) { - return honeybeetest.NewMockSocket(), nil, nil - }, + t.Run("no errors on successful first dial", func(t *testing.T) { + dial := func(ctx context.Context) (types.Socket, error) { + return honeybeetest.NewMockSocket(), nil } - retryMgr := NewRetryManager(RetryConfig{MaxRetries: 3, InitialDelay: 1 * time.Millisecond, MaxDelay: 5 * time.Millisecond}) - _, _, err := AcquireSocket(context.Background(), retryMgr, mockDialer, "ws://test", nil, nil, func(error) { - callbackCalled.Store(true) - }) + errCh := make(chan error, 1) + retryCfg, _ := NewRetryConfig( + WithMaxRetries(3), + WithInitialDelay(1*time.Millisecond), + WithMaxDelay(5*time.Millisecond), + WithJitterFactor(0.0), + ) + retryMgr := NewRetryManager(retryCfg) + + _, err := AcquireSocket(context.Background(), retryMgr, dial, errCh, nil) assert.NoError(t, err) - assert.False(t, callbackCalled.Load()) - }) - t.Run("nil callback does not panic on failed dial", func(t *testing.T) { - mockDialer := &honeybeetest.MockDialer{ - DialContextFunc: func(context.Context, string, http.Header) (types.Socket, *http.Response, error) { - return nil, nil, errors.New("dial failed") - }, - } - - retryMgr := NewRetryManager(RetryConfig{Disabled: true}) - assert.NotPanics(t, func() { - AcquireSocket(context.Background(), retryMgr, mockDialer, "ws://test", nil, nil, nil) - }) + assert.Len(t, errCh, 0) }) } - -func TestAcquireSocketPassesHeaders(t *testing.T) { - header := http.Header{"User-Agent": []string{"test-agent"}} - dialCalled := false - - mockDialer := &honeybeetest.MockDialer{ - DialContextFunc: func(ctx context.Context, url string, h http.Header) (types.Socket, *http.Response, error) { - assert.Equal(t, "test-agent", h.Get("User-Agent")) - dialCalled = true - return honeybeetest.NewMockSocket(), nil, nil - }, - } - - retryMgr := NewRetryManager(RetryConfig{MaxRetries: 0, InitialDelay: 1 * time.Millisecond, MaxDelay: 5 * time.Millisecond}) - _, _, err := AcquireSocket(context.Background(), retryMgr, mockDialer, "ws://test", header, nil, nil) - - assert.NoError(t, err) - assert.True(t, dialCalled) -}