145 lines
4.2 KiB
Go
145 lines
4.2 KiB
Go
package process
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"strings"
|
|
"sync"
|
|
|
|
"github.com/nats-io/nats.go/jetstream"
|
|
)
|
|
|
|
// EventHub is the one pull of a module's durable consumer on the shared events stream (novox/hq ADR 0635 §5): a module
|
|
// has one durable, `<node>_<module>`, so two loops pulling it each take messages the other needs. Every part of the
|
|
// module that hears events registers a handler by subject pattern, and the hub pulls once and hands each message to the
|
|
// handler whose pattern it matches. A message no handler takes is acknowledged and said, never left to be redelivered.
|
|
// The hub never makes the stream or the consumer: they are the controller's, and a bus without them is an error of Start.
|
|
//
|
|
// Register every handler, then Start once: a message heard before its handler is registered would be taken by none.
|
|
type EventHub struct {
|
|
logf func(string, ...any)
|
|
|
|
mu sync.RWMutex
|
|
routes []hubRoute
|
|
cons jetstream.ConsumeContext
|
|
}
|
|
|
|
type hubRoute struct {
|
|
pattern string
|
|
handle func(jetstream.Msg)
|
|
}
|
|
|
|
// NewEventHub is a hub saying through logf (nil says nothing).
|
|
func NewEventHub(logf func(string, ...any)) *EventHub {
|
|
if logf == nil {
|
|
logf = func(string, ...any) {}
|
|
}
|
|
return &EventHub{logf: logf}
|
|
}
|
|
|
|
// Handle registers handle for the subjects matching pattern (`*` one token, `>` the rest). The handler acknowledges
|
|
// the message itself. A pattern that overlaps one registered (one matches every subject of the other) is refused,
|
|
// so a message always has one taker.
|
|
func (h *EventHub) Handle(pattern string, handle func(jetstream.Msg)) error {
|
|
if pattern == "" || handle == nil {
|
|
return fmt.Errorf("an event handler needs a subject pattern and a handler")
|
|
}
|
|
h.mu.Lock()
|
|
defer h.mu.Unlock()
|
|
for _, r := range h.routes {
|
|
if subjectsOverlap(r.pattern, pattern) {
|
|
return fmt.Errorf("the events %q overlap %q, already handled: a message has one taker", pattern, r.pattern)
|
|
}
|
|
}
|
|
h.routes = append(h.routes, hubRoute{pattern, handle})
|
|
return nil
|
|
}
|
|
|
|
// Patterns are the registered subject patterns, in order.
|
|
func (h *EventHub) Patterns() []string {
|
|
h.mu.RLock()
|
|
defer h.mu.RUnlock()
|
|
out := make([]string, len(h.routes))
|
|
for i, r := range h.routes {
|
|
out[i] = r.pattern
|
|
}
|
|
return out
|
|
}
|
|
|
|
// Start binds the module's durable on the shared events stream and pulls it; a call after a start that took does
|
|
// nothing. It makes no stream and no consumer: a consumer the bus does not hold yet is the error of this call, and a
|
|
// later call (the module's next change of membership) binds it, so a part is not lost for the bus's order.
|
|
func (h *EventHub) Start(ctx context.Context, js jetstream.JetStream, cred Account) error {
|
|
h.mu.Lock()
|
|
defer h.mu.Unlock()
|
|
if h.cons != nil {
|
|
return nil
|
|
}
|
|
cons, err := cred.EventConsumer(ctx, js, EventsStream)
|
|
if err == nil {
|
|
h.cons, err = HearEvents(cons, h.hear, h.logf)
|
|
}
|
|
if err != nil {
|
|
return fmt.Errorf("the consumer %s on %s is not heard: %w", cred.EventConsumerName(), EventsStream, err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// Stop ends the pull.
|
|
func (h *EventHub) Stop() {
|
|
h.mu.Lock()
|
|
defer h.mu.Unlock()
|
|
if h.cons != nil {
|
|
h.cons.Stop()
|
|
h.cons = nil
|
|
}
|
|
}
|
|
|
|
func (h *EventHub) hear(m jetstream.Msg) {
|
|
h.mu.RLock()
|
|
var take func(jetstream.Msg)
|
|
for _, r := range h.routes {
|
|
if subjectMatches(r.pattern, m.Subject()) {
|
|
take = r.handle
|
|
break
|
|
}
|
|
}
|
|
h.mu.RUnlock()
|
|
if take == nil {
|
|
h.logf("events: %s is taken by no handler of this module, acknowledged", m.Subject())
|
|
_ = m.Ack()
|
|
return
|
|
}
|
|
take(m)
|
|
}
|
|
|
|
// subjectMatches says whether subject is one the pattern (`*` one token, `>` the rest) names.
|
|
func subjectMatches(pattern, subject string) bool {
|
|
pt, st := strings.Split(pattern, "."), strings.Split(subject, ".")
|
|
for i, t := range pt {
|
|
if t == ">" {
|
|
return len(st) > i
|
|
}
|
|
if i >= len(st) || (t != "*" && t != st[i]) {
|
|
return false
|
|
}
|
|
}
|
|
return len(pt) == len(st)
|
|
}
|
|
|
|
// subjectsOverlap says whether some subject matches both patterns.
|
|
func subjectsOverlap(a, b string) bool {
|
|
at, bt := strings.Split(a, "."), strings.Split(b, ".")
|
|
for i := 0; ; i++ {
|
|
if i >= len(at) || i >= len(bt) {
|
|
return len(at) == len(bt)
|
|
}
|
|
if at[i] == ">" || bt[i] == ">" {
|
|
return true
|
|
}
|
|
if at[i] != "*" && bt[i] != "*" && at[i] != bt[i] {
|
|
return false
|
|
}
|
|
}
|
|
}
|