diff --git a/node-tools/internal/bus/bus.go b/node-tools/internal/bus/bus.go index c1dd0e6..ed41eb7 100644 --- a/node-tools/internal/bus/bus.go +++ b/node-tools/internal/bus/bus.go @@ -10,7 +10,6 @@ package bus import ( - "context" "crypto/sha256" "crypto/tls" "crypto/x509" @@ -71,6 +70,12 @@ type Membership struct { // SeatTraffic is what the module's code may submit, say, take, ask, answer and read on seats that name // their caller or their kind (novox/hq ADR 0259): the runtime does for it only what is listed here. SeatTraffic *SeatTraffic `json:"seat-traffic,omitempty"` + // Streams are the module's own streams (novox/hq ADR 0363 §2): an event one of them carries is published + // on that stream's subject, never on EVENTS (streams.go). + Streams []StreamIssued `json:"streams,omitempty"` + // Hears are the module streams the module hears, each through a durable consumer on that stream the + // runtime binds beside the one on EVENTS (ADR 0363 §8). + Hears []HeardIssued `json:"hears,omitempty"` } // Served is an address a tool is answered on; `{tool}` stands for the tool's name. @@ -715,7 +720,9 @@ func (c *Conn) answerOf(subject string, data []byte) (Answered, bool, error) { // PublishAs emits an event as a module: published into JetStream and awaited, de-duplicated by its // own id (ADR 0042). The body is the payload; the metadata rides as headers. func (c *Conn) PublishAs(module string, env Envelope) error { - msg := nats.NewMsg("mesh.mod." + module + ".event." + env.Key) + // On the module's own stream when its membership names one carrying the event (ADR 0363 §2), on EVENTS + // otherwise — and only there, as before, for a membership that names no stream. + msg := nats.NewMsg(eventSubject(c.Membership(module), module, env.Key)) for k, v := range env.Headers { msg.Header.Set(k, v) } @@ -937,64 +944,34 @@ var NakDelay = 5 * time.Second // handed over, and a burst would end in max-deliver with nothing failed (novox/hq issue 387). var EventHeartbeat = 10 * time.Second -// ConsumeAs reads a module's durable consumer and hands each event to deliver (novox/hq ADR 0198): -// acknowledged when deliver returns nil, negatively acknowledged after NakDelay when it returns an -// error, terminated when it is not an event at all. The consumer is the controller's to create; this -// binds to it and never makes one. Runs until stopped. +// ConsumeAs reads a module's durable consumer on EVENTS, and each one on a module's stream its +// membership names (novox/hq ADR 0363 §8), and hands each event to deliver one at a time (novox/hq +// ADR 0198): acknowledged when deliver returns nil, negatively acknowledged after NakDelay when it +// returns an error, terminated when it is not an event at all. The consumers are the controller's to +// create; this binds to them and never makes one. Runs until stopped. func (c *Conn) ConsumeAs(module string, deliver func(Envelope) error) (func(), error) { durable := ConsumerOf(c.node, module) js, err := jetstream.New(c.nc) if err != nil { return nil, err } - ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) - defer cancel() + h := newHearing(c, js, module, deliver) // Bound by name, never created and never matched against a subject: the consumer and its // filters are the controller's (design 29 §3), exactly as the module's own runtime bound it. - consumer, err := js.Consumer(ctx, EventsStream, durable) + events, doing, err := h.bind(EventsStream, durable) if err != nil { - return nil, fmt.Errorf("binding %s's consumer %s: %w", module, durable, err) - } - reading, err := consumer.Consume(func(msg jetstream.Msg) { - env, ok := envelopeOf(msg.Subject(), msg.Headers(), msg.Data()) - if !ok { - // Unparseable: redelivering bytes no version of a handler can read is a loop. - _ = msg.Term() - return - } - // Kept alive while the bundle works on it, as a worker's work is (seat.go): the consumer's ack wait - // (30 seconds) is for a runtime that died, not a handler that is slow — the runtime gives a handler - // launch.EventTimeout. Without it an event handled longer than the ack wait was offered again while - // it was handled, and counted toward max-deliver (novox/hq issue 387). - done, stopped := make(chan struct{}), make(chan struct{}) - go func() { - defer close(stopped) - tick := time.NewTicker(EventHeartbeat) - defer tick.Stop() - for { - select { - case <-done: - return - case <-tick.C: - _ = msg.InProgress() - } - } - }() - err := deliver(env) - close(done) - // No in-progress after the answer: one after a Nak would stretch its delay to the whole ack wait. - <-stopped - if err != nil { - _ = msg.NakWithDelay(NakDelay) - return - } - _ = msg.Ack() - }, jetstream.PullMaxMessages(1)) - if err != nil { - return nil, fmt.Errorf("reading %s's consumer %s: %w", module, durable, err) + return nil, fmt.Errorf("%s %s's consumer %s: %w", doing, module, durable, err) } + // And every consumer on a module's stream its membership names (novox/hq ADR 0363 §8), followed as the + // membership changes. A membership naming none binds nothing more: the module hears EVENTS alone. + h.start() var once sync.Once - return func() { once.Do(reading.Stop) }, nil + return func() { + once.Do(func() { + h.stop() + events.Stop() + }) + }, nil } // envelopeOf rebuilds the envelope a module sees from a delivered event: its key is the emitter and diff --git a/node-tools/internal/bus/streams.go b/node-tools/internal/bus/streams.go new file mode 100644 index 0000000..831225e --- /dev/null +++ b/node-tools/internal/bus/streams.go @@ -0,0 +1,251 @@ +package bus + +import ( + "context" + "sort" + "strings" + "sync" + "time" + + "github.com/nats-io/nats.go/jetstream" +) + +// A module's own stream, as the runtime publishes to it and hears it (novox/hq ADR 0363 §8, issue 725). +// +// A module may declare a stream of its own; the controller creates it and says, in the module's +// membership, which of its events go there and on which subject (`streams`), and, in the membership of +// each module granted to hear it, which durable consumer on it to bind (`hears`). The runtime publishes and +// binds exactly as the membership says. A membership naming neither is what every membership was before: +// every event on EVENTS, and only the consumer on EVENTS bound. + +// StreamIssued is one of a module's own streams as its membership names it: the events it carries, as the +// bus spells them (`*` one token, `>` the rest), and the subject an event of them is published on, `{event}` +// standing for the event. +type StreamIssued struct { + Stream string `json:"stream"` + Events []string `json:"events"` + Subject string `json:"subject"` +} + +// HeardIssued is one module stream an assignment hears: the stream and the durable consumer on it the +// runtime binds. The consumer is the controller's to make, to filter and to delete. +type HeardIssued struct { + Stream string `json:"stream"` + Consumer string `json:"consumer"` +} + +// eventSubject is where a module's event is published: on the subject of the first of its membership's +// streams that carries it, else on EVENTS — `mesh.mod..event.`, as every event was before. +func eventSubject(m *Membership, module, event string) string { + if m != nil { + for _, s := range m.Streams { + if s.Subject == "" || !strings.Contains(s.Subject, "{event}") { + continue + } + for _, pattern := range s.Events { + if eventMatches(pattern, event) { + return strings.ReplaceAll(s.Subject, "{event}", event) + } + } + } + } + return "mesh.mod." + module + ".event." + event +} + +// eventMatches says whether an event's name falls under a pattern as the bus matches subjects: `*` is one +// token, `>` (or `**`, the manifest's spelling) the one or more tokens left. +func eventMatches(pattern, event string) bool { + if pattern == "" || event == "" { + return false + } + ps, es := strings.Split(pattern, "."), strings.Split(event, ".") + for i, p := range ps { + if p == ">" || p == "**" { + return i == len(ps)-1 && len(es) > i + } + if i >= len(es) || (p != "*" && p != es[i]) { + return false + } + } + return len(ps) == len(es) +} + +// RebindEvery is how often a consumer the membership names and the bus did not let the runtime bind is +// tried again: the controller makes the consumer and sends the user list carrying its grant in the +// composition that issues the membership, so one may arrive before the other (ADR 0363 §6). +var RebindEvery = 30 * time.Second + +// hearing is one module's consumers: the one on EVENTS, and each on a module stream its membership names. +// Every event of all of them is handed to the module one at a time, as the consumer on EVENTS alone did. +type hearing struct { + c *Conn + js jetstream.JetStream + module string + deliver func(Envelope) error + // serial hands the module one event at a time across its consumers (novox/hq issue 387). + serial sync.Mutex + + mu sync.Mutex + stopped bool + bound map[HeardIssued]jetstream.ConsumeContext + said map[HeardIssued]bool + kick chan struct{} + done chan struct{} +} + +func newHearing(c *Conn, js jetstream.JetStream, module string, deliver func(Envelope) error) *hearing { + return &hearing{c: c, js: js, module: module, deliver: deliver, + bound: map[HeardIssued]jetstream.ConsumeContext{}, said: map[HeardIssued]bool{}, + kick: make(chan struct{}, 1), done: make(chan struct{})} +} + +// bind reads one durable consumer by name, and on failure says what it was doing. +func (h *hearing) bind(stream, durable string) (jetstream.ConsumeContext, string, error) { + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + consumer, err := h.js.Consumer(ctx, stream, durable) + if err != nil { + return nil, "binding", err + } + reading, err := consumer.Consume(h.handle, jetstream.PullMaxMessages(1)) + if err != nil { + return nil, "reading", err + } + return reading, "", nil +} + +// handle hands one delivered event to the module: acknowledged when it took it, offered again after NakDelay +// when it did not, terminated when it is not an event at all. +func (h *hearing) handle(msg jetstream.Msg) { + env, ok := envelopeOf(msg.Subject(), msg.Headers(), msg.Data()) + if !ok { + // Unparseable: redelivering bytes no version of a handler can read is a loop. + _ = msg.Term() + return + } + // Kept alive while the bundle works on it, as a worker's work is (seat.go): the consumer's ack wait + // (30 seconds) is for a runtime that died, not a handler that is slow — the runtime gives a handler + // launch.EventTimeout. Without it an event handled longer than the ack wait was offered again while + // it was handled, and counted toward max-deliver (novox/hq issue 387). Kept alive too while it waits + // for an event from another of the module's consumers to be handled. + done, stopped := make(chan struct{}), make(chan struct{}) + go func() { + defer close(stopped) + tick := time.NewTicker(EventHeartbeat) + defer tick.Stop() + for { + select { + case <-done: + return + case <-tick.C: + _ = msg.InProgress() + } + } + }() + h.serial.Lock() + err := h.deliver(env) + h.serial.Unlock() + close(done) + // No in-progress after the answer: one after a Nak would stretch its delay to the whole ack wait. + <-stopped + if err != nil { + _ = msg.NakWithDelay(NakDelay) + return + } + _ = msg.Ack() +} + +// start binds what the membership names now, and follows it: a new membership binds what it adds and stops +// what it no longer names, and a consumer not bound yet is tried again every RebindEvery. +func (h *hearing) start() { + h.reconcile() + h.c.OnMembership(func(m Membership) { + if m.Module != h.module { + return + } + select { + case h.kick <- struct{}{}: + default: + } + }) + go func() { + tick := time.NewTicker(RebindEvery) + defer tick.Stop() + for { + select { + case <-h.done: + return + case <-h.kick: + case <-tick.C: + } + h.reconcile() + } + }() +} + +// stop stops every consumer on a module stream this module reads; the one on EVENTS is its caller's. +func (h *hearing) stop() { + h.mu.Lock() + defer h.mu.Unlock() + if h.stopped { + return + } + h.stopped = true + close(h.done) + for heard, reading := range h.bound { + reading.Stop() + delete(h.bound, heard) + } +} + +// reconcile makes what is bound what the module's membership names now. +func (h *hearing) reconcile() { + want := map[HeardIssued]bool{} + if m := h.c.Membership(h.module); m != nil { + for _, heard := range m.Hears { + if heard.Stream != "" && heard.Consumer != "" { + want[heard] = true + } + } + } + h.mu.Lock() + defer h.mu.Unlock() + if h.stopped { + return + } + for heard, reading := range h.bound { + if !want[heard] { + reading.Stop() + delete(h.bound, heard) + h.c.Logf("[mesh-tools] %s no longer hears %s: its membership stopped naming its consumer %s", + h.module, heard.Stream, heard.Consumer) + } + } + for heard := range h.said { + if !want[heard] { + delete(h.said, heard) + } + } + missing := make([]HeardIssued, 0, len(want)) + for heard := range want { + if h.bound[heard] == nil { + missing = append(missing, heard) + } + } + sort.Slice(missing, func(i, j int) bool { return missing[i].Stream < missing[j].Stream }) + for _, heard := range missing { + reading, doing, err := h.bind(heard.Stream, heard.Consumer) + if err != nil { + // Said once, not at every attempt; said again when it is bound. + if !h.said[heard] { + h.c.Logf("[mesh-tools] %s does not hear %s yet: %s its consumer %s failed: %v; tried again every %s", + h.module, heard.Stream, doing, heard.Consumer, err, RebindEvery) + h.said[heard] = true + } + continue + } + h.bound[heard] = reading + delete(h.said, heard) + h.c.Logf("[mesh-tools] %s hears %s on its consumer %s", h.module, heard.Stream, heard.Consumer) + } +} diff --git a/node-tools/internal/bus/streams_internal_test.go b/node-tools/internal/bus/streams_internal_test.go new file mode 100644 index 0000000..be4cefc --- /dev/null +++ b/node-tools/internal/bus/streams_internal_test.go @@ -0,0 +1,48 @@ +package bus + +import "testing" + +// novox/hq ADR 0363 §2, §3: an event goes to the stream whose pattern carries it, as the bus matches subjects. +func TestAnEventFallsUnderAStreamsPatternAsTheBusMatchesIt(t *testing.T) { + for _, c := range []struct { + pattern, event string + want bool + }{ + {"mail.*.message.added", "mail.home.message.added", true}, + {"mail.*.message.added", "mail.home.message.removed", false}, + {"mail.*.message.added", "mail.home.work.message.added", false}, + {"mail.>", "mail.home.message.added", true}, + {"mail.**", "mail.home", true}, + {"mail.>", "mail", false}, + {"status", "status", true}, + {"status", "status.changed", false}, + {"", "status", false}, + } { + if got := eventMatches(c.pattern, c.event); got != c.want { + t.Errorf("%q under %q: %v, want %v", c.event, c.pattern, got, c.want) + } + } +} + +// A membership naming no stream publishes every event on EVENTS, exactly as before (issue 725). +func TestAnEventNoStreamCarriesIsPublishedOnEvents(t *testing.T) { + mail := &Membership{Module: "mail", Streams: []StreamIssued{{Stream: "MAIL", Events: []string{"mail.*.message.added"}, + Subject: "mesh.stream.mail.event.{event}"}}} + for _, c := range []struct { + m *Membership + event string + want string + }{ + {nil, "mail.home.message.added", "mesh.mod.mail.event.mail.home.message.added"}, + {&Membership{Module: "mail"}, "mail.home.message.added", "mesh.mod.mail.event.mail.home.message.added"}, + {mail, "mail.home.message.added", "mesh.stream.mail.event.mail.home.message.added"}, + {mail, "status", "mesh.mod.mail.event.status"}, + } { + if got := eventSubject(c.m, "mail", c.event); got != c.want { + t.Errorf("%s: %s, want %s", c.event, got, c.want) + } + } + if got := keyOf("mesh.stream.mail.event.mail.home.message.added"); got != "mail.mail.home.message.added" { + t.Errorf("a stream's event reaches its consumer as %q, not as on EVENTS", got) + } +} diff --git a/node-tools/internal/bus/streams_test.go b/node-tools/internal/bus/streams_test.go new file mode 100644 index 0000000..317eeaf --- /dev/null +++ b/node-tools/internal/bus/streams_test.go @@ -0,0 +1,198 @@ +package bus_test + +import ( + "encoding/json" + "errors" + "fmt" + "sync" + "testing" + "time" + + "github.com/novox/mesh-tools/node-tools/internal/bus" + mt "github.com/novox/mesh-tools/node-tools/internal/meshtest" +) + +// streamMesh is a mesh with the mail module's own stream MAIL, and a connection on anchor following the +// memberships of the modules named. +func streamMesh(t *testing.T, memberships ...bus.Membership) (*mt.Mesh, *bus.Conn) { + t.Helper() + mesh := mt.New(t) + mesh.ModuleStream(t, "mail", "MAIL") + for _, m := range memberships { + mesh.Issue(t, m) + } + conn, err := bus.Connect(bus.Credential{URL: mt.URL(t), Module: "node-tools", Node: "anchor"}) + if err != nil { + t.Fatal(err) + } + conn.Logf = func(string, ...any) {} + t.Cleanup(conn.Close) + for _, m := range memberships { + conn.Follow(m.Module) + } + mt.Until(t, func() error { + for _, m := range memberships { + if conn.Membership(m.Module) == nil { + return errors.New(m.Module + " not issued yet") + } + } + return nil + }) + return mesh, conn +} + +// novox/hq ADR 0363 §2, §8 (issue 725): an event one of the module's streams carries is published on that +// stream's subject and kept by that stream alone; any other event of it goes to EVENTS as before. +func TestAnEventADeclaredStreamCarriesIsPublishedOnItsSubjectAlone(t *testing.T) { + mail := mt.MembershipOf("mail", "anchor", true, nil) + mail.Streams = []bus.StreamIssued{{Stream: "MAIL", Events: []string{"mail.*.message.added"}, + Subject: "mesh.stream.mail.event.{event}"}} + mesh, conn := streamMesh(t, mail) + + landed := mesh.NextEvent(t, "mesh.stream.mail.event.>") + if err := conn.PublishAs("mail", bus.Envelope{Key: "mail.home.message.added", Body: json.RawMessage(`{"id":1}`)}); err != nil { + t.Fatal(err) + } + select { + case got := <-landed: + if got != "mesh.stream.mail.event.mail.home.message.added" { + t.Errorf("published on %s", got) + } + case <-time.After(5 * time.Second): + t.Fatal("the event did not land on its stream's subject") + } + if err := conn.PublishAs("mail", bus.Envelope{Key: "status", Body: json.RawMessage(`{}`)}); err != nil { + t.Fatal(err) + } + if n := mesh.Messages(t, "MAIL"); n != 1 { + t.Errorf("MAIL holds %d messages, want the one event it carries", n) + } + if n := mesh.Messages(t, "EVENTS"); n != 1 { + t.Errorf("EVENTS holds %d messages, want only the event no stream carries", n) + } +} + +// novox/hq ADR 0363 §8 (issue 725): a module hears, beside EVENTS, every module stream its membership names, +// one event at a time, with the key it reads on EVENTS; a membership naming none hears EVENTS alone. +func TestAModuleHearsEveryStreamItsMembershipNames(t *testing.T) { + reader := mt.MembershipOf("reader", "anchor", false, nil) + mesh, conn := streamMesh(t, reader) + mesh.Consumer(t, "anchor", "reader", []string{"mesh.mod.alpha.event.>"}, 5*time.Second) + mesh.ConsumerOn(t, "MAIL", "anchor", "reader", []string{"mesh.stream.mail.event.mail.home.>"}, 5*time.Second) + + var mu sync.Mutex + heard := map[string]int{} + handling := 0 + stop, err := conn.ConsumeAs("reader", func(env bus.Envelope) error { + mu.Lock() + handling++ + if handling > 1 { + t.Error("two events were handed to the module at once") + } + mu.Unlock() + time.Sleep(50 * time.Millisecond) + mu.Lock() + handling-- + heard[env.Key]++ + mu.Unlock() + return nil + }) + if err != nil { + t.Fatal(err) + } + t.Cleanup(stop) + count := func(key string) int { + mu.Lock() + defer mu.Unlock() + return heard[key] + } + + // Its membership names no stream: EVENTS alone, as before. + mesh.Emit(t, "mesh.mod.alpha.event.happened", map[string]any{"n": 1}) + mesh.Emit(t, "mesh.stream.mail.event.mail.home.message.added", map[string]any{"n": 1}) + mt.Until(t, func() error { + if count("alpha.happened") != 1 { + return errors.New("the event on EVENTS was not heard") + } + return nil + }) + if _, notYet := mesh.PendingOn(t, "MAIL", "anchor", "reader"); notYet != 1 || count("mail.mail.home.message.added") != 0 { + t.Fatalf("a module whose membership names no stream heard MAIL (%d not yet delivered)", notYet) + } + + // Its membership now names its consumer on MAIL: bound, and what waited there is heard. + reader.Hears = []bus.HeardIssued{{Stream: "MAIL", Consumer: "anchor_reader"}} + mesh.Issue(t, reader) + mt.Until(t, func() error { + if count("mail.mail.home.message.added") != 1 { + return errors.New("the event on MAIL was not heard") + } + return nil + }) + for n := 2; n <= 4; n++ { + mesh.Emit(t, "mesh.mod.alpha.event.happened", map[string]any{"n": n}) + mesh.Emit(t, "mesh.stream.mail.event.mail.home.message.added", map[string]any{"n": n}) + } + mt.Until(t, func() error { + if a, m := count("alpha.happened"), count("mail.mail.home.message.added"); a != 4 || m != 4 { + return fmt.Errorf("heard %d on EVENTS and %d on MAIL, want 4 and 4", a, m) + } + for _, stream := range []string{"EVENTS", "MAIL"} { + if ack, notYet := mesh.PendingOn(t, stream, "anchor", "reader"); ack != 0 || notYet != 0 { + return fmt.Errorf("%s still holds %d unacknowledged, %d undelivered", stream, ack, notYet) + } + } + return nil + }) + + // A membership that stops naming it stops what it reads there. + reader.Hears = nil + mesh.Issue(t, reader) + mt.Until(t, func() error { + if mesh.Consumers(t, "MAIL") != 1 { + return errors.New("MAIL's consumer is gone") + } + mesh.Emit(t, "mesh.stream.mail.event.mail.home.message.added", map[string]any{"n": 0}) + time.Sleep(300 * time.Millisecond) + if _, notYet := mesh.PendingOn(t, "MAIL", "anchor", "reader"); notYet == 0 { + return errors.New("MAIL is still read") + } + return nil + }) +} + +// novox/hq ADR 0363 §6 (issue 725): a consumer the membership names before the controller made it is bound once +// it exists, without the module's events on EVENTS waiting for it. +func TestAConsumerNamedBeforeItExistsIsBoundOnceItDoes(t *testing.T) { + restore := bus.RebindEvery + bus.RebindEvery = 200 * time.Millisecond + t.Cleanup(func() { bus.RebindEvery = restore }) + reader := mt.MembershipOf("reader", "anchor", false, nil) + reader.Hears = []bus.HeardIssued{{Stream: "MAIL", Consumer: "anchor_reader"}} + mesh, conn := streamMesh(t, reader) + mesh.Consumer(t, "anchor", "reader", []string{"mesh.mod.alpha.event.>"}, 5*time.Second) + + var mu sync.Mutex + heard := map[string]int{} + stop, err := conn.ConsumeAs("reader", func(env bus.Envelope) error { + mu.Lock() + heard[env.Key]++ + mu.Unlock() + return nil + }) + if err != nil { + t.Fatalf("a consumer on a module stream not made yet kept the module from hearing EVENTS: %v", err) + } + t.Cleanup(stop) + mesh.Emit(t, "mesh.mod.alpha.event.happened", map[string]any{}) + mesh.ConsumerOn(t, "MAIL", "anchor", "reader", []string{"mesh.stream.mail.event.>"}, 5*time.Second) + mesh.Emit(t, "mesh.stream.mail.event.mail.home.message.added", map[string]any{}) + mt.Until(t, func() error { + mu.Lock() + defer mu.Unlock() + if heard["alpha.happened"] != 1 || heard["mail.mail.home.message.added"] != 1 { + return fmt.Errorf("heard %v", heard) + } + return nil + }) +} diff --git a/node-tools/internal/meshtest/meshtest.go b/node-tools/internal/meshtest/meshtest.go index 98f251c..9276e8f 100644 --- a/node-tools/internal/meshtest/meshtest.go +++ b/node-tools/internal/meshtest/meshtest.go @@ -271,3 +271,49 @@ func (m *Mesh) Consumers(t *testing.T, stream string) int { } return info.State.Consumers } + +// ModuleStream makes a module's own stream afresh as the controller does from the catalogue (novox/hq ADR +// 0363 §1): named in capitals, taking `mesh.stream..event.>`, a subject space EVENTS does not take. +func (m *Mesh) ModuleStream(t *testing.T, module, name string) { + t.Helper() + _ = m.js.DeleteStream(name) + if _, err := m.js.AddStream(&nats.StreamConfig{Name: name, Subjects: []string{"mesh.stream." + module + ".event.>"}, + Retention: nats.LimitsPolicy, Discard: nats.DiscardOld}); err != nil { + t.Fatal(err) + } +} + +// ConsumerOn makes a module's durable consumer on a machine on any stream, as Consumer does on EVENTS. +func (m *Mesh) ConsumerOn(t *testing.T, stream, node, module string, filters []string, ackWait time.Duration) { + t.Helper() + cfg := &nats.ConsumerConfig{Durable: node + "_" + module, AckPolicy: nats.AckExplicitPolicy, + AckWait: ackWait, MaxDeliver: 10, DeliverPolicy: nats.DeliverNewPolicy} + if len(filters) == 1 { + cfg.FilterSubject = filters[0] + } else { + cfg.FilterSubjects = filters + } + if _, err := m.js.AddConsumer(stream, cfg); err != nil { + t.Fatal(err) + } +} + +// PendingOn is what a module's consumer on a stream still holds, as Pending is on EVENTS. +func (m *Mesh) PendingOn(t *testing.T, stream, node, module string) (ackPending, notDelivered uint64) { + t.Helper() + info, err := m.js.ConsumerInfo(stream, node+"_"+module) + if err != nil { + t.Fatal(err) + } + return uint64(info.NumAckPending), info.NumPending +} + +// Messages is how many messages a stream holds. +func (m *Mesh) Messages(t *testing.T, stream string) uint64 { + t.Helper() + info, err := m.js.StreamInfo(stream) + if err != nil { + t.Fatal(err) + } + return info.State.Msgs +}