45 lines
1.7 KiB
Go
45 lines
1.7 KiB
Go
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.<seat>.event.<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, <seat>.<event>", 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()
|
|
})
|
|
}
|