prepare code skeleton to develop remaining managers
This commit is contained in:
Jay
@@ -0,0 +1,67 @@
|
||||
package prism
|
||||
|
||||
import (
|
||||
"context"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
type ChallengeReceived struct {
|
||||
Challenge string
|
||||
At time.Time
|
||||
}
|
||||
|
||||
type AuthResponseSent struct {
|
||||
At time.Time
|
||||
}
|
||||
|
||||
type AuthResponseFailed struct {
|
||||
Err error
|
||||
At time.Time
|
||||
}
|
||||
|
||||
type AuthManager struct {
|
||||
envoy *Envoy
|
||||
respond func(challenge string) ([]byte, error)
|
||||
challenge string
|
||||
|
||||
inbox <-chan InboxMessage
|
||||
events <-chan OutboundPoolEvent
|
||||
|
||||
ctx context.Context
|
||||
cancel context.CancelFunc
|
||||
mu sync.Mutex
|
||||
wg sync.WaitGroup
|
||||
}
|
||||
|
||||
func NewAuthManager(e *Envoy, respond func(string) ([]byte, error)) *AuthManager {
|
||||
// SubscribeInbox([]string{"AUTH"})
|
||||
// SubscribeEvents()
|
||||
// context.WithCancel from e.Context()
|
||||
// wg.Add(2); go m.routeInbox(); go m.handleEvents()
|
||||
return nil
|
||||
}
|
||||
|
||||
func (m *AuthManager) Close() {
|
||||
// m.cancel(); m.wg.Wait()
|
||||
}
|
||||
|
||||
func (m *AuthManager) routeInbox() {
|
||||
defer m.wg.Done()
|
||||
// select ctx.Done | inbox msg
|
||||
// msg: envelope.FindAuthChallenge → on ErrWrongFieldType or err: continue
|
||||
// store m.challenge with lock
|
||||
// record ChallengeReceived
|
||||
// call m.respond(challenge)
|
||||
// on err: record AuthResponseFailed; continue
|
||||
// envelope.EncloseAuthResponse(signed)
|
||||
// m.envoy.Send(...)
|
||||
// on err: record AuthResponseFailed; continue
|
||||
// record AuthResponseSent
|
||||
}
|
||||
|
||||
func (m *AuthManager) handleEvents() {
|
||||
defer m.wg.Done()
|
||||
// select ctx.Done | event
|
||||
// EventDisconnected: m.challenge = "" with lock
|
||||
}
|
||||
@@ -0,0 +1,58 @@
|
||||
package prism
|
||||
|
||||
import (
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestAuthManager(t *testing.T) {
|
||||
t.Run("challenge triggers callback and send", func(t *testing.T) {
|
||||
// p, envoy := newMockEnvoy(t)
|
||||
// signed := []byte(`{"kind":22242,...}`)
|
||||
// callback records invocation, returns signed
|
||||
// m := NewAuthManager(envoy, callback)
|
||||
// t.Cleanup(m.Close)
|
||||
// p.connect(); p.receive(envelope.EncloseAuthChallenge("abc"))
|
||||
// Eventually: callback called with "abc"
|
||||
// Eventually: p.sent contains AUTH response envelope wrapping signed
|
||||
// assert ChallengeReceived and AuthResponseSent recorded on observer
|
||||
})
|
||||
|
||||
t.Run("callback error recorded", func(t *testing.T) {
|
||||
// callback returns errors.New("signing failed")
|
||||
// p.receive(envelope.EncloseAuthChallenge("abc"))
|
||||
// Never: anything on p.sent
|
||||
// assert ChallengeReceived and AuthResponseFailed recorded
|
||||
})
|
||||
|
||||
t.Run("send error recorded", func(t *testing.T) {
|
||||
// callback succeeds
|
||||
// send returns error
|
||||
// assert AuthResponseFailed recorded
|
||||
})
|
||||
|
||||
t.Run("disconnect resets state", func(t *testing.T) {
|
||||
// p.connect(); feed AUTH "abc"; callback fires
|
||||
// p.disconnect()
|
||||
// assert m.challenge == "" (inspect via exported accessor or package-level test access)
|
||||
// p.connect(); feed AUTH "def"
|
||||
// assert callback fires again with "def"
|
||||
})
|
||||
|
||||
t.Run("replacement challenge", func(t *testing.T) {
|
||||
// feed AUTH "abc"; feed AUTH "def" without disconnect
|
||||
// assert callback called twice; second call receives "def"
|
||||
// assert two AUTH response envelopes sent
|
||||
})
|
||||
|
||||
t.Run("malformed AUTH ignored", func(t *testing.T) {
|
||||
// p.receive(envelope.EncloseAuthResponse([]byte("{}"))) — second element is object
|
||||
// Never: callback called
|
||||
// assert no panic
|
||||
})
|
||||
|
||||
t.Run("close cleans up", func(t *testing.T) {
|
||||
// m.Close()
|
||||
// feed another AUTH
|
||||
// Never: callback invoked after close
|
||||
})
|
||||
}
|
||||
+81
-27
@@ -5,11 +5,14 @@ import (
|
||||
"git.wisehodl.dev/jay/go-honeybee"
|
||||
"git.wisehodl.dev/jay/go-mana-component"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// ----------------------------------------------------------------------------
|
||||
// Async Helpers
|
||||
// ----------------------------------------------------------------------------
|
||||
|
||||
const (
|
||||
TestTimeout = 2 * time.Second
|
||||
@@ -27,6 +30,10 @@ func Never(t *testing.T, condition func() bool, msg string) {
|
||||
assert.Never(t, condition, NegativeTestTimeout, TestTick, msg)
|
||||
}
|
||||
|
||||
// ----------------------------------------------------------------------------
|
||||
// Mock Pool
|
||||
// ----------------------------------------------------------------------------
|
||||
|
||||
type mockPool struct {
|
||||
plugin EmbassyPlugin
|
||||
ctx context.Context
|
||||
@@ -36,40 +43,46 @@ type mockPool struct {
|
||||
events chan honeybee.PoolEvent
|
||||
inbox chan honeybee.InboxMessage
|
||||
sent chan []byte
|
||||
sendMu sync.Mutex
|
||||
sendErr error
|
||||
}
|
||||
|
||||
// Mock Pool
|
||||
|
||||
func newMockPool(t *testing.T) *mockPool {
|
||||
t.Helper()
|
||||
|
||||
ctx := component.MustNew(context.Background(), "prism", "test")
|
||||
url := "wss://test"
|
||||
p := &mockPool{}
|
||||
|
||||
added := make(chan struct{})
|
||||
removed := make(chan struct{})
|
||||
events := make(chan honeybee.PoolEvent, 16)
|
||||
inbox := make(chan honeybee.InboxMessage, 16)
|
||||
sent := make(chan []byte, 16)
|
||||
p.ctx = component.MustNew(context.Background(), "prism", "test")
|
||||
p.url = "wss://test"
|
||||
|
||||
plugin := EmbassyPlugin{
|
||||
Connect: func(url string) error { close(added); return nil },
|
||||
Remove: func(url string) error { close(removed); return nil },
|
||||
Send: func(_ string, data []byte) error { sent <- data; return nil },
|
||||
Events: events,
|
||||
Inbox: inbox,
|
||||
p.added = make(chan struct{})
|
||||
p.removed = make(chan struct{})
|
||||
p.events = make(chan honeybee.PoolEvent, 16)
|
||||
p.inbox = make(chan honeybee.InboxMessage, 16)
|
||||
p.sent = make(chan []byte, 16)
|
||||
sendFunc := func(_ string, data []byte) error {
|
||||
p.sendMu.Lock()
|
||||
err := p.sendErr
|
||||
p.sendMu.Unlock()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
p.sent <- data
|
||||
return nil
|
||||
}
|
||||
|
||||
return &mockPool{
|
||||
plugin: plugin,
|
||||
ctx: ctx,
|
||||
url: url,
|
||||
added: added,
|
||||
removed: removed,
|
||||
events: events,
|
||||
inbox: inbox,
|
||||
sent: sent,
|
||||
p.plugin = EmbassyPlugin{
|
||||
Connect: func(url string, opts ...honeybee.ConnectOption) error {
|
||||
close(p.added)
|
||||
return nil
|
||||
},
|
||||
Remove: func(url string) error { close(p.removed); return nil },
|
||||
Send: sendFunc,
|
||||
Events: p.events,
|
||||
Inbox: p.inbox,
|
||||
}
|
||||
|
||||
return p
|
||||
}
|
||||
|
||||
func (p *mockPool) connect() {
|
||||
@@ -90,15 +103,56 @@ func (p *mockPool) receive(data []byte) {
|
||||
}
|
||||
}
|
||||
|
||||
// MockEnvoy
|
||||
func (p *mockPool) setSendError(err error) {
|
||||
p.sendMu.Lock()
|
||||
defer p.sendMu.Unlock()
|
||||
p.sendErr = err
|
||||
}
|
||||
|
||||
func newMockEnvoy(t *testing.T) (*mockPool, *Envoy) {
|
||||
// ----------------------------------------------------------------------------
|
||||
// MockEnvoy
|
||||
// ----------------------------------------------------------------------------
|
||||
|
||||
func newMockEnvoy(t *testing.T, opts ...EmbassyOption) (*mockPool, *Envoy) {
|
||||
t.Helper()
|
||||
|
||||
p := newMockPool(t)
|
||||
emb := NewEmbassy(p.ctx, p.plugin)
|
||||
emb := NewEmbassy(p.ctx, p.plugin, opts...)
|
||||
err := emb.Dispatch(p.url)
|
||||
assert.NoError(t, err)
|
||||
envoy := emb.Call(p.url)
|
||||
return p, envoy
|
||||
}
|
||||
|
||||
// ----------------------------------------------------------------------------
|
||||
// MockObserver
|
||||
// ----------------------------------------------------------------------------
|
||||
|
||||
type observerRecord struct {
|
||||
peerID string
|
||||
event any
|
||||
}
|
||||
|
||||
type mockObserver struct {
|
||||
mu sync.Mutex
|
||||
records []observerRecord
|
||||
}
|
||||
|
||||
func (o *mockObserver) Record(peerID string, event any) {
|
||||
o.mu.Lock()
|
||||
defer o.mu.Unlock()
|
||||
o.records = append(o.records, observerRecord{peerID, event})
|
||||
}
|
||||
|
||||
// EventsOf returns all recorded events of type T.
|
||||
func EventsOf[T any](o *mockObserver) []T {
|
||||
o.mu.Lock()
|
||||
defer o.mu.Unlock()
|
||||
var out []T
|
||||
for _, r := range o.records {
|
||||
if v, ok := r.event.(T); ok {
|
||||
out = append(out, v)
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
@@ -0,0 +1,52 @@
|
||||
package prism
|
||||
|
||||
import (
|
||||
"context"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
type NoticeReceived struct {
|
||||
Message string
|
||||
At time.Time
|
||||
}
|
||||
|
||||
type Notice struct {
|
||||
PeerID string
|
||||
Message string
|
||||
Timestamp time.Time
|
||||
}
|
||||
|
||||
type NoticeHandler struct {
|
||||
envoy *Envoy
|
||||
inbox <-chan InboxMessage
|
||||
notices chan Notice
|
||||
|
||||
ctx context.Context
|
||||
cancel context.CancelFunc
|
||||
wg sync.WaitGroup
|
||||
}
|
||||
|
||||
func NewNoticeHandler(e *Envoy) *NoticeHandler {
|
||||
// SubscribeInbox([]string{"NOTICE"})
|
||||
// make(chan Notice, 16)
|
||||
// context.WithCancel from e.Context()
|
||||
// wg.Add(1); go h.route()
|
||||
return nil
|
||||
}
|
||||
|
||||
func (h *NoticeHandler) Notices() <-chan Notice { return h.notices }
|
||||
|
||||
func (h *NoticeHandler) Close() {
|
||||
// h.cancel()
|
||||
// h.wg.Wait()
|
||||
// close(h.notices)
|
||||
}
|
||||
|
||||
func (h *NoticeHandler) route() {
|
||||
defer h.wg.Done()
|
||||
// select ctx.Done | inbox msg
|
||||
// msg: envelope.FindNotice → on err continue
|
||||
// push Notice{PeerID, Message, time.Now()} to h.notices (blocking)
|
||||
// e.Observer().Record(e.PeerID(), NoticeReceived{...})
|
||||
}
|
||||
@@ -0,0 +1,33 @@
|
||||
package prism
|
||||
|
||||
import (
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestNoticeHandler(t *testing.T) {
|
||||
t.Run("notice delivered", func(t *testing.T) {
|
||||
// p, envoy := newMockEnvoy(t)
|
||||
// h := NewNoticeHandler(envoy)
|
||||
// t.Cleanup(h.Close)
|
||||
// p.receive(envelope.EncloseNotice("hello"))
|
||||
// Eventually: receive from h.Notices()
|
||||
// assert Message == "hello", PeerID == p.url, Timestamp non-zero
|
||||
})
|
||||
|
||||
t.Run("malformed ignored", func(t *testing.T) {
|
||||
// p.receive([]byte("not json"))
|
||||
// Never: nothing readable from h.Notices() within NegativeTestTimeout
|
||||
// assert no panic
|
||||
})
|
||||
|
||||
t.Run("observer notified", func(t *testing.T) {
|
||||
// wire mockObserver via envoy (requires constructor option or mock envoy)
|
||||
// p.receive(envelope.EncloseNotice("hello"))
|
||||
// Eventually: mockObserver recorded NoticeReceived{Message: "hello"}
|
||||
})
|
||||
|
||||
t.Run("close cleans up", func(t *testing.T) {
|
||||
// h.Close()
|
||||
// assert <-h.Notices() returns zero Notice, ok == false
|
||||
})
|
||||
}
|
||||
+139
@@ -0,0 +1,139 @@
|
||||
package prism
|
||||
|
||||
import (
|
||||
"context"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
// ----------------------------------------------------------------------------
|
||||
// Observables
|
||||
// ----------------------------------------------------------------------------
|
||||
|
||||
type PublishDispatched struct {
|
||||
EventID string
|
||||
At time.Time
|
||||
}
|
||||
|
||||
type PublishAccepted struct {
|
||||
EventID string
|
||||
At time.Time
|
||||
}
|
||||
|
||||
type PublishRejected struct {
|
||||
EventID string
|
||||
Message string
|
||||
At time.Time
|
||||
}
|
||||
|
||||
type PublishTimeout struct {
|
||||
EventID string
|
||||
At time.Time
|
||||
}
|
||||
|
||||
type PublishSendFailed struct {
|
||||
EventID string
|
||||
Err error
|
||||
At time.Time
|
||||
}
|
||||
|
||||
// ----------------------------------------------------------------------------
|
||||
// Types
|
||||
// ----------------------------------------------------------------------------
|
||||
|
||||
type publishResult struct {
|
||||
accepted bool
|
||||
message string
|
||||
err error
|
||||
}
|
||||
|
||||
type pendingEntry struct {
|
||||
eventID string
|
||||
eventJSON []byte
|
||||
sent bool
|
||||
result chan publishResult
|
||||
once sync.Once
|
||||
timer *time.Timer
|
||||
}
|
||||
|
||||
// ----------------------------------------------------------------------------
|
||||
// Event Publisher
|
||||
// ----------------------------------------------------------------------------
|
||||
|
||||
type EventPublisher struct {
|
||||
envoy *Envoy
|
||||
pending map[string]*pendingEntry // event id -> pending entry
|
||||
|
||||
inbox <-chan InboxMessage
|
||||
events <-chan OutboundPoolEvent
|
||||
|
||||
ctx context.Context
|
||||
cancel context.CancelFunc
|
||||
mu sync.Mutex
|
||||
wg sync.WaitGroup
|
||||
}
|
||||
|
||||
func NewEventPublisher(e *Envoy) *EventPublisher {
|
||||
// SubscribeInbox([]string{"OK"})
|
||||
// SubscribeEvents()
|
||||
// context.WithCancel from e.Context()
|
||||
// make pending map
|
||||
// wg.Add(2); go p.routeInbox(); go p.handleEvents()
|
||||
return nil
|
||||
}
|
||||
|
||||
func (p *EventPublisher) Publish(eventID string, eventJSON []byte, timeout time.Duration) (bool, string, error) {
|
||||
// build pendingEntry with result chan (buffered 1) and sync.Once
|
||||
// compute deadline; create timer that fires deliver(timeout error) via once
|
||||
// timer should lock -> deregister -> record PublishTimeout -> unlock -> deliver
|
||||
// mu.Lock: insert into pending map
|
||||
// if envoy connected: EncloseEvent + Send
|
||||
// on send error: deregister, deliver send error, return
|
||||
// on success: mark sent, record PublishDispatched
|
||||
// mu.Unlock
|
||||
// block on entry.result
|
||||
// return received publishResult fields
|
||||
return false, "", nil
|
||||
}
|
||||
|
||||
func (p *EventPublisher) Close() {
|
||||
// p.cancel()
|
||||
// mu.Lock: iterate pending, call once.Do(deliver cancellation error) for each
|
||||
// mu.Unlock
|
||||
// p.wg.Wait()
|
||||
}
|
||||
|
||||
func (p *EventPublisher) deliver(entry *pendingEntry, result publishResult) {
|
||||
entry.once.Do(func() {
|
||||
entry.timer.Stop()
|
||||
entry.result <- result
|
||||
})
|
||||
}
|
||||
|
||||
func (p *EventPublisher) routeInbox() {
|
||||
defer p.wg.Done()
|
||||
// select ctx.Done | inbox msg
|
||||
// msg: envelope.FindOK → on err continue
|
||||
// mu.Lock; look up pending by eventID
|
||||
// if not found: mu.Unlock; continue
|
||||
// deregister; mu.Unlock
|
||||
// record PublishAccepted or PublishRejected
|
||||
// deliver(entry, publishResult{accepted, message, nil})
|
||||
}
|
||||
|
||||
func (p *EventPublisher) handleEvents() {
|
||||
defer p.wg.Done()
|
||||
// select ctx.Done | event
|
||||
// EventConnected:
|
||||
// mu.Lock; iterate pending where !sent
|
||||
// Send each; on error: deregister, deliver send error, record PublishSendFailed
|
||||
// on success: mark sent, record PublishDispatched
|
||||
// mu.Unlock
|
||||
// EventDisconnected:
|
||||
// mu.Lock; mark all sent entries as unsent
|
||||
// mu.Unlock
|
||||
}
|
||||
|
||||
func (p *EventPublisher) deregister(eventID string) {
|
||||
delete(p.pending, eventID)
|
||||
}
|
||||
@@ -0,0 +1,76 @@
|
||||
package prism
|
||||
|
||||
import (
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestEventPublisher(t *testing.T) {
|
||||
t.Run("publish accepted", func(t *testing.T) {
|
||||
// p, envoy := newMockEnvoy(t); p.connect()
|
||||
// pub := NewEventPublisher(envoy); t.Cleanup(pub.Close)
|
||||
// go: accepted, msg, err := pub.Publish(id, json, TestTimeout)
|
||||
// Eventually: EVENT envelope on p.sent
|
||||
// p.receive(envelope.EncloseOK(id, true, ""))
|
||||
// assert accepted==true, msg=="", err==nil
|
||||
})
|
||||
|
||||
t.Run("publish rejected", func(t *testing.T) {
|
||||
// same setup; feed EncloseOK(id, false, "blocked: not on allowlist")
|
||||
// assert accepted==false, msg=="blocked: not on allowlist", err==nil
|
||||
})
|
||||
|
||||
t.Run("publish timeout", func(t *testing.T) {
|
||||
// pub.Publish(id, json, 75*time.Millisecond) — do not feed OK
|
||||
// assert err != nil (timeout error) after ~75ms
|
||||
})
|
||||
|
||||
t.Run("concurrent correlation", func(t *testing.T) {
|
||||
// Publish A and B in separate goroutines
|
||||
// feed OK for B, then OK for A
|
||||
// assert each caller receives its own result
|
||||
})
|
||||
|
||||
t.Run("publish while disconnected times out", func(t *testing.T) {
|
||||
// do not connect; Publish with short timeout
|
||||
// assert timeout error; Never: EVENT on p.sent
|
||||
})
|
||||
|
||||
t.Run("held event sent on connect", func(t *testing.T) {
|
||||
// do not connect; go pub.Publish(id, json, TestTimeout)
|
||||
// Never: EVENT on p.sent (yet)
|
||||
// p.connect()
|
||||
// Eventually: EVENT on p.sent
|
||||
// p.receive(envelope.EncloseOK(id, true, ""))
|
||||
// assert caller returns (true, "", nil)
|
||||
})
|
||||
|
||||
t.Run("resend after disconnect", func(t *testing.T) {
|
||||
// p.connect(); go pub.Publish
|
||||
// Eventually: first EVENT sent
|
||||
// p.disconnect(); p.connect()
|
||||
// Eventually: second EVENT sent
|
||||
// p.receive(EncloseOK); assert caller returns successfully
|
||||
})
|
||||
|
||||
t.Run("send failure delivers error", func(t *testing.T) {
|
||||
// Publish; assert caller receives send error immediately (not timeout)
|
||||
})
|
||||
|
||||
t.Run("disconnect then send failure on reconnect", func(t *testing.T) {
|
||||
// p.connect(); go pub.Publish (EVENT sent)
|
||||
// p.disconnect()
|
||||
// p.connect()
|
||||
// assert caller receives send error
|
||||
})
|
||||
|
||||
t.Run("close cancels pending", func(t *testing.T) {
|
||||
// go pub.Publish (blocks)
|
||||
// pub.Close() before feeding OK
|
||||
// assert caller receives cancellation error
|
||||
})
|
||||
|
||||
t.Run("unknown OK ignored", func(t *testing.T) {
|
||||
// p.receive(envelope.EncloseOK("no-such-id", true, ""))
|
||||
// assert no panic, no side effects
|
||||
})
|
||||
}
|
||||
Reference in New Issue
Block a user