diff --git a/pool.go b/pool.go index 2de26c3..29242e8 100644 --- a/pool.go +++ b/pool.go @@ -64,8 +64,8 @@ type PoolStats struct { } type PeerStats struct { - ID string - Worker WorkerStats + ID string + Session SessionStats } type PoolPlugin struct { @@ -82,20 +82,20 @@ type PoolPlugin struct { type PoolConfig struct { InboxBufferSize int EventsBufferSize int - WorkerConfig WorkerConfig + SessionConfig SessionConfig RequestHeader http.Header } type PoolOption func(*PoolConfig) func NewPoolConfig(opts ...PoolOption) (*PoolConfig, error) { - workerCfg, _ := NewWorkerConfig() + sessionCfg, _ := NewSessionConfig() defaultHeader := http.Header{} defaultHeader.Set("User-Agent", "honeybee/0.1.0") cfg := &PoolConfig{ InboxBufferSize: 256, EventsBufferSize: 10, - WorkerConfig: *workerCfg, + SessionConfig: *sessionCfg, RequestHeader: defaultHeader, } for _, o := range opts { @@ -119,9 +119,9 @@ func WithEventsBufferSize(value int) PoolOption { } } -func WithWorkerConfig(wc WorkerConfig) PoolOption { +func WithSessionConfig(wc SessionConfig) PoolOption { return func(c *PoolConfig) { - c.WorkerConfig = wc + c.SessionConfig = wc } } @@ -142,7 +142,7 @@ func ValidatePoolConfig(c *PoolConfig) error { if c.EventsBufferSize < 1 { return fmt.Errorf("invalid events buffer size: %d", c.EventsBufferSize) } - if err := ValidateWorkerConfig(&c.WorkerConfig); err != nil { + if err := ValidateSessionConfig(&c.SessionConfig); err != nil { return err } return nil @@ -153,8 +153,8 @@ func ValidatePoolConfig(c *PoolConfig) error { // ---------------------------------------------------------------------------- type Peer struct { - id string - worker Worker + id string + session *session } type Pool struct { @@ -240,8 +240,8 @@ func (p *Pool) Stats() PoolStats { peerStats := make([]PeerStats, 0, count) for id, peer := range p.peers { peerStats = append(peerStats, PeerStats{ - ID: id, - Worker: peer.worker.Stats(), + ID: id, + Session: peer.session.Stats(), }) } @@ -267,8 +267,8 @@ func (p *Pool) PeerStats(id string) (PeerStats, error) { } return PeerStats{ - ID: id, - Worker: peer.worker.Stats(), + ID: id, + Session: peer.session.Stats(), }, nil } @@ -284,7 +284,7 @@ func (p *Pool) Close() { } p.closed = true - p.cancel() // closes all workers + p.cancel() // closes all sessions // remove all peers p.peers = make(map[string]*Peer) @@ -356,8 +356,8 @@ func (p *Pool) Connect(id string, opts ...ConnectOption) error { dialFn = o.dialFn } - wc := p.config.WorkerConfig - worker, err := NewWorker(p.ctx, id, dialFn, &wc, p.handler) + wc := p.config.SessionConfig + session, err := newSession(p.ctx, id, dialFn, &wc, p.handler) if err != nil { return err } @@ -369,6 +369,7 @@ func (p *Pool) Connect(id string, opts ...ConnectOption) error { Retire: func(err error) { p.mu.Lock() if p.closed { + p.mu.Unlock() return } delete(p.peers, id) @@ -378,10 +379,10 @@ func (p *Pool) Connect(id string, opts ...ConnectOption) error { } p.wg.Go(func() { - worker.Start(pool) + session.Start(pool) }) - p.peers[id] = &Peer{id: id, worker: worker} + p.peers[id] = &Peer{id: id, session: session} if p.logger != nil { p.logger.Debug("registered peer", "peer", id) @@ -413,7 +414,7 @@ func (p *Pool) Remove(id string) error { } delete(p.peers, id) - peer.worker.Stop() + peer.session.Stop() if p.logger != nil { p.logger.Debug("disconnected from peer", "peer", id) @@ -440,7 +441,7 @@ func (p *Pool) Send(id string, data []byte) error { return ErrPeerNotFound } - err = peer.worker.Send(data) + err = peer.session.Send(data) if err != nil { return err } diff --git a/pool_test.go b/pool_test.go index 5aba45a..ac27825 100644 --- a/pool_test.go +++ b/pool_test.go @@ -185,11 +185,11 @@ func TestPoolRemove(t *testing.T) { func TestPoolDialFailed(t *testing.T) { t.Run("EventDialFailed received after failed dial", func(t *testing.T) { - wc, _ := NewWorkerConfig(WithRetryDisabled()) + wc, _ := NewSessionConfig(WithRetryDisabled()) poolCfg, _ := NewPoolConfig( WithInboxBufferSize(256), WithEventsBufferSize(10), - WithWorkerConfig(*wc), + WithSessionConfig(*wc), ) pool, err := NewPool(context.Background(), poolCfg, nil) assert.NoError(t, err) @@ -242,8 +242,8 @@ func TestPoolDialFailed(t *testing.T) { func TestPoolRetire(t *testing.T) { t.Run("peer absent and EventRetired emitted after self-retire", func(t *testing.T) { - wc, _ := NewWorkerConfig(WithRetryDisabled()) - poolCfg, _ := NewPoolConfig(WithWorkerConfig(*wc)) + wc, _ := NewSessionConfig(WithRetryDisabled()) + poolCfg, _ := NewPoolConfig(WithSessionConfig(*wc)) pool, _ := NewPool(context.Background(), poolCfg, nil) pool.dialer = &honeybeetest.MockDialer{ @@ -274,8 +274,8 @@ func TestPoolRetire(t *testing.T) { }) t.Run("concurrent Remove and self-retire is safe", func(t *testing.T) { - wc, _ := NewWorkerConfig(WithRetryDisabled()) - poolCfg, _ := NewPoolConfig(WithWorkerConfig(*wc)) + wc, _ := NewSessionConfig(WithRetryDisabled()) + poolCfg, _ := NewPoolConfig(WithSessionConfig(*wc)) pool, _ := NewPool(context.Background(), poolCfg, nil) pool.dialer = &honeybeetest.MockDialer{ diff --git a/worker.go b/session.go similarity index 81% rename from worker.go rename to session.go index 5cbc2a3..bd60c23 100644 --- a/worker.go +++ b/session.go @@ -24,7 +24,7 @@ var ErrConnectionUnavailable = errors.New("connection unavailable") // Config // ---------------------------------------------------------------------------- -type WorkerConfig struct { +type SessionConfig struct { ConnectionConfig transport.ConnectionConfig RetryConfig transport.RetryConfig RetryDisabled bool @@ -32,12 +32,12 @@ type WorkerConfig struct { ReconnectDelay time.Duration } -type WorkerOption func(*WorkerConfig) +type SessionOption func(*SessionConfig) -func NewWorkerConfig(opts ...WorkerOption) (*WorkerConfig, error) { +func NewSessionConfig(opts ...SessionOption) (*SessionConfig, error) { connCfg, _ := transport.NewConnectionConfig() retryCfg, _ := transport.NewRetryConfig() - cfg := &WorkerConfig{ + cfg := &SessionConfig{ ConnectionConfig: *connCfg, RetryConfig: retryCfg, KeepaliveTimeout: 60 * time.Second, @@ -46,44 +46,44 @@ func NewWorkerConfig(opts ...WorkerOption) (*WorkerConfig, error) { for _, o := range opts { o(cfg) } - if err := ValidateWorkerConfig(cfg); err != nil { + if err := ValidateSessionConfig(cfg); err != nil { return nil, err } return cfg, nil } -func WithConnectionConfig(value transport.ConnectionConfig) WorkerOption { - return func(c *WorkerConfig) { +func WithConnectionConfig(value transport.ConnectionConfig) SessionOption { + return func(c *SessionConfig) { c.ConnectionConfig = value } } -func WithRetryConfig(value transport.RetryConfig) WorkerOption { - return func(c *WorkerConfig) { +func WithRetryConfig(value transport.RetryConfig) SessionOption { + return func(c *SessionConfig) { c.RetryConfig = value } } -func WithRetryDisabled() WorkerOption { - return func(c *WorkerConfig) { +func WithRetryDisabled() SessionOption { + return func(c *SessionConfig) { c.RetryDisabled = true } } // When KeepaliveTimeout is set to zero, keepalive functionality is disabled. -func WithKeepaliveTimeout(value time.Duration) WorkerOption { - return func(c *WorkerConfig) { +func WithKeepaliveTimeout(value time.Duration) SessionOption { + return func(c *SessionConfig) { c.KeepaliveTimeout = value } } -func WithReconnectDelay(value time.Duration) WorkerOption { - return func(c *WorkerConfig) { +func WithReconnectDelay(value time.Duration) SessionOption { + return func(c *SessionConfig) { c.ReconnectDelay = value } } -func ValidateWorkerConfig(c *WorkerConfig) error { +func ValidateSessionConfig(c *SessionConfig) error { if !c.RetryDisabled { if err := transport.ValidateRetryConfig(c.RetryConfig); err != nil { return err @@ -99,17 +99,10 @@ func ValidateWorkerConfig(c *WorkerConfig) error { } // ---------------------------------------------------------------------------- -// Worker +// Session // ---------------------------------------------------------------------------- -type Worker interface { - Start(pool PoolPlugin) - Stop() - Send(data []byte) error - Stats() WorkerStats -} - -type WorkerStats struct { +type SessionStats struct { ConnectionAvailable bool Connection transport.ConnectionStats ChanIncoming int @@ -119,7 +112,7 @@ type WorkerStats struct { TotalRestarts uint64 } -type DefaultWorker struct { +type session struct { url string dialFn func(context.Context) (types.Socket, error) conn atomic.Pointer[transport.Connection] @@ -128,7 +121,7 @@ type DefaultWorker struct { ctx context.Context cancel context.CancelFunc - config WorkerConfig + config SessionConfig handler slog.Handler logger *slog.Logger @@ -137,28 +130,28 @@ type DefaultWorker struct { restartCount *atomic.Uint64 } -func NewWorker( +func newSession( ctx context.Context, url string, dialFn func(context.Context) (types.Socket, error), - config *WorkerConfig, + config *SessionConfig, handler slog.Handler, -) (*DefaultWorker, error) { +) (*session, error) { if config == nil { - config, _ = NewWorkerConfig() + config, _ = NewSessionConfig() } - if err := ValidateWorkerConfig(config); err != nil { + if err := ValidateSessionConfig(config); err != nil { return nil, err } if component.FromContext(ctx) == nil { - ctx = component.MustNew(ctx, "honeybee", "worker") + ctx = component.MustNew(ctx, "honeybee", "session") } else { - ctx = component.MustExtend(ctx, "worker") + ctx = component.MustExtend(ctx, "session") } ctx, cancel := context.WithCancel(ctx) - w := &DefaultWorker{ + w := &session{ url: url, dialFn: dialFn, @@ -182,13 +175,13 @@ func NewWorker( return w, nil } -func (w *DefaultWorker) Start(pool PoolPlugin) { +func (w *session) Start(pool PoolPlugin) { if w.logger != nil { w.logger.Debug("starting") } var wg sync.WaitGroup - wg.Go(func() { w.runSession(w.ctx, pool) }) + wg.Go(func() { w.run(w.ctx, pool) }) if w.logger != nil { w.logger.Debug("started") @@ -201,7 +194,7 @@ func (w *DefaultWorker) Start(pool PoolPlugin) { } } -func (w *DefaultWorker) runSession(ctx context.Context, pool PoolPlugin) { +func (w *session) run(ctx context.Context, pool PoolPlugin) { for { var retryMgr *transport.RetryManager if !w.config.RetryDisabled { @@ -298,7 +291,7 @@ func (w *DefaultWorker) runSession(ctx context.Context, pool PoolPlugin) { w.conn.Store(nil) pool.Events <- PoolEvent{ID: w.url, Kind: EventDisconnected, At: time.Now()} - // exit if worker is shutting down + // exit if session is shutting down select { case <-ctx.Done(): return @@ -311,7 +304,7 @@ func (w *DefaultWorker) runSession(ctx context.Context, pool PoolPlugin) { } } -func (w *DefaultWorker) startKeepalive() ( +func (w *session) startKeepalive() ( stop func(), inactive func() <-chan time.Time, heartbeat func(), @@ -354,14 +347,14 @@ func (w *DefaultWorker) startKeepalive() ( return } -func (w *DefaultWorker) Stop() { +func (w *session) Stop() { if w.logger != nil { w.logger.Info("shutting down") } w.cancel() } -func (w *DefaultWorker) Send(data []byte) error { +func (w *session) Send(data []byte) error { conn := w.conn.Load() if conn == nil { // connection not established by session @@ -383,7 +376,7 @@ func (w *DefaultWorker) Send(data []byte) error { return nil } -func (w *DefaultWorker) Stats() WorkerStats { +func (w *session) Stats() SessionStats { connectionAvailable := false incomingLen := 0 connStats := transport.ConnectionStats{} @@ -395,7 +388,7 @@ func (w *DefaultWorker) Stats() WorkerStats { connStats = conn.Stats() } - return WorkerStats{ + return SessionStats{ ConnectionAvailable: connectionAvailable, Connection: connStats, ChanIncoming: incomingLen, diff --git a/worker_test.go b/session_test.go similarity index 86% rename from worker_test.go rename to session_test.go index 8722259..12d88fb 100644 --- a/worker_test.go +++ b/session_test.go @@ -14,7 +14,7 @@ import ( "time" ) -func makeWorkerContext(t *testing.T) ( +func makeSessionContext(t *testing.T) ( inbox chan types.InboxMessage, events chan PoolEvent, pool PoolPlugin, @@ -31,19 +31,19 @@ func makeWorkerContext(t *testing.T) ( return } -func makeWorker( +func makeSession( t *testing.T, socket *honeybeetest.MockSocket, - config *WorkerConfig, + config *SessionConfig, ctx context.Context, cancel context.CancelFunc, -) *DefaultWorker { +) *session { t.Helper() if config == nil { - config, _ = NewWorkerConfig(WithReconnectDelay(0 * time.Second)) + config, _ = NewSessionConfig(WithReconnectDelay(0 * time.Second)) } dialFn := func(_ context.Context) (types.Socket, error) { return socket, nil } - return &DefaultWorker{ + return &session{ ctx: ctx, cancel: cancel, url: "wss://test", @@ -56,14 +56,14 @@ func makeWorker( } } -func TestWorkerSession(t *testing.T) { +func TestSession(t *testing.T) { t.Run("EventConnected emitted after dial succeeds", func(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() mockSocket := honeybeetest.NewMockSocket() - w := makeWorker(t, mockSocket, nil, ctx, cancel) - _, events, pool := makeWorkerContext(t) + w := makeSession(t, mockSocket, nil, ctx, cancel) + _, events, pool := makeSessionContext(t) var wg sync.WaitGroup wg.Go(func() { @@ -80,16 +80,16 @@ func TestWorkerSession(t *testing.T) { }, "expected EventConnected") }) - t.Run("dial failure exhausted - worker exits cleanly, no connected/disconnected events", func(t *testing.T) { + t.Run("dial failure exhausted - session exits cleanly, no connected/disconnected events", func(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() - config, _ := NewWorkerConfig(WithRetryDisabled()) - w := makeWorker(t, nil, config, ctx, cancel) + config, _ := NewSessionConfig(WithRetryDisabled()) + w := makeSession(t, nil, config, ctx, cancel) w.dialFn = func(_ context.Context) (types.Socket, error) { return nil, fmt.Errorf("connection refused") } - _, events, pool := makeWorkerContext(t) + _, events, pool := makeSessionContext(t) var wg sync.WaitGroup wg.Go(func() { w.Start(pool) }) @@ -103,7 +103,7 @@ func TestWorkerSession(t *testing.T) { default: return false } - }, "expected worker to exit after terminal dial failure") + }, "expected session to exit after terminal dial failure") honeybeetest.Never(t, func() bool { select { @@ -119,12 +119,12 @@ func TestWorkerSession(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() - config, _ := NewWorkerConfig(WithRetryDisabled()) - w := makeWorker(t, nil, config, ctx, cancel) + config, _ := NewSessionConfig(WithRetryDisabled()) + w := makeSession(t, nil, config, ctx, cancel) w.dialFn = func(_ context.Context) (types.Socket, error) { return nil, fmt.Errorf("connection refused") } - _, events, pool := makeWorkerContext(t) + _, events, pool := makeSessionContext(t) retired := atomic.Bool{} var retiredErr atomic.Pointer[error] @@ -148,7 +148,7 @@ func TestWorkerSession(t *testing.T) { default: return false } - }, "expected worker to exit cleanly") + }, "expected session to exit cleanly") honeybeetest.Never(t, func() bool { select { @@ -164,12 +164,12 @@ func TestWorkerSession(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() - w := makeWorker(t, nil, nil, ctx, cancel) + w := makeSession(t, nil, nil, ctx, cancel) w.dialFn = func(dialCtx context.Context) (types.Socket, error) { <-dialCtx.Done() return nil, dialCtx.Err() } - _, _, pool := makeWorkerContext(t) + _, _, pool := makeSessionContext(t) retired := atomic.Bool{} pool.Retire = func(_ error) { retired.Store(true) } @@ -187,7 +187,7 @@ func TestWorkerSession(t *testing.T) { default: return false } - }, "expected worker to exit after Stop") + }, "expected session to exit after Stop") assert.False(t, retired.Load(), "expected Retire not called") }) @@ -197,8 +197,8 @@ func TestWorkerSession(t *testing.T) { defer cancel() mockSocket := honeybeetest.NewMockSocket() - w := makeWorker(t, mockSocket, nil, ctx, cancel) - _, events, pool := makeWorkerContext(t) + w := makeSession(t, mockSocket, nil, ctx, cancel) + _, events, pool := makeSessionContext(t) retired := atomic.Bool{} pool.Retire = func(_ error) { retired.Store(true) } @@ -222,12 +222,12 @@ func TestWorkerSession(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() - w := makeWorker(t, nil, nil, ctx, cancel) + w := makeSession(t, nil, nil, ctx, cancel) w.dialFn = func(dialCtx context.Context) (types.Socket, error) { <-dialCtx.Done() return nil, dialCtx.Err() } - _, events, pool := makeWorkerContext(t) + _, events, pool := makeSessionContext(t) var wg sync.WaitGroup wg.Go(func() { w.Start(pool) }) @@ -253,8 +253,8 @@ func TestWorkerSession(t *testing.T) { defer cancel() _, mockSocket, _, outgoingData := setupTestConnection(t) - w := makeWorker(t, mockSocket, nil, ctx, cancel) - _, events, pool := makeWorkerContext(t) + w := makeSession(t, mockSocket, nil, ctx, cancel) + _, events, pool := makeSessionContext(t) var wg sync.WaitGroup wg.Go(func() { @@ -302,8 +302,8 @@ func TestWorkerSession(t *testing.T) { } } - w := makeWorker(t, mockSocket, nil, ctx, cancel) - inbox, events, pool := makeWorkerContext(t) + w := makeSession(t, mockSocket, nil, ctx, cancel) + inbox, events, pool := makeSessionContext(t) var wg sync.WaitGroup wg.Go(func() { @@ -343,13 +343,13 @@ func TestWorkerSession(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() - config, _ := NewWorkerConfig( + config, _ := NewSessionConfig( WithReconnectDelay(0), WithKeepaliveTimeout(60*time.Millisecond), ) _, mockSocket, incomingData, _ := setupTestConnection(t) - w := makeWorker(t, mockSocket, config, ctx, cancel) - _, events, pool := makeWorkerContext(t) + w := makeSession(t, mockSocket, config, ctx, cancel) + _, events, pool := makeSessionContext(t) var wg sync.WaitGroup wg.Go(func() { w.Start(pool) }) @@ -400,12 +400,12 @@ func TestWorkerSession(t *testing.T) { mockSocket, incomingData, _ := honeybeetest.SetupTestSocket(t) mockSocket.SetPongHandlerFunc = func(h func(string) error) { pongHandler = h } - config, _ := NewWorkerConfig( + config, _ := NewSessionConfig( WithReconnectDelay(0), WithKeepaliveTimeout(60*time.Millisecond), ) - w := makeWorker(t, mockSocket, config, ctx, cancel) - _, events, pool := makeWorkerContext(t) + w := makeSession(t, mockSocket, config, ctx, cancel) + _, events, pool := makeSessionContext(t) var wg sync.WaitGroup wg.Go(func() { w.Start(pool) }) @@ -451,13 +451,13 @@ func TestWorkerSession(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() - config, _ := NewWorkerConfig( + config, _ := NewSessionConfig( WithReconnectDelay(0), WithKeepaliveTimeout(30*time.Millisecond), ) _, mockSocket, _, _ := setupTestConnection(t) - w := makeWorker(t, mockSocket, config, ctx, cancel) - _, events, pool := makeWorkerContext(t) + w := makeSession(t, mockSocket, config, ctx, cancel) + _, events, pool := makeSessionContext(t) var wg sync.WaitGroup wg.Go(func() { w.Start(pool) }) @@ -496,9 +496,9 @@ func TestWorkerSession(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() - _, events, pool := makeWorkerContext(t) + _, events, pool := makeSessionContext(t) _, mockSocket, incomingData, _ := setupTestConnection(t) - w := makeWorker(t, mockSocket, nil, ctx, cancel) + w := makeSession(t, mockSocket, nil, ctx, cancel) var wg sync.WaitGroup wg.Go(func() { @@ -539,9 +539,9 @@ func TestWorkerSession(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() - _, events, pool := makeWorkerContext(t) + _, events, pool := makeSessionContext(t) mockSocket := honeybeetest.NewMockSocket() - w := makeWorker(t, mockSocket, nil, ctx, cancel) + w := makeSession(t, mockSocket, nil, ctx, cancel) var wg sync.WaitGroup wg.Go(func() { @@ -582,11 +582,11 @@ func TestWorkerSession(t *testing.T) { t.Run("parent context cancel exits cleanly and wg drains", func(t *testing.T) { parentCtx, parentCancel := context.WithCancel(context.Background()) - workerCtx, workerCancel := context.WithCancel(parentCtx) + sessionCtx, sessionCancel := context.WithCancel(parentCtx) - _, events, pool := makeWorkerContext(t) + _, events, pool := makeSessionContext(t) mockSocket := honeybeetest.NewMockSocket() - w := makeWorker(t, mockSocket, nil, workerCtx, workerCancel) + w := makeSession(t, mockSocket, nil, sessionCtx, sessionCancel) var wg sync.WaitGroup wg.Go(func() { @@ -603,7 +603,7 @@ func TestWorkerSession(t *testing.T) { }, "expected EventConnected") // drain events after parent cancel — we don't assert what they are, - // only that the worker exits + // only that the session exits parentCancel() done := make(chan struct{}) @@ -622,8 +622,8 @@ func TestWorkerSession(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() - w := makeWorker(t, nil, nil, ctx, cancel) - _, events, pool := makeWorkerContext(t) + w := makeSession(t, nil, nil, ctx, cancel) + _, events, pool := makeSessionContext(t) dialErr := errors.New("connection refused") w.dialFn = func(_ context.Context) (types.Socket, error) { @@ -646,12 +646,12 @@ func TestWorkerSession(t *testing.T) { }, "expected EventDialFailed") }) - t.Run("no EventDialFailed when worker is stopped mid-dial", func(t *testing.T) { + t.Run("no EventDialFailed when session is stopped mid-dial", func(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() - w := makeWorker(t, nil, nil, ctx, cancel) - _, events, pool := makeWorkerContext(t) + w := makeSession(t, nil, nil, ctx, cancel) + _, events, pool := makeSessionContext(t) w.dialFn = func(dialCtx context.Context) (types.Socket, error) { <-dialCtx.Done() @@ -672,7 +672,7 @@ func TestWorkerSession(t *testing.T) { default: return false } - }, "expected worker to exit after Stop") + }, "expected session to exit after Stop") honeybeetest.Never(t, func() bool { select { @@ -685,7 +685,7 @@ func TestWorkerSession(t *testing.T) { }) } -func TestWorkerSend(t *testing.T) { +func TestSession_Send(t *testing.T) { t.Run("data sent to mock socket", func(t *testing.T) { conn, _, _, outgoingData := setupTestConnection(t) defer conn.Close() @@ -695,7 +695,7 @@ func TestWorkerSend(t *testing.T) { heartbeat := make(chan struct{}) heartbeatCount := atomic.Int32{} - w := &DefaultWorker{ + w := &session{ ctx: ctx, cancel: cancel, url: "wss://test", @@ -740,7 +740,7 @@ func TestWorkerSend(t *testing.T) { heartbeat := make(chan struct{}) heartbeatCount := atomic.Int32{} - w := &DefaultWorker{ + w := &session{ ctx: ctx, cancel: cancel, url: "wss://test", @@ -768,13 +768,13 @@ func TestWorkerSend(t *testing.T) { }) t.Run("returns error if connection is unavailable", func(t *testing.T) { - // no connection available to worker + // no connection available to session ctx, cancel := context.WithCancel(context.Background()) heartbeat := make(chan struct{}) - w := &DefaultWorker{ + w := &session{ ctx: ctx, cancel: cancel, url: "wss://test", diff --git a/transport/connection.go b/transport/connection.go index dc613f9..d6ac6d9 100644 --- a/transport/connection.go +++ b/transport/connection.go @@ -23,7 +23,6 @@ var ( ErrConnectionClosed = errors.New("connection closed") ErrWriteFailed = errors.New("write failed") ErrNilSocket = errors.New("socket cannot be nil") - ErrSocketExists = errors.New("socket already exists") ErrFailedWriteDeadline = errors.New("failed to set write deadline") ErrPeerClosedClean = errors.New("peer closed connection cleanly") ErrPeerClosedUnexpected = errors.New("peer closed connection unexpectedly") @@ -367,7 +366,12 @@ func (c *Connection) startPinger() { timer.Stop() return case <-timer.C: - deadline := time.Now().Add(c.config.WriteTimeout) + var deadline time.Time + if c.config.WriteTimeout > 0 { + deadline = time.Now().Add(c.config.WriteTimeout) + } else { + deadline = time.Now().Add(15 * time.Second) + } err := c.socket.WriteControl(websocket.PingMessage, nil, deadline) if err != nil { return