Refactored package structure.

initiator/config.go

@@ -0,0 +1,189 @@
+package initiator
+
+import (
+	"git.wisehodl.dev/jay/go-honeybee/transport"
+	"time"
+)
+
+// Types
+
+type WorkerFactory func(
+	id string,
+	conn *transport.Connection,
+	onReconnect func() (*transport.Connection, error),
+) Worker
+
+// Pool Config
+
+type PoolConfig struct {
+	ConnectionConfig *transport.ConnectionConfig
+	WorkerFactory    WorkerFactory
+	WorkerConfig     *WorkerConfig
+}
+
+type PoolOption func(*PoolConfig) error
+
+func NewPoolConfig(options ...PoolOption) (*PoolConfig, error) {
+	conf := GetDefaultPoolConfig()
+	if err := applyPoolOptions(conf, options...); err != nil {
+		return nil, err
+	}
+	if err := ValidatePoolConfig(conf); err != nil {
+		return nil, err
+	}
+	return conf, nil
+}
+
+func GetDefaultPoolConfig() *PoolConfig {
+	return &PoolConfig{
+		ConnectionConfig: nil,
+		WorkerFactory:    nil,
+		WorkerConfig:     nil,
+	}
+}
+
+func applyPoolOptions(config *PoolConfig, options ...PoolOption) error {
+	for _, option := range options {
+		if err := option(config); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func ValidatePoolConfig(config *PoolConfig) error {
+	var err error
+
+	if config.ConnectionConfig != nil {
+		err = transport.ValidateConnectionConfig(config.ConnectionConfig)
+		if err != nil {
+			return err
+		}
+	}
+
+	if config.WorkerConfig != nil {
+		err = ValidateWorkerConfig(config.WorkerConfig)
+		if err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func WithConnectionConfig(cc *transport.ConnectionConfig) PoolOption {
+	return func(c *PoolConfig) error {
+		err := transport.ValidateConnectionConfig(cc)
+		if err != nil {
+			return err
+		}
+		c.ConnectionConfig = cc
+		return nil
+	}
+}
+
+func WithWorkerConfig(wc *WorkerConfig) PoolOption {
+	return func(c *PoolConfig) error {
+		err := ValidateWorkerConfig(wc)
+		if err != nil {
+			return err
+		}
+		c.WorkerConfig = wc
+		return nil
+	}
+}
+
+func WithWorkerFactory(wf WorkerFactory) PoolOption {
+	return func(c *PoolConfig) error {
+		c.WorkerFactory = wf
+		return nil
+	}
+}
+
+// Worker Config
+
+type WorkerConfig struct {
+	IdleTimeout  time.Duration
+	MaxQueueSize int
+}
+
+type WorkerOption func(*WorkerConfig) error
+
+func NewWorkerConfig(options ...WorkerOption) (*WorkerConfig, error) {
+	conf := GetDefaultWorkerConfig()
+	if err := applyWorkerOptions(conf, options...); err != nil {
+		return nil, err
+	}
+	if err := ValidateWorkerConfig(conf); err != nil {
+		return nil, err
+	}
+	return conf, nil
+}
+
+func GetDefaultWorkerConfig() *WorkerConfig {
+	return &WorkerConfig{
+		IdleTimeout:  20 * time.Second,
+		MaxQueueSize: 0, // disabled by default
+	}
+}
+
+func applyWorkerOptions(config *WorkerConfig, options ...WorkerOption) error {
+	for _, option := range options {
+		if err := option(config); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func ValidateWorkerConfig(config *WorkerConfig) error {
+	err := validateIdleTimeout(config.IdleTimeout)
+	if err != nil {
+		return err
+	}
+
+	err = validateMaxQueueSize(config.MaxQueueSize)
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func validateMaxQueueSize(value int) error {
+	if value < 0 {
+		return InvalidMaxQueueSize
+	}
+	return nil
+}
+
+func validateIdleTimeout(value time.Duration) error {
+	if value < 0 {
+		return InvalidIdleTimeout
+	}
+	return nil
+}
+
+// When IdleTimeout is set to zero, idle timeouts are disabled.
+func WithIdleTimeout(value time.Duration) WorkerOption {
+	return func(c *WorkerConfig) error {
+		err := validateIdleTimeout(value)
+		if err != nil {
+			return err
+		}
+		c.IdleTimeout = value
+		return nil
+	}
+}
+
+// When MaxQueueSize is set to zero, queue limits are disabled.
+func WithMaxQueueSize(value int) WorkerOption {
+	return func(c *WorkerConfig) error {
+		err := validateMaxQueueSize(value)
+		if err != nil {
+			return err
+		}
+		c.MaxQueueSize = value
+		return nil
+	}
+}

initiator/config_pool_test.go

@@ -0,0 +1,110 @@
+package initiator
+
+import (
+	"git.wisehodl.dev/jay/go-honeybee/transport"
+	"github.com/stretchr/testify/assert"
+	"testing"
+	"time"
+)
+
+func TestNewPoolConfig(t *testing.T) {
+	conf, err := NewPoolConfig()
+	assert.NoError(t, err)
+
+	assert.Equal(t, conf, &PoolConfig{
+		ConnectionConfig: nil,
+		WorkerConfig:     nil,
+		WorkerFactory:    nil,
+	})
+}
+
+func TestDefaultPoolConfig(t *testing.T) {
+	conf := GetDefaultPoolConfig()
+
+	assert.Equal(t, conf, &PoolConfig{
+		ConnectionConfig: nil,
+		WorkerConfig:     nil,
+		WorkerFactory:    nil,
+	})
+}
+
+func TestApplyPoolOptions(t *testing.T) {
+	conf := &PoolConfig{}
+	err := applyPoolOptions(
+		conf,
+		WithConnectionConfig(&transport.ConnectionConfig{}),
+	)
+
+	assert.NoError(t, err)
+	assert.Equal(t, 0*time.Second, conf.ConnectionConfig.WriteTimeout)
+}
+
+func TestWithConnectionConfig(t *testing.T) {
+	conf := &PoolConfig{}
+	opt := WithConnectionConfig(&transport.ConnectionConfig{WriteTimeout: 1 * time.Second})
+	err := applyPoolOptions(conf, opt)
+	assert.NoError(t, err)
+	assert.NotNil(t, conf.ConnectionConfig)
+	assert.Equal(t, 1*time.Second, conf.ConnectionConfig.WriteTimeout)
+
+	// invalid config is rejected
+	conf = &PoolConfig{}
+	opt = WithConnectionConfig(&transport.ConnectionConfig{WriteTimeout: -1 * time.Second})
+	err = applyPoolOptions(conf, opt)
+	assert.Error(t, err)
+}
+
+func TestValidatePoolConfig(t *testing.T) {
+	cases := []struct {
+		name        string
+		conf        PoolConfig
+		wantErr     error
+		wantErrText string
+	}{
+		{
+			name: "valid empty",
+			conf: *&PoolConfig{},
+		},
+		{
+			name: "valid defaults",
+			conf: *GetDefaultPoolConfig(),
+		},
+		{
+			name: "valid complete",
+			conf: PoolConfig{
+				ConnectionConfig: &transport.ConnectionConfig{},
+			},
+		},
+		{
+			name: "invalid connection config",
+			conf: PoolConfig{
+				ConnectionConfig: &transport.ConnectionConfig{
+					Retry: &transport.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 := ValidatePoolConfig(&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)
+		})
+	}
+}

initiator/errors.go

@@ -0,0 +1,17 @@
+package initiator
+
+import "errors"
+import "fmt"
+
+var (
+	InvalidIdleTimeout  = errors.New("idle timeout cannot be negative")
+	InvalidMaxQueueSize = errors.New("maximum queue size cannot be negative")
+)
+
+func NewConfigError(text string) error {
+	return fmt.Errorf("configuration error: %s", text)
+}
+
+func NewPoolError(text string) error {
+	return fmt.Errorf("pool error: %s", text)
+}

initiator/pool.go

@@ -0,0 +1,248 @@
+package initiator
+
+import (
+	"git.wisehodl.dev/jay/go-honeybee/transport"
+	"git.wisehodl.dev/jay/go-honeybee/types"
+	"log/slog"
+	"sync"
+	"time"
+)
+
+// Types
+
+type peer struct {
+	conn *transport.Connection
+	stop chan struct{}
+}
+
+type InboxMessage struct {
+	ID         string
+	Data       []byte
+	ReceivedAt time.Time
+}
+
+type PoolEventKind int
+
+const (
+	EventConnected PoolEventKind = iota
+	EventDisconnected
+)
+
+func (s PoolEventKind) String() string {
+	switch s {
+	case EventConnected:
+		return "connected"
+	case EventDisconnected:
+		return "disconnected"
+	default:
+		return "unknown"
+	}
+}
+
+type PoolEvent struct {
+	ID   string
+	Kind PoolEventKind
+}
+
+// Pool
+
+type Pool struct {
+	peers  map[string]*peer
+	inbox  chan InboxMessage
+	events chan PoolEvent
+	errors chan error
+	done   chan struct{}
+
+	dialer types.Dialer
+	config *PoolConfig
+	logger *slog.Logger
+
+	mu     sync.RWMutex
+	wg     sync.WaitGroup
+	closed bool
+}
+
+func NewPool(config *PoolConfig, logger *slog.Logger) (*Pool, error) {
+	if config == nil {
+		config = GetDefaultPoolConfig()
+	}
+
+	if err := ValidatePoolConfig(config); err != nil {
+		return nil, err
+	}
+
+	p := &Pool{
+		peers:  make(map[string]*peer),
+		inbox:  make(chan InboxMessage, 256),
+		events: make(chan PoolEvent, 10),
+		errors: make(chan error, 10),
+		done:   make(chan struct{}),
+		dialer: transport.NewDialer(),
+		config: config,
+		logger: logger,
+	}
+
+	return p, nil
+}
+
+func (p *Pool) Peers() map[string]*peer {
+	return p.peers
+}
+
+func (p *Pool) Inbox() chan InboxMessage {
+	return p.inbox
+}
+
+func (p *Pool) Events() chan PoolEvent {
+	return p.events
+}
+
+func (p *Pool) Errors() chan error {
+	return p.errors
+}
+
+func (p *Pool) Close() {
+	p.mu.Lock()
+	if p.closed {
+		p.mu.Unlock()
+		return
+	}
+
+	p.closed = true
+	close(p.done)
+
+	peers := p.peers
+	p.peers = make(map[string]*peer)
+
+	p.mu.Unlock()
+
+	for _, conn := range peers {
+		conn.conn.Close()
+	}
+
+	go func() {
+		p.wg.Wait()
+		close(p.inbox)
+		close(p.events)
+		close(p.errors)
+	}()
+}
+
+func (p *Pool) Connect(id string) error {
+	id, err := transport.NormalizeURL(id)
+	if err != nil {
+		return err
+	}
+
+	// Check for existing connection in pool
+	p.mu.Lock()
+	if p.closed {
+		p.mu.Unlock()
+		return NewPoolError("pool is closed")
+	}
+	_, exists := p.peers[id]
+	p.mu.Unlock()
+
+	if exists {
+		return NewPoolError("connection already exists")
+	}
+
+	// Create new connection
+	var logger *slog.Logger
+	if p.logger != nil {
+		logger = p.logger.With("id", id)
+	}
+	conn, err := transport.NewConnection(id, p.config.ConnectionConfig, logger)
+	if err != nil {
+		return err
+	}
+	conn.SetDialer(p.dialer)
+
+	// Attempt to connect
+	if err := conn.Connect(); err != nil {
+		return err
+	}
+
+	p.mu.Lock()
+	if p.closed {
+		// The pool closed while this connection was established.
+		p.mu.Unlock()
+		conn.Close()
+		return NewPoolError("pool is closed")
+	}
+
+	// Add connection to pool
+	stop := make(chan struct{})
+	if _, exists := p.peers[id]; exists {
+		// Another process connected to this url while this one was connecting
+		// Discard this connection and retain the existing one
+		p.mu.Unlock()
+		conn.Close()
+		return NewPoolError("connection already exists")
+	}
+	p.peers[id] = &peer{conn: conn, stop: stop}
+	p.mu.Unlock()
+
+	// TODO: start this connection's incoming message forwarder
+
+	select {
+	case p.events <- PoolEvent{ID: id, Kind: EventConnected}:
+	case <-p.done:
+		return nil
+	}
+
+	return nil
+}
+
+func (p *Pool) Remove(id string) error {
+	id, err := transport.NormalizeURL(id)
+	if err != nil {
+		return err
+	}
+
+	p.mu.Lock()
+	if p.closed {
+		p.mu.Unlock()
+		return NewPoolError("pool is closed")
+	}
+
+	peer, exists := p.peers[id]
+	if !exists {
+		p.mu.Unlock()
+		return NewPoolError("connection not found")
+	}
+	delete(p.peers, id)
+	p.mu.Unlock()
+
+	close(peer.stop)
+	peer.conn.Close()
+
+	select {
+	case p.events <- PoolEvent{ID: id, Kind: EventDisconnected}:
+	case <-p.done:
+		return nil
+	}
+
+	return nil
+}
+
+func (p *Pool) Send(id string, data []byte) error {
+	id, err := transport.NormalizeURL(id)
+	if err != nil {
+		return err
+	}
+
+	p.mu.RLock()
+	defer p.mu.RUnlock()
+
+	if p.closed {
+		return NewPoolError("pool is closed")
+	}
+
+	peer, exists := p.peers[id]
+	if !exists {
+		return NewPoolError("connection not found")
+	}
+
+	return peer.conn.Send(data)
+}

initiator/pool_test.go

@@ -0,0 +1,222 @@
+package initiator
+
+import (
+	"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"
+	"net/http"
+	"testing"
+	"time"
+)
+
+func TestPoolConnect(t *testing.T) {
+	t.Run("successfully adds connection", func(t *testing.T) {
+		mockSocket := honeybeetest.NewMockSocket()
+		mockDialer := &honeybeetest.MockDialer{
+			DialFunc: func(string, http.Header) (types.Socket, *http.Response, error) {
+				return mockSocket, nil, nil
+			},
+		}
+
+		pool, err := NewPool(nil, nil)
+		assert.NoError(t, err)
+
+		pool.dialer = mockDialer
+
+		err = pool.Connect("wss://test")
+		assert.NoError(t, err)
+
+		assert.Eventually(t, func() bool {
+			select {
+			case event := <-pool.events:
+				return event.ID == "wss://test" && event.Kind == EventConnected
+			default:
+				return false
+			}
+		}, honeybeetest.TestTimeout, honeybeetest.TestTick)
+
+		_, exists := pool.peers["wss://test"]
+		assert.True(t, exists)
+
+		pool.Close()
+	})
+
+	t.Run("does not add duplicate", func(t *testing.T) {
+		mockSocket := honeybeetest.NewMockSocket()
+		mockDialer := &honeybeetest.MockDialer{
+			DialFunc: func(string, http.Header) (types.Socket, *http.Response, error) {
+				return mockSocket, nil, nil
+			},
+		}
+
+		pool, err := NewPool(nil, nil)
+		assert.NoError(t, err)
+		pool.dialer = mockDialer
+
+		err = pool.Connect("wss://test")
+		assert.NoError(t, err)
+
+		// trailing slash normalizes to same key
+		err = pool.Connect("wss://test/")
+		assert.Error(t, err)
+		assert.ErrorContains(t, err, "already exists")
+
+		pool.mu.RLock()
+		assert.Len(t, pool.peers, 1)
+		pool.mu.RUnlock()
+
+		pool.Close()
+	})
+
+	t.Run("fails to add connection", func(t *testing.T) {
+		pool, err := NewPool(
+			&PoolConfig{
+				ConnectionConfig: &transport.ConnectionConfig{
+					Retry: &transport.RetryConfig{
+						MaxRetries:   1,
+						InitialDelay: 1 * time.Millisecond,
+						MaxDelay:     5 * time.Millisecond,
+					}},
+			}, nil)
+		assert.NoError(t, err)
+		pool.dialer = &honeybeetest.MockDialer{
+			DialFunc: func(string, http.Header) (types.Socket, *http.Response, error) {
+				return nil, nil, fmt.Errorf("dial failed")
+			},
+		}
+
+		err = pool.Connect("wss://test")
+		assert.Error(t, err)
+
+		pool.mu.RLock()
+		assert.Len(t, pool.peers, 0)
+		pool.mu.RUnlock()
+
+		select {
+		case event := <-pool.events:
+			t.Fatalf("unexpected event: %+v", event)
+		default:
+		}
+
+		pool.Close()
+	})
+}
+
+func TestPoolRemove(t *testing.T) {
+	t.Run("removes known url", func(t *testing.T) {
+		mockSocket := honeybeetest.NewMockSocket()
+		mockDialer := &honeybeetest.MockDialer{
+			DialFunc: func(string, http.Header) (types.Socket, *http.Response, error) {
+				return mockSocket, nil, nil
+			},
+		}
+
+		pool, err := NewPool(nil, nil)
+		assert.NoError(t, err)
+		pool.dialer = mockDialer
+
+		pool.Connect("wss://test")
+		expectEvent(t, pool.events, "wss://test", EventConnected)
+
+		err = pool.Remove("wss://test/")
+		assert.NoError(t, err)
+
+		// expect a disconnected event
+		expectEvent(t, pool.events, "wss://test", EventDisconnected)
+
+		// connection no longer in pool
+		pool.mu.Lock()
+		defer pool.mu.Unlock()
+		_, ok := pool.peers["wss://peer2"]
+		assert.False(t, ok, "connection is still in pool")
+	})
+
+	t.Run("unknown url returns error", func(t *testing.T) {
+		mockSocket := honeybeetest.NewMockSocket()
+		mockDialer := &honeybeetest.MockDialer{
+			DialFunc: func(string, http.Header) (types.Socket, *http.Response, error) {
+				return mockSocket, nil, nil
+			},
+		}
+
+		pool, err := NewPool(nil, nil)
+		assert.NoError(t, err)
+		pool.dialer = mockDialer
+
+		// remove unknown connection
+		err = pool.Remove("wss://unknown")
+		assert.ErrorContains(t, err, "connection not found")
+	})
+
+	t.Run("closed pool returns error", func(t *testing.T) {
+		mockSocket := honeybeetest.NewMockSocket()
+		mockDialer := &honeybeetest.MockDialer{
+			DialFunc: func(string, http.Header) (types.Socket, *http.Response, error) {
+				return mockSocket, nil, nil
+			},
+		}
+
+		pool, err := NewPool(nil, nil)
+		assert.NoError(t, err)
+		pool.dialer = mockDialer
+
+		// close pool
+		pool.Close()
+
+		// attempt to remove connection
+		err = pool.Remove("wss://test")
+		assert.ErrorContains(t, err, "pool is closed")
+	})
+
+}
+
+func TestPoolSend(t *testing.T) {
+	mockSocket := honeybeetest.NewMockSocket()
+	outgoingData := make(chan honeybeetest.MockOutgoingData, 10)
+	mockSocket.WriteMessageFunc = func(msgType int, data []byte) error {
+		outgoingData <- honeybeetest.MockOutgoingData{MsgType: msgType, Data: data}
+		return nil
+	}
+	mockDialer := &honeybeetest.MockDialer{
+		DialFunc: func(string, http.Header) (types.Socket, *http.Response, error) {
+			return mockSocket, nil, nil
+		},
+	}
+
+	pool, err := NewPool(nil, nil)
+	assert.NoError(t, err)
+	pool.dialer = mockDialer
+
+	err = pool.Connect("wss://test")
+	assert.NoError(t, err)
+	expectEvent(t, pool.events, "wss://test", EventConnected)
+
+	err = pool.Send("wss://test", []byte("hello"))
+	assert.NoError(t, err)
+
+	honeybeetest.ExpectWrite(t, outgoingData, websocket.TextMessage, []byte("hello"))
+
+	pool.Close()
+}
+
+func expectEvent(
+	t *testing.T,
+	events chan PoolEvent,
+	expectedURL string,
+	expectedKind PoolEventKind,
+) {
+	t.Helper()
+	assert.Eventually(t, func() bool {
+		select {
+		case e := <-events:
+			return e.ID == expectedURL && e.Kind == expectedKind
+		default:
+			return false
+		}
+	}, honeybeetest.TestTimeout, honeybeetest.TestTick,
+		fmt.Sprintf("expected event: URL=%q, Kind=%q",
+			expectedURL, expectedKind.String()))
+}

initiator/worker.go

@@ -0,0 +1,87 @@
+package initiator
+
+import (
+	"git.wisehodl.dev/jay/go-honeybee/transport"
+	"log/slog"
+	"sync"
+	"time"
+)
+
+// Types
+
+type WorkerContext struct {
+	Inbox    chan<- InboxMessage
+	Events   chan<- PoolEvent
+	Errors   chan<- error
+	Stop     <-chan struct{}
+	PoolDone <-chan struct{}
+	Logger   *slog.Logger
+}
+
+// Worker
+
+type Worker struct {
+	id          string
+	config      *WorkerConfig
+	onReconnect func() (*transport.Connection, error)
+}
+
+func NewWorker(
+	id string,
+	config *WorkerConfig,
+	onReconnect func() (*transport.Connection, error),
+	logger *slog.Logger,
+
+) (*Worker, error) {
+	w := &Worker{
+		id:          id,
+		config:      config,
+		onReconnect: onReconnect,
+	}
+
+	return w, nil
+}
+
+func (w *Worker) Start(
+	inbox chan<- InboxMessage,
+	events chan<- PoolEvent,
+	stop <-chan struct{},
+	poolDone <-chan struct{},
+	wg *sync.WaitGroup,
+) {
+}
+
+func (w *Worker) runReader(conn *transport.Connection,
+	messages chan<- []byte,
+	heartbeat chan<- time.Time,
+	reconnect chan<- struct{},
+	newConn <-chan *transport.Connection,
+	stop <-chan struct{},
+	poolDone <-chan struct{},
+
+) {
+}
+
+func (w *Worker) runForwarder(
+	messages <-chan []byte,
+	inbox chan<- []byte,
+	stop <-chan struct{},
+	poolDone <-chan struct{},
+	maxQueueSize int,
+) {
+}
+
+func (w *Worker) runHealthMonitor(
+	heartbeat <-chan time.Time,
+	stop <-chan struct{},
+	poolDone <-chan struct{},
+) {
+}
+
+func (w *Worker) runReconnector(
+	reconnect <-chan struct{},
+	newConn chan<- *transport.Connection,
+	stop <-chan struct{},
+	poolDone <-chan struct{},
+) {
+}

initiator/worker_test.go

@@ -0,0 +1 @@
+package initiator