diff --git a/go/process/event_hub_integration_test.go b/go/process/event_hub_integration_test.go index f8fa9bfa..00ee817f 100644 --- a/go/process/event_hub_integration_test.go +++ b/go/process/event_hub_integration_test.go @@ -65,3 +65,58 @@ func TestTheHubBindsALateConsumerAndSaysItIsLive(t *testing.T) { } hub.Stop() } + +// W1.12: a module hears a seat's event kept last per subject on its own stream through its durable, made by the maker +// after the events were said, and is handed the last of each subject; without the consumer the call says so. +func TestAModuleHearsASeatEventKeptOnItsOwnStream(t *testing.T) { + nc, err := nats.Connect(testbus.URL(t)) + if err != nil { + t.Fatal(err) + } + defer nc.Close() + js, err := jetstream.New(nc) + if err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + kept, subjects := "node-engine.running", []string{"mesh.seat.node-engine.event.running.>"} + if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: process.SeatStreamName(kept), Subjects: subjects, MaxMsgsPerSubject: 1}); err != nil { + t.Fatal(err) + } + for _, m := range []struct{ subject, fp string }{ + {"mesh.seat.node-engine.event.running.ace.mesh-tools", "fp1"}, + {"mesh.seat.node-engine.event.running.ace.mesh-tools", "fp2"}, + {"mesh.seat.node-engine.event.running.g14.mesh-tools", "fp3"}, + } { + if _, err := js.Publish(ctx, m.subject, []byte(m.fp)); err != nil { + t.Fatal(err) + } + } + cred := process.Account{Node: "ace", Module: "mesh-grants"} + got := make(chan string, 4) + hear := func(m jetstream.Msg) { got <- m.Subject() + "=" + string(m.Data()); _ = m.Ack() } + if _, err := cred.HearSeatEvents(ctx, js, kept, hear, t.Logf); err == nil { + t.Fatal("no consumer was made, and hearing said nothing") + } + if _, err := js.CreateOrUpdateConsumer(ctx, process.SeatStreamName(kept), cred.SeatEventConsumerConfig(subjects, true)); err != nil { + t.Fatal(err) + } + cc, err := cred.HearSeatEvents(ctx, js, kept, hear, t.Logf) + if err != nil { + t.Fatal(err) + } + defer cc.Stop() + heard := map[string]bool{} + for len(heard) < 2 { + select { + case h := <-got: + heard[h] = true + case <-ctx.Done(): + t.Fatalf("heard %v", heard) + } + } + if !heard["mesh.seat.node-engine.event.running.ace.mesh-tools=fp2"] || !heard["mesh.seat.node-engine.event.running.g14.mesh-tools=fp3"] { + t.Errorf("heard %v, want the last of each subject", heard) + } +} diff --git a/go/process/event_hub_test.go b/go/process/event_hub_test.go index 790e3f59..07acd8d8 100644 --- a/go/process/event_hub_test.go +++ b/go/process/event_hub_test.go @@ -35,4 +35,18 @@ func TestTheHubMatchesSubjectsAndRefusesOverlappingPatterns(t *testing.T) { if err := h.Handle("mesh.mod.q.event.>", nop); err != nil { t.Fatal(err) } + // W1.12: a seat's event kept in a stream of its own is heard on one bus name and one durable, the maker's and the + // hearer's alike; an event kept last per subject is delivered from the last of each subject. + a := Account{Node: "ace", Module: "mesh-grants"} + if got := SeatStreamName("node-engine.running"); got != "SEAT_node-engine_running" { + t.Errorf("the bus's name of node-engine.running is %q", got) + } + cfg := a.SeatEventConsumerConfig([]string{"mesh.seat.node-engine.event.running.>"}, true) + if cfg.Durable != "ace_mesh-grants" || cfg.DeliverPolicy != jetstream.DeliverLastPerSubjectPolicy || + cfg.AckPolicy != jetstream.AckExplicitPolicy || len(cfg.FilterSubjects) != 1 { + t.Errorf("the durable on a stream kept last per subject: %+v", cfg) + } + if cfg := a.SeatEventConsumerConfig(nil, false); cfg.DeliverPolicy != jetstream.DeliverAllPolicy { + t.Errorf("the durable on a stream kept with an age starts from its first message, got %v", cfg.DeliverPolicy) + } } diff --git a/go/process/seatstream.go b/go/process/seatstream.go new file mode 100644 index 00000000..c1295dab --- /dev/null +++ b/go/process/seatstream.go @@ -0,0 +1,60 @@ +package process + +import ( + "context" + "fmt" + "strings" + + "github.com/nats-io/nats.go/jetstream" +) + +// A seat's event kept in a stream of its own (seatdef's `keep`: the last message of each subject, or a stream with an age +// and a cap), not on the shared events stream (novox/hq design 47: the node-engine seat's `running..`, the +// seat mesh-delivery's `owed`). A generated contract names that stream as `.` (Stream, seatdef.KeptStream); +// these are the one rule, for the part that makes the stream and its consumers (the grants part, from each module's +// `consumes`) and for every part that hears it, of the stream's name on the bus and the module's durable on it. + +// SeatStreamPrefix starts the bus's name of every stream a seat's event is kept in apart from its family. +const SeatStreamPrefix = "SEAT_" + +// SeatStreamName is the bus's name of the stream a contract names kept (`.`): the bus takes no dot in a +// stream's name, so `SEAT__`, each dot an underscore (`node-engine.running` is `SEAT_node-engine_running`). +func SeatStreamName(kept string) string { + return SeatStreamPrefix + strings.ReplaceAll(kept, ".", "_") +} + +// SeatEventConsumerConfig is the module's durable on a seat event's own stream as its maker makes it: _, as +// on the shared stream, filtered to the subjects the module consumes, acknowledged one by one; for an event kept last +// per subject it starts from the last message of each subject, never from now, so a consumer made after a subject was +// last said still hears it, and two parts' consumers made at different times hold the same (design 47). +func (a Account) SeatEventConsumerConfig(subjects []string, lastPerSubject bool) jetstream.ConsumerConfig { + cfg := jetstream.ConsumerConfig{Durable: a.EventConsumerName(), AckPolicy: jetstream.AckExplicitPolicy, FilterSubjects: subjects} + if lastPerSubject { + cfg.DeliverPolicy = jetstream.DeliverLastPerSubjectPolicy + } + return cfg +} + +// HearSeatEvents binds the module's durable on the stream of kept (a contract's Stream) and pulls it, handing each +// message to handle, which acknowledges it; every error the bus reports while hearing is said through logf. Like +// EventHub it makes neither the stream nor the consumer: one the bus does not hold yet is this call's error, naming +// both, and a later call binds it. The caller stops the returned context. +func (a Account) HearSeatEvents(ctx context.Context, js jetstream.JetStream, kept string, handle func(jetstream.Msg), logf func(string, ...any)) (jetstream.ConsumeContext, error) { + if handle == nil { + return nil, fmt.Errorf("hearing %s needs a handler", kept) + } + if logf == nil { + logf = func(string, ...any) {} + } + stream := SeatStreamName(kept) + cons, err := a.EventConsumer(ctx, js, stream) + if err != nil { + return nil, fmt.Errorf("the consumer %s on %s (the seat's stream %s) is not there: %w", a.EventConsumerName(), stream, kept, err) + } + cc, err := HearEvents(cons, handle, logf) + if err != nil { + return nil, fmt.Errorf("the consumer %s on %s (the seat's stream %s) is not heard: %w", a.EventConsumerName(), stream, kept, err) + } + logf("events: the consumer %s on %s (the seat's stream %s) is heard", a.EventConsumerName(), stream, kept) + return cc, nil +}