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..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 }