127 lines
4.9 KiB
Go
127 lines
4.9 KiB
Go
package bus
|
|
|
|
import (
|
|
"encoding/json"
|
|
"fmt"
|
|
"sort"
|
|
"strconv"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/nats-io/nats.go"
|
|
)
|
|
|
|
// A module reading its own stream back (novox/hq ADR 0363; the operator, 2026-10-30: a build step's whole failure output
|
|
// is said on its build module's stream, and the module's tools read it there). As a log's read: one batch direct get,
|
|
// no consumer, and only a stream the module's own membership names (`streams`); the bus grants its owner that stream's
|
|
// info and direct get and no other's (mesh-sdk buscompose ownStreamGrants).
|
|
|
|
// StreamEntry is one event of a module's own stream: its sequence, its event (the subject after the stream's
|
|
// `…event.`), its body and when the bus took it.
|
|
type StreamEntry struct {
|
|
Sequence uint64 `json:"sequence"`
|
|
Event string `json:"event"`
|
|
Body json.RawMessage `json:"body"`
|
|
At time.Time `json:"at"`
|
|
}
|
|
|
|
// StreamPage is one page of a stream's read: Next is the sequence to continue from, 0 when nothing follows; Last the
|
|
// stream's last sequence.
|
|
type StreamPage struct {
|
|
Entries []StreamEntry `json:"entries"`
|
|
Next uint64 `json:"next"`
|
|
Last uint64 `json:"last"`
|
|
}
|
|
|
|
// streamBusName is a stream's local name as the bus holds it (mesh-sdk catalogue.StreamBusName): capitals, a hyphen an
|
|
// underscore; a bus name is itself.
|
|
func streamBusName(name string) string { return strings.ToUpper(strings.ReplaceAll(name, "-", "_")) }
|
|
|
|
// issuedStream is the stream of its own a module may read by a name, local or the bus's, or the reason it may not.
|
|
func (c *Conn) issuedStream(module, name string) (StreamIssued, error) {
|
|
m := c.Membership(module)
|
|
if m == nil {
|
|
return StreamIssued{}, fmt.Errorf("%s has no membership issued on %s yet, so no stream of it is readable "+
|
|
"until the mesh issues one (novox/hq ADR 0363)", module, c.node)
|
|
}
|
|
var names []string
|
|
for _, s := range m.Streams {
|
|
if s.Stream == streamBusName(name) && strings.Contains(s.Subject, "{event}") {
|
|
return s, nil
|
|
}
|
|
names = append(names, s.Stream)
|
|
}
|
|
sort.Strings(names)
|
|
issued := "none"
|
|
if len(names) > 0 {
|
|
issued = strings.Join(names, ", ")
|
|
}
|
|
return StreamIssued{}, fmt.Errorf("%s keeps no stream called %q: a module reads the streams it declares under `streams`, "+
|
|
"and no other's; it was issued: %s (novox/hq ADR 0363)", module, name, issued)
|
|
}
|
|
|
|
// StreamRead is a module's own stream's events from a sequence on, whose event names match a pattern (`*` one name, `>`
|
|
// the rest, last; empty is every event), at most limit (LogReadDefault when 0, at most LogReadMax) and about LogPageBytes.
|
|
func (c *Conn) StreamRead(module, name, events string, from uint64, limit int) (*StreamPage, error) {
|
|
s, err := c.issuedStream(module, name)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
switch {
|
|
case limit == 0:
|
|
limit = LogReadDefault
|
|
case limit < 0 || limit > LogReadMax:
|
|
return nil, fmt.Errorf("a read's limit is 1 to %d events, %d when none is given, not %d", LogReadMax, LogReadDefault, limit)
|
|
}
|
|
if events == "" {
|
|
events = ">"
|
|
}
|
|
parts := strings.Split(events, ".")
|
|
for i, p := range parts {
|
|
if !(p == ">" && i == len(parts)-1) && p != "*" && !logToken.MatchString(p) {
|
|
return nil, fmt.Errorf("%q is not an event pattern: names, `*` for one and `>` for the rest, last", events)
|
|
}
|
|
}
|
|
prefix, _, _ := strings.Cut(s.Subject, "{event}")
|
|
missing := func() error {
|
|
return fmt.Errorf("the stream %s is issued and not on the bus yet: the controller creates it from the catalogue "+
|
|
"on its next raise, and reads it directly once it allows that (novox/hq ADR 0363)", s.Stream)
|
|
}
|
|
msgs, next, err := c.directBatch(s.Stream, prefix+events, from, limit, s.Stream, missing)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
page := &StreamPage{Entries: []StreamEntry{}, Next: next}
|
|
for _, msg := range msgs {
|
|
e, err := streamEntryOf(s.Stream, prefix, msg)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
page.Entries = append(page.Entries, *e)
|
|
}
|
|
info, err := c.logInfo(LogIssued{Name: s.Stream, Stream: s.Stream})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
page.Last = info.Last
|
|
return page, nil
|
|
}
|
|
|
|
// streamEntryOf is an event a direct get answered: its event is its subject after prefix.
|
|
func streamEntryOf(stream, prefix string, msg *nats.Msg) (*StreamEntry, error) {
|
|
subject := msg.Header.Get("Nats-Subject")
|
|
event, ok := strings.CutPrefix(subject, prefix)
|
|
if !ok {
|
|
return nil, fmt.Errorf("the bus answered an event of %s on %q, which is not under %s", stream, subject, prefix)
|
|
}
|
|
seq, err := strconv.ParseUint(msg.Header.Get("Nats-Sequence"), 10, 64)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("the bus answered an event of %s with no sequence", stream)
|
|
}
|
|
at, err := time.Parse(time.RFC3339Nano, msg.Header.Get("Nats-Time-Stamp"))
|
|
if err != nil {
|
|
return nil, fmt.Errorf("the bus answered event %d of %s with no time: %v", seq, stream, err)
|
|
}
|
|
return &StreamEntry{Sequence: seq, Event: event, Body: valueOf(msg.Data), At: at.UTC()}, nil
|
|
}
|