diff --git a/go/seatlist/processbus.go b/go/seatlist/processbus.go new file mode 100644 index 0000000..b9a3e88 --- /dev/null +++ b/go/seatlist/processbus.go @@ -0,0 +1,44 @@ +package seatlist + +import ( + "encoding/json" + "fmt" + "strings" + + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-sdk/go/seatdef" +) + +// EventHandlers is where a process registers the handler of a subject on the module's one pull of its durable consumer +// (process.EventHub). Register every handler, then start the hub. +type EventHandlers interface { + Handle(pattern string, handle func(jetstream.Msg)) error +} + +type processBus struct { + call func(string, any) (json.RawMessage, error) + hub EventHandlers +} + +// ProcessBus is the bus of a process on its own credential: a verb of the seat is called through its connection +// (process.Conn.Call), and the seat's events are heard as handlers of the module's event hub, on the seat's own subject +// (`mesh.seat..event.`). The process starts the hub after the reader is started, so every handler is registered. +func ProcessBus(call func(string, any) (json.RawMessage, error), hub EventHandlers) Bus { + return processBus{call: call, hub: hub} +} + +func (b processBus) Ask(key string, body any) (json.RawMessage, error) { return b.call(key, body) } + +func (b processBus) Subscribe(pattern string, handle func(key string, body json.RawMessage) error) error { + seat, event, ok := strings.Cut(pattern, ".") + if !ok || seat == "" || event == "" { + return fmt.Errorf("%q is no seat event's pattern, .", pattern) + } + return b.hub.Handle(seatdef.EventSubject(seat, event), func(m jetstream.Msg) { + // A reader that could not take the event says so in its status and reads again at the next event or by hand + // (Reload): the message is not delivered again for it. + _ = handle(pattern, m.Data()) + _ = m.Ack() + }) +} diff --git a/go/seatlist/processbus_test.go b/go/seatlist/processbus_test.go new file mode 100644 index 0000000..478bfa9 --- /dev/null +++ b/go/seatlist/processbus_test.go @@ -0,0 +1,46 @@ +package seatlist + +import ( + "encoding/json" + "testing" + + "github.com/nats-io/nats.go/jetstream" +) + +type hubStub map[string]func(jetstream.Msg) + +func (h hubStub) Handle(pattern string, f func(jetstream.Msg)) error { h[pattern] = f; return nil } + +type msgStub struct { + jetstream.Msg + data []byte + acked *int +} + +func (m msgStub) Data() []byte { return m.data } +func (m msgStub) Ack() error { *m.acked++; return nil } + +func TestAProcessBusAsksByKeyAndHearsTheSeatsEventsOnTheHub(t *testing.T) { + var asked string + hub := hubStub{} + bus := ProcessBus(func(key string, body any) (json.RawMessage, error) { asked = key; return json.RawMessage(`{}`), nil }, hub) + if _, err := bus.Ask(Key(), nil); err != nil || asked != Key() { + t.Fatalf("asked %q, %v", asked, err) + } + var got []string + if err := bus.Subscribe(Pattern(ChangedEvent), func(key string, body json.RawMessage) error { got = append(got, key+" "+string(body)); return nil }); err != nil { + t.Fatal(err) + } + h := hub["mesh.seat."+Seat+".event."+ChangedEvent] + if h == nil { + t.Fatalf("handlers on %v, want the seat's own subject", hub) + } + acked := 0 + h(msgStub{data: []byte(`{"seat":"x"}`), acked: &acked}) + if len(got) != 1 || got[0] != Pattern(ChangedEvent)+` {"seat":"x"}` || acked != 1 { + t.Errorf("heard %v, acked %d", got, acked) + } + if err := bus.Subscribe("nodot", nil); err == nil { + t.Error("a pattern that names no seat event was taken") + } +}