jay/go-mana-prism
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

align with honeybee updates. change ID to URL semantics.

Browse Source
This commit is contained in:
Jay
2026-06-08 09:43:19 -04:00
parent 69f9219819
commit 42e6e2af83
10 changed files with 88 additions and 73 deletions
Download Patch File Download Diff File Expand all files Collapse all files
auth.go
+4 -4
View File
@@ -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()})
}
}
}
embassy.go
+28 -13
View File
@@ -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)
}
embassy_test.go
+5 -5
View File
@@ -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) {
envoy_test.go
+4 -4
View File
@@ -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(),
}
helpers_test.go
+4 -4
View File
@@ -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(),
}
notice.go
+3 -3
View File
@@ -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,
})
notice_test.go
+1 -1
View File
@@ -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())
})
publish.go
+5 -5
View File
@@ -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})
request.go
+28 -28
View File
@@ -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,
}
request_test.go
+6 -6
View File
@@ -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