110 lines
4.5 KiB
Go
110 lines
4.5 KiB
Go
package buscompose
|
|
|
|
import (
|
|
"fmt"
|
|
"sort"
|
|
)
|
|
|
|
// moduleStreamPrefix starts every subject a module's own stream takes.
|
|
const moduleStreamPrefix = "mesh.stream."
|
|
|
|
// ModuleStreamSubject is where a module's event on its own stream lands: `mesh.stream.<module>.event.<event>`.
|
|
// The token before `.event.` is the emitter, as on EVENTS, so a runtime reads the same key from it.
|
|
func ModuleStreamSubject(module, event string) string {
|
|
return moduleStreamPrefix + module + ".event." + event
|
|
}
|
|
|
|
// A ModuleStream is one module's declared stream as the bus holds it.
|
|
type ModuleStream struct {
|
|
Module string
|
|
// Name is its name on the bus, `MAIL`.
|
|
Name string
|
|
// Events are the events it carries, as the bus spells them (WirePattern).
|
|
Events []string
|
|
MaxAgeDays int
|
|
MaxMiB int
|
|
// Restricted says a trusted config value of the module grants who hears it (ADR 0363 §4).
|
|
Restricted bool
|
|
// WorkQueue says this stream holds ADR 0436 §6's work-queue retention: a message is removed once a
|
|
// consumer acknowledges it, and the bus refuses a new one when full rather than dropping its oldest.
|
|
WorkQueue bool
|
|
// Duplicates is the work-queue's duplicate window in hours (novox/hq ADR 0436 §6); meaningless unless
|
|
// WorkQueue.
|
|
Duplicates int
|
|
}
|
|
|
|
// Subjects are the subjects the stream takes.
|
|
func (s ModuleStream) Subjects() []string {
|
|
var out []string
|
|
for _, e := range s.Events {
|
|
out = append(out, ModuleStreamSubject(s.Module, e))
|
|
}
|
|
sort.Strings(out)
|
|
return out
|
|
}
|
|
|
|
// Stream is the stream the controller asserts: kept by limits, its age and size the module's, dropping
|
|
// its oldest when full — a consumer that was away catches up within the age, and the module's own record
|
|
// answers for anything older.
|
|
func (s ModuleStream) Stream() Stream {
|
|
why := fmt.Sprintf("%s's own stream (novox/hq ADR 0363): its events, kept %d days and at most %d MiB so a "+
|
|
"consumer that was away catches up", s.Module, s.MaxAgeDays, s.MaxMiB)
|
|
if s.Restricted {
|
|
why += "; heard only by the consumers its trusted config value grants"
|
|
}
|
|
if s.WorkQueue {
|
|
why += "; a work-queue (novox/hq ADR 0436 §6): a message is removed once its one consumer " +
|
|
"acknowledges it, and a full stream refuses a new one rather than dropping what it holds"
|
|
}
|
|
out := Stream{
|
|
Name: s.Name,
|
|
Subjects: s.Subjects(),
|
|
Retention: RetentionLimits,
|
|
MaxAge: s.MaxAgeDays * 24 * 60 * 60,
|
|
MaxBytes: int64(s.MaxMiB) * 1024 * 1024,
|
|
Why: why,
|
|
// Said on the bus, so a dead letter kept from it ages out with it (novox/hq ADR 0376 §3).
|
|
Restricted: s.Restricted,
|
|
}
|
|
if s.WorkQueue {
|
|
out.Retention = RetentionWorkQueue
|
|
out.DiscardNew = true
|
|
out.DuplicatesSeconds = s.Duplicates * 60 * 60
|
|
// Said on the bus too, reusing the one flag a restricted stream's dead letters already age and hide
|
|
// by (novox/hq ADR 0436 §6, ADR 0376 §3-4): a work-queue stream's grant is always empty (ADR 0436
|
|
// §6 refuses one that carries a grant), so this never collides with a trusted-config restriction.
|
|
out.Restricted = true
|
|
// And its own metadata, apart from the bus's work-queue retention SEAT_* and other mesh-internal
|
|
// streams share without being one of this clause's: only this flag says a dead letter kept from the
|
|
// stream is drop-only (novox/hq ADR 0436 §6), never the stream's Retention alone.
|
|
out.WorkQueue = true
|
|
}
|
|
return out
|
|
}
|
|
|
|
// QueueOwner is the one process of a module that is the durable consumer of one of its own work-queue
|
|
// streams (novox/hq ADR 0436 §1, §2, §6): named in that process's bus.queues, and checked at registration
|
|
// to be the only one (the indexer's judge, novox/hq ADR 0578 §1).
|
|
type QueueOwner struct {
|
|
// Process is the process resource's id — its bus account's name, with the node and module (novox/hq ADR
|
|
// 0436 §2).
|
|
Process string
|
|
// Stream is the work-queue stream's name on the bus.
|
|
Stream string
|
|
// MaxDeliver bounds how many times a message is handed over before it is given up as a dead letter
|
|
// (novox/hq ADR 0436 §6).
|
|
MaxDeliver int
|
|
}
|
|
|
|
// Heard is a module's stream another module hears, and the subjects it hears of it: its `consumes`
|
|
// within the stream and, for a restricted stream, within what the provider's trusted config value grants it
|
|
// (ADR 0363 §5). Composed only with at least one subject.
|
|
type Heard struct {
|
|
Stream string
|
|
Emitter string
|
|
Filters []string
|
|
// Restricted says the stream is restricted by its provider's trusted config value (ADR 0363 §4): then it
|
|
// is heard only by a module on an account of its own (ADR 0376 §1, ComposedHears).
|
|
Restricted bool
|
|
}
|