127 lines
4.9 KiB
Go
127 lines
4.9 KiB
Go
package buscompose
|
|
|
|
import (
|
|
"slices"
|
|
"sort"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/nats-io/nats.go/jetstream"
|
|
|
|
"git.novox.be/novox/mesh-sdk/go/seatdef"
|
|
)
|
|
|
|
// **A seat's event kept in a stream of its own** (novox/hq design 47, *What each step says*; the seat definition's `keep`):
|
|
// the node-engine's `running.<node>.<module>` last per subject with no age limit, the gate's and the checker's `owed` for a
|
|
// week within 64 MiB. The event's subject lies under `mesh.seat.*.event.>`, which the shared events stream already takes, and
|
|
// the bus refuses two streams taking one subject: so the kept stream takes no subject of its own but sources the shared
|
|
// stream's messages of that subject, and keeps them by its own limits (the shared stream's seven days never end a
|
|
// `running`). The part makes the stream, and each hearing module's durable on it from the module's `consumes` (the bus name
|
|
// is seatdef.KeptStream's, the one name for every part that hears it).
|
|
|
|
// KeptEvent is a seat's event with a stream of its own.
|
|
type KeptEvent struct {
|
|
Seat string
|
|
Event string
|
|
// Stream is its name on the bus (seatdef.KeptStream).
|
|
Stream string
|
|
// Filter is the subject that carries every instance of the event, a part of its subject `*`.
|
|
Filter string
|
|
// Last says the stream keeps the last message of each subject; MaxAgeDays (0 for no age limit) and MaxMiB bound it
|
|
// as the event's keep says, its defaults 90 days and 16 MiB (design 47 G §396, §331).
|
|
Last bool
|
|
MaxAgeDays int
|
|
MaxMiB int
|
|
Why string
|
|
}
|
|
|
|
// KeptEventsOf is every event the seat definitions keep in a stream of its own, sorted by stream name. Pure.
|
|
func KeptEventsOf(defs map[string]seatdef.Definition) []KeptEvent {
|
|
var out []KeptEvent
|
|
for _, d := range defs {
|
|
for _, e := range d.Events {
|
|
if e.Keep == nil {
|
|
continue
|
|
}
|
|
out = append(out, KeptEvent{Seat: d.Seat, Event: e.Name, Stream: seatdef.KeptStream(d.Seat, e),
|
|
Filter: seatdef.EventSubject(d.Seat, e.Local()), Last: e.KeptLastPerSubject(), MaxAgeDays: e.KeptAgeDays(), MaxMiB: e.KeptMiB(),
|
|
Why: "the seat " + d.Seat + "'s event " + e.Name + ", kept apart from the shared events stream (design 47): " + keptHow(e)})
|
|
}
|
|
}
|
|
sort.Slice(out, func(i, j int) bool { return out[i].Stream < out[j].Stream })
|
|
return out
|
|
}
|
|
|
|
func keptHow(e seatdef.Message) string {
|
|
if e.KeptLastPerSubject() {
|
|
return "the last message of each subject, no age limit"
|
|
}
|
|
return "every message within its age and size"
|
|
}
|
|
|
|
// LastPerSubject says a hearing consumer starts from the last message of each subject, never from now.
|
|
func (k KeptEvent) LastPerSubject() bool { return k.Last }
|
|
|
|
// StreamConfig is the kept stream as it is created: sourced from the shared events stream by the event's subject, with
|
|
// its own limits as the event's keep says (design 47 G §396): its age (none for 0), its cap, and one message a subject
|
|
// when it keeps the last of each. A stream at its cap drops its oldest message (§331).
|
|
func (k KeptEvent) StreamConfig() jetstream.StreamConfig {
|
|
cfg := jetstream.StreamConfig{
|
|
Name: k.Stream,
|
|
Description: k.Why,
|
|
Storage: jetstream.FileStorage,
|
|
Replicas: 1,
|
|
Sources: []*jetstream.StreamSource{{Name: EventsStream, FilterSubject: k.Filter}},
|
|
MaxAge: time.Duration(k.MaxAgeDays) * 24 * time.Hour,
|
|
MaxBytes: int64(k.MaxMiB) << 20,
|
|
}
|
|
if k.Last {
|
|
cfg.MaxMsgsPerSubject = 1
|
|
}
|
|
return cfg
|
|
}
|
|
|
|
// KeptHeard is the part of a module's `consumes` (`<seat>.<event>`) that names a kept event, as the Heard it makes: the
|
|
// module's durable on the kept stream, filtered to the subject it consumes, from the last message of each subject for an
|
|
// event so kept. ok is false for a consume that names none.
|
|
func KeptHeard(kept []KeptEvent, consume string) (h Heard, ok bool) {
|
|
seat, rest, named := strings.Cut(consume, ".")
|
|
if !named {
|
|
return Heard{}, false
|
|
}
|
|
for _, k := range kept {
|
|
local := strings.TrimPrefix(k.Filter, seatdef.EventSubject(k.Seat, ""))
|
|
if k.Seat != seat || (rest != k.Event && rest != local) {
|
|
continue
|
|
}
|
|
return Heard{Stream: k.Stream, Emitter: k.Seat, Filters: []string{k.Filter}, LastPerSubject: k.LastPerSubject()}, true
|
|
}
|
|
return Heard{}, false
|
|
}
|
|
|
|
// SplitKept divides a module's consumes into the ones the shared events stream carries and the kept events' hearing, merged
|
|
// by stream, filters sorted without repeats.
|
|
func SplitKept(kept []KeptEvent, consumes []string) (rest []string, hears []Heard) {
|
|
byStream := map[string]*Heard{}
|
|
var order []string
|
|
for _, c := range consumes {
|
|
h, ok := KeptHeard(kept, c)
|
|
if !ok {
|
|
rest = append(rest, c)
|
|
continue
|
|
}
|
|
if byStream[h.Stream] == nil {
|
|
byStream[h.Stream] = &Heard{Stream: h.Stream, Emitter: h.Emitter, LastPerSubject: h.LastPerSubject}
|
|
order = append(order, h.Stream)
|
|
}
|
|
byStream[h.Stream].Filters = append(byStream[h.Stream].Filters, h.Filters...)
|
|
}
|
|
for _, s := range order {
|
|
h := byStream[s]
|
|
slices.Sort(h.Filters)
|
|
h.Filters = slices.Compact(h.Filters)
|
|
hears = append(hears, *h)
|
|
}
|
|
return rest, hears
|
|
}
|