diff --git a/pool.go b/pool.go index 74370fc..88d2796 100644 --- a/pool.go +++ b/pool.go @@ -178,19 +178,12 @@ func NewPool(ctx context.Context, config *PoolConfig, handler slog.Handler, logger = slog.New(handler).With(slog.Any("component", c)) } - var dialer types.Dialer - // if config.ConnectionConfig.Dialer != nil { - // dialer = config.ConnectionConfig.Dialer - // } else { - // dialer = transport.NewDialer() - // } - return &Pool{ peers: make(map[string]*Peer), inbox: make(chan types.InboxMessage, config.InboxBufferSize), events: make(chan PoolEvent, config.EventsBufferSize), - dialer: dialer, + dialer: transport.NewDialer(), config: config, handler: handler, logger: logger, @@ -296,14 +289,14 @@ func (p *Pool) Close() { type ConnectOption func(*connectOptions) type connectOptions struct { - dialer types.Dialer + dialFn func(context.Context) (types.Socket, error) } -// WithDialer returns a ConnectOption that overrides the pool dialer for this -// connection only. -func WithDialer(d types.Dialer) ConnectOption { +// WithDialFunc returns a ConnectOption that overrides the dial function for +// this connection. +func WithDialFunc(f func(context.Context) (types.Socket, error)) ConnectOption { return func(o *connectOptions) { - o.dialer = d + o.dialFn = f } } @@ -328,23 +321,26 @@ func (p *Pool) Connect(id string, opts ...ConnectOption) error { return ErrPeerExists } - wc := p.config.WorkerConfig - worker, err := NewWorker(p.ctx, id, nil, &wc, p.handler) - if err != nil { - return err - } - o := &connectOptions{} for _, opt := range opts { opt(o) } - // effectiveDialer := p.dialer - // if o.dialer != nil { - // effectiveDialer = o.dialer - // } - // cc := p.config.ConnectionConfig.Clone() - // cc.Dialer = effectiveDialer + // TODO: pass headers + dialFn := func(ctx context.Context) (types.Socket, error) { + socket, _, err := p.dialer.DialContext(ctx, id, nil) + return socket, err + } + + if o.dialFn != nil { + dialFn = o.dialFn + } + + wc := p.config.WorkerConfig + worker, err := NewWorker(p.ctx, id, dialFn, &wc, p.handler) + if err != nil { + return err + } pool := PoolPlugin{ Inbox: p.inbox, diff --git a/pool_test.go b/pool_test.go index 96f5125..39fbf7f 100644 --- a/pool_test.go +++ b/pool_test.go @@ -4,7 +4,6 @@ import ( "context" "fmt" "git.wisehodl.dev/jay/go-honeybee/honeybeetest" - "git.wisehodl.dev/jay/go-honeybee/transport" "git.wisehodl.dev/jay/go-honeybee/types" "github.com/gorilla/websocket" "github.com/stretchr/testify/assert" @@ -17,14 +16,15 @@ import ( func setupPool(t *testing.T) (*Pool, *honeybeetest.MockDialer) { t.Helper() dialer := &honeybeetest.MockDialer{ - DialContextFunc: func(context.Context, string, http.Header) (types.Socket, *http.Response, error) { + DialContextFunc: func( + context.Context, string, http.Header, + ) (types.Socket, *http.Response, error) { return honeybeetest.NewMockSocket(), nil, nil }, } - cc, _ := transport.NewConnectionConfig(transport.WithConnectionDialer(dialer)) - config, _ := NewPoolConfig(WithConnectionConfig(*cc)) - pool, err := NewPool(context.Background(), config, nil) - assert.NoError(t, err) + config, _ := NewPoolConfig() + pool, _ := NewPool(context.Background(), config, nil) + pool.dialer = dialer return pool, dialer } @@ -86,29 +86,28 @@ func TestPoolConnect(t *testing.T) { } func TestPoolConnectWithDialer(t *testing.T) { - t.Run("per-call dialer is used instead of pool dialer", func(t *testing.T) { + t.Run("per-call dial function is used instead of pool dialer", func(t *testing.T) { + config, _ := NewPoolConfig() + pool, err := NewPool(context.Background(), config, nil) + assert.NoError(t, err) + perCallUsed := false - perCallDialer := &honeybeetest.MockDialer{ - DialContextFunc: func(ctx context.Context, url string, h http.Header) (types.Socket, *http.Response, error) { - perCallUsed = true - return honeybeetest.NewMockSocket(), nil, nil - }, + dialFn := func(_ context.Context) (types.Socket, error) { + perCallUsed = true + return honeybeetest.NewMockSocket(), nil } // pool dialer should NOT be called - poolDialer := &honeybeetest.MockDialer{ - DialContextFunc: func(context.Context, string, http.Header) (types.Socket, *http.Response, error) { + pool.dialer = &honeybeetest.MockDialer{ + DialContextFunc: func( + context.Context, string, http.Header, + ) (types.Socket, *http.Response, error) { t.Error("pool dialer should not be called when per-call dialer is provided") return nil, nil, fmt.Errorf("unexpected call") }, } - cc, _ := transport.NewConnectionConfig(transport.WithConnectionDialer(poolDialer)) - config, _ := NewPoolConfig(WithConnectionConfig(*cc)) - pool, err := NewPool(context.Background(), config, nil) - assert.NoError(t, err) - - err = pool.Connect("wss://test", WithDialer(perCallDialer)) + err = pool.Connect("wss://test", WithDialFunc(dialFn)) assert.NoError(t, err) honeybeetest.Eventually(t, func() bool { @@ -182,25 +181,25 @@ func TestPoolRemove(t *testing.T) { } func TestPoolDialFailed(t *testing.T) { - t.Run("EventDialFailed surfaces with correct peer ID, non-nil Err, and non-zero At", func(t *testing.T) { - dialErr := fmt.Errorf("connection refused") - failingDialer := &honeybeetest.MockDialer{ - DialContextFunc: func(context.Context, string, http.Header) (types.Socket, *http.Response, error) { - return nil, nil, dialErr - }, - } - cc, _ := transport.NewConnectionConfig( - transport.WithConnectionDialer(failingDialer), - transport.WithRetryDisabled(), - ) + t.Run("EventDialFailed received after failed dial", func(t *testing.T) { + wc, _ := NewWorkerConfig(WithRetryDisabled()) poolCfg, _ := NewPoolConfig( WithInboxBufferSize(256), WithEventsBufferSize(10), - WithConnectionConfig(*cc), + WithWorkerConfig(*wc), ) pool, err := NewPool(context.Background(), poolCfg, nil) assert.NoError(t, err) + dialErr := fmt.Errorf("connection refused") + pool.dialer = &honeybeetest.MockDialer{ + DialContextFunc: func( + context.Context, string, http.Header, + ) (types.Socket, *http.Response, error) { + return nil, nil, dialErr + }, + } + err = pool.Connect("wss://test") assert.NoError(t, err) @@ -245,17 +244,19 @@ func TestPoolSend(t *testing.T) { outgoingData <- honeybeetest.MockOutgoingData{MsgType: msgType, Data: data} return nil } - mockDialer := &honeybeetest.MockDialer{ - DialContextFunc: func(context.Context, string, http.Header) (types.Socket, *http.Response, error) { + + config, _ := NewPoolConfig() + pool, err := NewPool(context.Background(), config, nil) + assert.NoError(t, err) + + pool.dialer = &honeybeetest.MockDialer{ + DialContextFunc: func( + context.Context, string, http.Header, + ) (types.Socket, *http.Response, error) { return mockSocket, nil, nil }, } - cc, _ := transport.NewConnectionConfig(transport.WithConnectionDialer(mockDialer)) - config, _ := NewPoolConfig(WithConnectionConfig(*cc)) - pool, err := NewPool(context.Background(), config, nil) - assert.NoError(t, err) - err = pool.Connect("wss://test") assert.NoError(t, err) expectEvent(t, pool.events, "wss://test", EventConnected) diff --git a/transport/socket.go b/transport/socket.go index 3fbb106..c11f5bb 100644 --- a/transport/socket.go +++ b/transport/socket.go @@ -85,14 +85,22 @@ func DialWithRetry( for { if logger != nil { - logger.Debug("dialing", "attempt", mgr.RetryCount()+1) + if mgr == nil { + logger.Debug("dialing") + } else { + logger.Debug("dialing", "attempt", mgr.RetryCount()+1) + } } // dial socket, err := dialFn(ctx) if err == nil { if logger != nil { - logger.Debug("dial successful", "attempt", mgr.RetryCount()+1) + if mgr == nil { + logger.Debug("dial successful") + } else { + logger.Debug("dial successful", "attempt", mgr.RetryCount()+1) + } } return socket, nil } diff --git a/transport/socket_test.go b/transport/socket_test.go index 4e5d27e..2697635 100644 --- a/transport/socket_test.go +++ b/transport/socket_test.go @@ -7,6 +7,7 @@ import ( "git.wisehodl.dev/jay/go-honeybee/honeybeetest" "git.wisehodl.dev/jay/go-honeybee/types" "github.com/stretchr/testify/assert" + "sync" "sync/atomic" "testing" "time" @@ -252,8 +253,13 @@ func TestDialWithRetry_DialErrors(t *testing.T) { } } + var mu sync.Mutex var gotErrors []error - onError := func(err error) { gotErrors = append(gotErrors, err) } + onError := func(err error) { + mu.Lock() + defer mu.Unlock() + gotErrors = append(gotErrors, err) + } retryCfg, _ := NewRetryConfig( WithMaxRetries(3), @@ -269,9 +275,13 @@ func TestDialWithRetry_DialErrors(t *testing.T) { }() honeybeetest.Eventually(t, func() bool { + mu.Lock() + defer mu.Unlock() return len(gotErrors) == 2 }, "expected errors") + mu.Lock() + defer mu.Unlock() assert.Equal(t, dialErr1, gotErrors[0]) assert.Equal(t, dialErr2, gotErrors[1]) }) diff --git a/worker.go b/worker.go index 4a1627c..6160877 100644 --- a/worker.go +++ b/worker.go @@ -64,9 +64,9 @@ func WithRetryConfig(value transport.RetryConfig) WorkerOption { } } -func WithRetryDisabled(value bool) WorkerOption { +func WithRetryDisabled() WorkerOption { return func(c *WorkerConfig) { - c.RetryDisabled = value + c.RetryDisabled = true } } @@ -224,13 +224,8 @@ func (w *DefaultWorker) runSession(ctx context.Context, pool PoolPlugin) { } // setup new connection - conn, err := transport.NewConnection( + conn, _ := transport.NewConnection( ctx, socket, &w.config.ConnectionConfig, w.handler) - if err != nil { - // TODO: cleanup worker? surface error? - return - } - w.conn.Store(conn) pool.Events <- PoolEvent{ID: w.url, Kind: EventConnected, At: time.Now()}