diff --git a/auth.go b/auth.go index c377b84..a1805bf 100644 --- a/auth.go +++ b/auth.go @@ -85,16 +85,16 @@ func (m *AuthManager) routeInbox() { m.mu.Unlock() now := time.Now() - m.envoy.Observer().Record(msg.ID, ChallengeReceived{Challenge: challenge, At: now}) + m.envoy.Observer().Record(msg.URL, ChallengeReceived{Challenge: challenge, At: now}) signed, err := m.respond(challenge) if err != nil { - m.envoy.Observer().Record(msg.ID, AuthResponseFailed{Err: err, At: time.Now()}) + m.envoy.Observer().Record(msg.URL, AuthResponseFailed{Err: err, At: time.Now()}) continue } if err := m.envoy.Send([]byte(envelope.EncloseAuthResponse(signed))); err != nil { - m.envoy.Observer().Record(msg.ID, AuthResponseFailed{Err: err, At: time.Now()}) + m.envoy.Observer().Record(msg.URL, AuthResponseFailed{Err: err, At: time.Now()}) continue } @@ -102,7 +102,7 @@ func (m *AuthManager) routeInbox() { m.logger.Info("responded to auth", "challenge", challenge) } - m.envoy.Observer().Record(msg.ID, AuthResponseSent{At: time.Now()}) + m.envoy.Observer().Record(msg.URL, AuthResponseSent{At: time.Now()}) } } } diff --git a/embassy.go b/embassy.go index d2ff8bb..5cb9cb6 100644 --- a/embassy.go +++ b/embassy.go @@ -30,6 +30,7 @@ const ( EventConnected EmbassyEventKind = iota EventDisconnected EventDialFailed + EventRetired EventEmbassyUnknown ) @@ -41,20 +42,22 @@ func mapEmbassyEvent(kind honeybee.PoolEventKind) EmbassyEventKind { return EventDisconnected case honeybee.EventDialFailed: return EventDialFailed + case honeybee.EventRetired: + return EventRetired default: return EventEmbassyUnknown } } type PoolEvent struct { - ID string + URL string Kind EmbassyEventKind At time.Time Err error } type InboxMessage struct { - ID string + URL string Data []byte ReceivedAt time.Time } @@ -249,7 +252,7 @@ func (e *Embassy) routeEvents() { return } - url, err := honeybee.NormalizeURL(ev.ID) + url, err := honeybee.NormalizeURL(ev.URL) if err != nil { continue } @@ -266,7 +269,7 @@ func (e *Embassy) routeEvents() { case <-e.ctx.Done(): return case sub <- PoolEvent{ - ID: ev.ID, Kind: mapEmbassyEvent(ev.Kind), At: ev.At, Err: ev.Err, + URL: ev.URL, Kind: mapEmbassyEvent(ev.Kind), At: ev.At, Err: ev.Err, }: } } @@ -283,7 +286,7 @@ func (e *Embassy) routeInbox() { return } - url, err := honeybee.NormalizeURL(ev.ID) + url, err := honeybee.NormalizeURL(ev.URL) if err != nil { continue } @@ -300,7 +303,7 @@ func (e *Embassy) routeInbox() { case <-e.ctx.Done(): return case sub <- InboxMessage{ - ID: ev.ID, Data: ev.Data, ReceivedAt: ev.ReceivedAt}: + URL: ev.URL, Data: ev.Data, ReceivedAt: ev.ReceivedAt}: } } } @@ -318,9 +321,9 @@ type Envoy struct { send func(data []byte) error events <-chan PoolEvent inbox <-chan InboxMessage - eventSubs []chan<- PoolEvent - labelledInboxSubs map[string][]chan<- InboxMessage - inboxSubs []chan<- InboxMessage + eventSubs []chan PoolEvent + labelledInboxSubs map[string][]chan InboxMessage + inboxSubs []chan InboxMessage ctx context.Context cancel context.CancelFunc @@ -350,7 +353,7 @@ func newEnvoy( send: send, events: events, inbox: inbox, - labelledInboxSubs: make(map[string][]chan<- InboxMessage), + labelledInboxSubs: make(map[string][]chan InboxMessage), ctx: ctx, cancel: cancel, observer: observer, @@ -378,7 +381,7 @@ func (e *Envoy) Context() context.Context { return e.ctx } -func (e *Envoy) PeerID() string { +func (e *Envoy) URL() string { return e.url } @@ -408,7 +411,7 @@ func (e *Envoy) Dismiss() { e.eventSubs = nil e.inboxSubs = nil - e.labelledInboxSubs = make(map[string][]chan<- InboxMessage) + e.labelledInboxSubs = make(map[string][]chan InboxMessage) } func (e *Envoy) Send(data []byte) error { @@ -423,6 +426,18 @@ func (e *Envoy) SubscribeEvents() <-chan PoolEvent { return ch } +func (e *Envoy) UnsubscribeEvents(ch <-chan PoolEvent) { + e.mu.Lock() + defer e.mu.Unlock() + for i, sub := range e.eventSubs { + if sub == ch { + e.eventSubs[i] = e.eventSubs[len(e.eventSubs)-1] + e.eventSubs = e.eventSubs[:len(e.eventSubs)-1] + return + } + } +} + func (e *Envoy) SubscribeInbox(labels []string) <-chan InboxMessage { e.mu.Lock() defer e.mu.Unlock() @@ -430,7 +445,7 @@ func (e *Envoy) SubscribeInbox(labels []string) <-chan InboxMessage { e.inboxSubs = append(e.inboxSubs, ch) for _, label := range labels { if _, ok := e.labelledInboxSubs[label]; !ok { - e.labelledInboxSubs[label] = make([]chan<- InboxMessage, 0) + e.labelledInboxSubs[label] = make([]chan InboxMessage, 0) } e.labelledInboxSubs[label] = append(e.labelledInboxSubs[label], ch) } diff --git a/embassy_test.go b/embassy_test.go index b630f73..f4e1383 100644 --- a/embassy_test.go +++ b/embassy_test.go @@ -42,16 +42,16 @@ func TestEmbassy_Dispatch(t *testing.T) { }() p.events <- honeybee.PoolEvent{ - ID: p.url, Kind: honeybee.EventConnected, At: time.Now()} + URL: p.url, Kind: honeybee.EventConnected, At: time.Now()} p.events <- honeybee.PoolEvent{ - ID: "wss://other", Kind: honeybee.EventConnected, At: time.Now()} + URL: "wss://other", Kind: honeybee.EventConnected, At: time.Now()} p.inbox <- honeybee.InboxMessage{ - ID: p.url, + URL: p.url, Data: envelope.EncloseEvent([]byte("{}")), ReceivedAt: time.Now(), } p.inbox <- honeybee.InboxMessage{ - ID: "wss://other", + URL: "wss://other", Data: envelope.EncloseEvent([]byte("{}")), ReceivedAt: time.Now(), } @@ -114,7 +114,7 @@ func TestEmbassy_DialFailed_Routing(t *testing.T) { assert.Equal(t, EventDialFailed, got.Kind) assert.ErrorIs(t, got.Err, dialErr) - assert.Equal(t, p.url, got.ID) + assert.Equal(t, p.url, got.URL) }) t.Run("not routed to unrelated envoy subscriber", func(t *testing.T) { diff --git a/envoy_test.go b/envoy_test.go index 9fa40c9..a63971c 100644 --- a/envoy_test.go +++ b/envoy_test.go @@ -75,7 +75,7 @@ func TestEnvoy_IsConnected(t *testing.T) { }() events <- PoolEvent{ - ID: url, Kind: EventConnected, At: time.Now()} + URL: url, Kind: EventConnected, At: time.Now()} Eventually(t, func() bool { return envoy.IsConnected() @@ -101,7 +101,7 @@ func TestEnvoy_Events(t *testing.T) { }() events <- PoolEvent{ - ID: url, Kind: EventConnected, At: time.Now()} + URL: url, Kind: EventConnected, At: time.Now()} Eventually(t, func() bool { mu.RLock() @@ -133,12 +133,12 @@ func TestEnvoy_Inbox(t *testing.T) { }() inbox <- InboxMessage{ - ID: url, + URL: url, Data: envelope.EncloseEvent([]byte("{}")), ReceivedAt: time.Now(), } inbox <- InboxMessage{ - ID: url, + URL: url, Data: envelope.EncloseOK("id", true, "ok"), ReceivedAt: time.Now(), } diff --git a/helpers_test.go b/helpers_test.go index 0b6b760..521fea8 100644 --- a/helpers_test.go +++ b/helpers_test.go @@ -87,22 +87,22 @@ func newMockPool(t *testing.T) *mockPool { func (p *mockPool) connect() { p.events <- honeybee.PoolEvent{ - ID: p.url, Kind: honeybee.EventConnected, At: time.Now()} + URL: p.url, Kind: honeybee.EventConnected, At: time.Now()} } func (p *mockPool) disconnect() { p.events <- honeybee.PoolEvent{ - ID: p.url, Kind: honeybee.EventDisconnected, At: time.Now()} + URL: p.url, Kind: honeybee.EventDisconnected, At: time.Now()} } func (p *mockPool) dialFail(err error) { p.events <- honeybee.PoolEvent{ - ID: p.url, Kind: honeybee.EventDialFailed, Err: err, At: time.Now()} + URL: p.url, Kind: honeybee.EventDialFailed, Err: err, At: time.Now()} } func (p *mockPool) receive(data []byte) { p.inbox <- honeybee.InboxMessage{ - ID: p.url, + URL: p.url, Data: data, ReceivedAt: time.Now(), } diff --git a/notice.go b/notice.go index 8bbaffe..21062fc 100644 --- a/notice.go +++ b/notice.go @@ -14,7 +14,7 @@ type NoticeReceived struct { } type Notice struct { - PeerID string + Peer string Message string Timestamp time.Time } @@ -62,11 +62,11 @@ func (h *NoticeHandler) route() { } now := time.Now() h.notices <- Notice{ - PeerID: msg.ID, + Peer: msg.URL, Message: message, Timestamp: now, } - h.envoy.Observer().Record(msg.ID, NoticeReceived{ + h.envoy.Observer().Record(msg.URL, NoticeReceived{ Message: message, At: now, }) diff --git a/notice_test.go b/notice_test.go index b480f21..2963d23 100644 --- a/notice_test.go +++ b/notice_test.go @@ -26,7 +26,7 @@ func TestNoticeHandler(t *testing.T) { }, "notice not delivered") assert.Equal(t, "hello", got.Message) - assert.Equal(t, p.url, got.PeerID) + assert.Equal(t, p.url, got.Peer) assert.False(t, got.Timestamp.IsZero()) }) diff --git a/publish.go b/publish.go index 247cddc..b80c4cf 100644 --- a/publish.go +++ b/publish.go @@ -103,7 +103,7 @@ func (p *EventPublisher) Publish(eventID string, eventJSON []byte, timeout time. p.mu.Lock() p.deregister(eventID) p.mu.Unlock() - p.envoy.Observer().Record(p.envoy.PeerID(), + p.envoy.Observer().Record(p.envoy.URL(), PublishTimeout{EventID: eventID, At: time.Now()}) p.deliver(entry, publishResult{err: errors.New("publish timeout")}) }) @@ -148,12 +148,12 @@ func (p *EventPublisher) trySend(entry *pendingEntry) error { err := p.envoy.Send([]byte(envelope.EncloseEvent(entry.eventJSON))) if err != nil { - p.envoy.Observer().Record(p.envoy.PeerID(), + p.envoy.Observer().Record(p.envoy.URL(), PublishSendFailed{EventID: entry.eventID, Err: err, At: time.Now()}) return err } entry.sent = true - p.envoy.Observer().Record(p.envoy.PeerID(), + p.envoy.Observer().Record(p.envoy.URL(), PublishDispatched{EventID: entry.eventID, At: time.Now()}) return nil } @@ -188,10 +188,10 @@ func (p *EventPublisher) routeInbox() { } if accepted { - p.envoy.Observer().Record(msg.ID, + p.envoy.Observer().Record(msg.URL, PublishAccepted{EventID: eventID, At: time.Now()}) } else { - p.envoy.Observer().Record(msg.ID, + p.envoy.Observer().Record(msg.URL, PublishRejected{EventID: eventID, Message: message, At: time.Now()}) } p.deliver(entry, publishResult{accepted: accepted, message: message}) diff --git a/request.go b/request.go index 5640a74..f0abfe5 100644 --- a/request.go +++ b/request.go @@ -18,13 +18,13 @@ import ( // ---------------------------------------------------------------------------- type ReqEvent struct { - PeerID string + Peer string ReceivedAt time.Time Data []byte } type ReqClosed struct { - PeerID string + Peer string ReceivedAt time.Time Data string } @@ -34,38 +34,38 @@ type ReqClosed struct { // ---------------------------------------------------------------------------- type ReqDispatched struct { - SubID string + ID string DispatchedAt time.Time } type ReqSendFailed struct { - SubID string - Err error - At time.Time + ID string + Err error + At time.Time } type FirstEventReceived struct { - SubID string + ID string ReceivedAt time.Time } type StreamEOSEReceived struct { - SubID string + ID string ReceivedAt time.Time } type QueryEOSEReceived struct { - SubID string + ID string ReceivedAt time.Time } type MissedEOSE struct { - SubID string - At time.Time + ID string + At time.Time } type ClosedReceived struct { - SubID string + ID string ReceivedAt time.Time Message string } @@ -259,8 +259,8 @@ func (m *RequestManager) Query( return result, &cl, nil case <-ctx.Done(): // query timed out - m.observer.Record(m.envoy.PeerID(), - MissedEOSE{SubID: id, At: time.Now()}) + m.observer.Record(m.envoy.URL(), + MissedEOSE{ID: id, At: time.Now()}) if m.logger != nil { m.logger.Warn("missed eose", "req", id) } @@ -328,15 +328,15 @@ func (m *RequestManager) activate(req *request) { go func() { err := m.envoy.Send(envelope.EncloseReq(req.id, req.filters)) if err != nil { - m.observer.Record(m.envoy.PeerID(), - ReqSendFailed{SubID: req.id, Err: err, At: time.Now()}) + m.observer.Record(m.envoy.URL(), + ReqSendFailed{ID: req.id, Err: err, At: time.Now()}) if m.logger != nil { m.logger.Warn("req send failed", "req", req.id, "error", err) } return } - m.observer.Record(m.envoy.PeerID(), - ReqDispatched{SubID: req.id, DispatchedAt: time.Now()}) + m.observer.Record(m.envoy.URL(), + ReqDispatched{ID: req.id, DispatchedAt: time.Now()}) if m.logger != nil { m.logger.Debug("req sent", "req", req.id) } @@ -424,13 +424,13 @@ func (m *RequestManager) routeEvent(msg InboxMessage) { req.firstEventSeen = true reqSubID := req.id receivedAt := msg.ReceivedAt - go m.observer.Record(m.envoy.PeerID(), - FirstEventReceived{SubID: reqSubID, ReceivedAt: receivedAt}) + go m.observer.Record(m.envoy.URL(), + FirstEventReceived{ID: reqSubID, ReceivedAt: receivedAt}) } m.mu.Unlock() req.buffer <- ReqEvent{ - PeerID: msg.ID, + Peer: msg.URL, ReceivedAt: msg.ReceivedAt, Data: event, } @@ -453,11 +453,11 @@ func (m *RequestManager) routeEOSE(msg InboxMessage) { reqSubID := req.id receivedAt := msg.ReceivedAt if req.isQuery { - go m.observer.Record(m.envoy.PeerID(), - QueryEOSEReceived{SubID: reqSubID, ReceivedAt: receivedAt}) + go m.observer.Record(m.envoy.URL(), + QueryEOSEReceived{ID: reqSubID, ReceivedAt: receivedAt}) } else { - go m.observer.Record(m.envoy.PeerID(), - StreamEOSEReceived{SubID: reqSubID, ReceivedAt: receivedAt}) + go m.observer.Record(m.envoy.URL(), + StreamEOSEReceived{ID: reqSubID, ReceivedAt: receivedAt}) } if req.active && req.isQuery { // manually cleanup query state @@ -493,9 +493,9 @@ func (m *RequestManager) routeClosed(msg InboxMessage) { } reqSubID := req.id receivedAt := msg.ReceivedAt - go m.observer.Record(m.envoy.PeerID(), + go m.observer.Record(m.envoy.URL(), ClosedReceived{ - SubID: reqSubID, + ID: reqSubID, ReceivedAt: receivedAt, Message: message, }) @@ -504,7 +504,7 @@ func (m *RequestManager) routeClosed(msg InboxMessage) { } req.closedOnce.Do(func() { req.closed <- ReqClosed{ - PeerID: msg.ID, + Peer: msg.URL, ReceivedAt: msg.ReceivedAt, Data: message, } diff --git a/request_test.go b/request_test.go index d24cbc1..2d11584 100644 --- a/request_test.go +++ b/request_test.go @@ -111,7 +111,7 @@ func TestRequestManager_Stream(t *testing.T) { return len(EventsOf[ReqDispatched](obs)) == 1 }, "expected ReqDispatched observable") dispatched := EventsOf[ReqDispatched](obs) - assert.Equal(t, id, dispatched[0].SubID) + assert.Equal(t, id, dispatched[0].ID) }) t.Run("does not send req when disconnected", func(t *testing.T) { @@ -186,7 +186,7 @@ func TestRequestManager_Stream(t *testing.T) { return len(EventsOf[FirstEventReceived](obs)) == 1 }, "expected FirstEventReceived observable") first := EventsOf[FirstEventReceived](obs) - assert.Equal(t, id, first[0].SubID) + assert.Equal(t, id, first[0].ID) assert.False(t, first[0].ReceivedAt.IsZero()) }) @@ -247,7 +247,7 @@ func TestRequestManager_Stream(t *testing.T) { return len(EventsOf[StreamEOSEReceived](obs)) == 1 }, "expected StreamEOSEReceived observable") eoseEvents := EventsOf[StreamEOSEReceived](obs) - assert.Equal(t, id, eoseEvents[0].SubID) + assert.Equal(t, id, eoseEvents[0].ID) }) t.Run("closed deregisters and signals caller", func(t *testing.T) { @@ -289,7 +289,7 @@ func TestRequestManager_Stream(t *testing.T) { return len(EventsOf[ClosedReceived](obs)) == 1 }, "expected ClosedReceived observable") closedEvents := EventsOf[ClosedReceived](obs) - assert.Equal(t, id, closedEvents[0].SubID) + assert.Equal(t, id, closedEvents[0].ID) assert.Equal(t, "error: test", closedEvents[0].Message) Eventually(t, func() bool { @@ -342,7 +342,7 @@ func TestRequestManager_Stream(t *testing.T) { return len(EventsOf[ReqSendFailed](obs)) >= 1 }, "expected ReqSendFailed observable") failed := EventsOf[ReqSendFailed](obs) - assert.Equal(t, id, failed[0].SubID) + assert.Equal(t, id, failed[0].ID) assert.NotNil(t, failed[0].Err) Never(t, func() bool { @@ -486,7 +486,7 @@ func TestRequestManager_Query(t *testing.T) { return len(EventsOf[QueryEOSEReceived](obs)) == 1 }, "expected QueryEOSEReceived observable") qeose := EventsOf[QueryEOSEReceived](obs) - assert.Equal(t, querySubID, qeose[0].SubID) + assert.Equal(t, querySubID, qeose[0].ID) // CLOSE envelope should have been sent after EOSE var closeEnv []byte