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() }) }