From df14aeddaf8d27f00fa6a40f7c83d08af62d3b7a Mon Sep 17 00:00:00 2001 From: Jay Date: Sun, 7 Jun 2026 09:17:23 -0400 Subject: [PATCH] change id to url semantics --- pool.go | 54 ++++++++++++++++++++++++------------------------- pool_test.go | 10 ++++----- session.go | 12 +++++------ session_test.go | 6 +++--- types/types.go | 2 +- 5 files changed, 42 insertions(+), 42 deletions(-) diff --git a/pool.go b/pool.go index 4532a05..20798c0 100644 --- a/pool.go +++ b/pool.go @@ -46,7 +46,7 @@ const ( ) type PoolEvent struct { - ID string + URL string Kind PoolEventKind At time.Time Err error @@ -64,7 +64,7 @@ type PoolStats struct { } type PeerStats struct { - ID string + URL string Session SessionStats } @@ -153,7 +153,7 @@ func ValidatePoolConfig(c *PoolConfig) error { // ---------------------------------------------------------------------------- type Peer struct { - id string + url string session *session } @@ -238,9 +238,9 @@ func (p *Pool) Stats() PoolStats { count := len(p.peers) peerStats := make([]PeerStats, 0, count) - for id, peer := range p.peers { + for url, peer := range p.peers { peerStats = append(peerStats, PeerStats{ - ID: id, + URL: url, Session: peer.session.Stats(), }) } @@ -257,17 +257,17 @@ func (p *Pool) Stats() PoolStats { } } -func (p *Pool) PeerStats(id string) (PeerStats, error) { +func (p *Pool) PeerStats(url string) (PeerStats, error) { p.mu.RLock() defer p.mu.RUnlock() - peer, exists := p.peers[id] + peer, exists := p.peers[url] if !exists { return PeerStats{}, ErrPeerNotFound } return PeerStats{ - ID: id, + URL: url, Session: peer.session.Stats(), }, nil } @@ -317,12 +317,12 @@ func WithDialFunc(f func(context.Context) (types.Socket, error)) ConnectOption { } } -func (p *Pool) Connect(id string, opts ...ConnectOption) error { +func (p *Pool) Connect(url string, opts ...ConnectOption) error { if p.logger != nil { - p.logger.Info("connecting", "peer", id) + p.logger.Info("connecting", "peer", url) } - id, err := transport.NormalizeURL(id) + url, err := transport.NormalizeURL(url) if err != nil { return err } @@ -334,7 +334,7 @@ func (p *Pool) Connect(id string, opts ...ConnectOption) error { return ErrPoolClosed } - if _, exists := p.peers[id]; exists { + if _, exists := p.peers[url]; exists { return ErrPeerExists } @@ -348,7 +348,7 @@ func (p *Pool) Connect(id string, opts ...ConnectOption) error { hdr = p.config.RequestHeader.Clone() } dialFn := func(ctx context.Context) (types.Socket, error) { - socket, _, err := p.dialer.DialContext(ctx, id, hdr) + socket, _, err := p.dialer.DialContext(ctx, url, hdr) return socket, err } @@ -357,7 +357,7 @@ func (p *Pool) Connect(id string, opts ...ConnectOption) error { } sc := p.config.SessionConfig - session, err := newSession(p.ctx, id, dialFn, &sc, p.handler) + session, err := newSession(p.ctx, url, dialFn, &sc, p.handler) if err != nil { return err } @@ -372,9 +372,9 @@ func (p *Pool) Connect(id string, opts ...ConnectOption) error { p.mu.Unlock() return } - delete(p.peers, id) + delete(p.peers, url) p.mu.Unlock() - p.events <- PoolEvent{ID: id, Kind: EventRetired, Err: err, At: time.Now()} + p.events <- PoolEvent{URL: url, Kind: EventRetired, Err: err, At: time.Now()} }, } @@ -382,21 +382,21 @@ func (p *Pool) Connect(id string, opts ...ConnectOption) error { session.Start(pool) }) - p.peers[id] = &Peer{id: id, session: session} + p.peers[url] = &Peer{url: url, session: session} if p.logger != nil { - p.logger.Debug("registered peer", "peer", id) + p.logger.Debug("registered peer", "peer", url) } return nil } -func (p *Pool) Remove(id string) error { +func (p *Pool) Remove(url string) error { if p.logger != nil { - p.logger.Info("disconnecting", "peer", id) + p.logger.Info("disconnecting", "peer", url) } - id, err := transport.NormalizeURL(id) + url, err := transport.NormalizeURL(url) if err != nil { return err } @@ -408,23 +408,23 @@ func (p *Pool) Remove(id string) error { return ErrPoolClosed } - peer, exists := p.peers[id] + peer, exists := p.peers[url] if !exists { return ErrPeerNotFound } - delete(p.peers, id) + delete(p.peers, url) peer.session.Stop() if p.logger != nil { - p.logger.Debug("disconnected from peer", "peer", id) + p.logger.Debug("disconnected from peer", "peer", url) } return nil } -func (p *Pool) Send(id string, data []byte) error { - id, err := transport.NormalizeURL(id) +func (p *Pool) Send(url string, data []byte) error { + url, err := transport.NormalizeURL(url) if err != nil { return err } @@ -436,7 +436,7 @@ func (p *Pool) Send(id string, data []byte) error { return ErrPoolClosed } - peer, exists := p.peers[id] + peer, exists := p.peers[url] if !exists { return ErrPeerNotFound } diff --git a/pool_test.go b/pool_test.go index ac27825..f6adcac 100644 --- a/pool_test.go +++ b/pool_test.go @@ -41,7 +41,7 @@ func expectEvent( honeybeetest.Eventually(t, func() bool { select { case e := <-events: - return e.ID == expectedURL && e.Kind == expectedKind && !e.At.IsZero() + return e.URL == expectedURL && e.Kind == expectedKind && !e.At.IsZero() default: return false } @@ -60,7 +60,7 @@ func TestPoolConnect(t *testing.T) { honeybeetest.Eventually(t, func() bool { select { case event := <-pool.events: - return event.ID == "wss://test" && event.Kind == EventConnected + return event.URL == "wss://test" && event.Kind == EventConnected default: return false } @@ -116,7 +116,7 @@ func TestPoolConnectWithDialer(t *testing.T) { honeybeetest.Eventually(t, func() bool { select { case e := <-pool.events: - return e.ID == "wss://test" && e.Kind == EventConnected + return e.URL == "wss://test" && e.Kind == EventConnected default: return false } @@ -210,7 +210,7 @@ func TestPoolDialFailed(t *testing.T) { select { case e := <-pool.Events(): return e.Kind == EventDialFailed && - e.ID == "wss://test" && + e.URL == "wss://test" && e.Err == dialErr && !e.At.IsZero() default: @@ -264,7 +264,7 @@ func TestPoolRetire(t *testing.T) { honeybeetest.Eventually(t, func() bool { select { case e := <-pool.Events(): - return e.Kind == EventRetired && e.ID == "wss://test" + return e.Kind == EventRetired && e.URL == "wss://test" default: return false } diff --git a/session.go b/session.go index 10daf9d..eb90f44 100644 --- a/session.go +++ b/session.go @@ -218,11 +218,11 @@ func (w *session) dial( return } pool.Events <- PoolEvent{ - ID: w.url, Kind: EventDialFailed, Err: err, At: time.Now()} + URL: w.url, Kind: EventDialFailed, Err: err, At: time.Now()} } + socket, err := transport.DialWithRetry( - ctx, retryMgr, w.dialFn, onError, w.handler, - ) + ctx, retryMgr, w.dialFn, onError, w.handler) if err != nil { if pool.Retire != nil && !errors.Is(err, context.Canceled) && @@ -246,7 +246,7 @@ func (w *session) serve( // setup connection w.conn.Store(conn) - pool.Events <- PoolEvent{ID: w.url, Kind: EventConnected, At: time.Now()} + pool.Events <- PoolEvent{URL: w.url, Kind: EventConnected, At: time.Now()} // start keepalive service stopKeepalive, inactive, heartbeat := w.startKeepalive() @@ -273,7 +273,7 @@ session: } pool.Inbox <- types.InboxMessage{ - ID: w.url, Data: data, ReceivedAt: time.Now()} + URL: w.url, Data: data, ReceivedAt: time.Now()} pool.InboxCounter.Add(1) w.processedCount.Add(1) @@ -300,7 +300,7 @@ session: // tear down connection w.conn.Store(nil) - pool.Events <- PoolEvent{ID: w.url, Kind: EventDisconnected, At: time.Now()} + pool.Events <- PoolEvent{URL: w.url, Kind: EventDisconnected, At: time.Now()} if w.logger != nil { w.logger.Info("disconnected") diff --git a/session_test.go b/session_test.go index 12d88fb..0b31985 100644 --- a/session_test.go +++ b/session_test.go @@ -73,7 +73,7 @@ func TestSession(t *testing.T) { honeybeetest.Eventually(t, func() bool { select { case e := <-events: - return e.ID == w.url && e.Kind == EventConnected + return e.URL == w.url && e.Kind == EventConnected default: return false } @@ -334,7 +334,7 @@ func TestSession(t *testing.T) { return false } }, "expected message on Inbox") - assert.Equal(t, w.url, received.ID) + assert.Equal(t, w.url, received.URL) assert.Equal(t, []byte("hello"), received.Data) assert.False(t, received.ReceivedAt.IsZero(), "expected non-zero ReceivedAt") }) @@ -637,7 +637,7 @@ func TestSession(t *testing.T) { select { case e := <-events: return e.Kind == EventDialFailed && - e.ID == w.url && + e.URL == w.url && e.Err == dialErr && !e.At.IsZero() default: diff --git a/types/types.go b/types/types.go index 5847f1b..8618b37 100644 --- a/types/types.go +++ b/types/types.go @@ -31,7 +31,7 @@ type ReceivedMessage struct { } type InboxMessage struct { - ID string + URL string Data []byte ReceivedAt time.Time }