diff --git a/go/process/emit.go b/go/process/emit.go index f6746658..a489afcf 100644 --- a/go/process/emit.go +++ b/go/process/emit.go @@ -19,6 +19,18 @@ import ( // (EmitOptions.MessageID) is sent as x-event-id, and as the bus's own dedup id (Nats-Msg-Id), so a second // publish inside the stream's duplicate window is dropped; empty is a random one, as stdio.Emit on main. func (c *Conn) Emit(eventType string, body any, opts ...stdio.EmitOptions) error { + return c.emitOn("mesh.mod."+c.account.Module+".event."+eventType, body, opts) +} + +// EmitSeat says a seat's event as its holder (design 47 G A-10, §276, §498): one call, publishing on +// mesh.seat..event., with the same envelope, headers and message id as Emit. The grants part grants every +// holder the publish of its held seats' event subjects. +func (c *Conn) EmitSeat(seat, event string, body any, opts ...stdio.EmitOptions) error { + return c.emitOn("mesh.seat."+seat+".event."+event, body, opts) +} + +// emitOn publishes body on subject with the SDK's envelope. +func (c *Conn) emitOn(subject string, body any, opts []stdio.EmitOptions) error { data, err := json.Marshal(body) if err != nil { return err @@ -32,7 +44,7 @@ func (c *Conn) Emit(eventType string, body any, opts ...stdio.EmitOptions) error if id == "" { id = eventID() } - msg := nats.NewMsg("mesh.mod." + c.account.Module + ".event." + eventType) + msg := nats.NewMsg(subject) msg.Header.Set("x-event-id", id) msg.Header.Set("x-source", c.account.Module) msg.Header.Set("x-node", c.account.Node) diff --git a/go/process/emit_integration_test.go b/go/process/emit_integration_test.go index 1663b6e6..46dc2036 100644 --- a/go/process/emit_integration_test.go +++ b/go/process/emit_integration_test.go @@ -19,7 +19,7 @@ func TestEmitTwiceWithTheSameIDIsKeptToOneWithTheSDKsHeaders(t *testing.T) { account.Node = "novox" _, adminJetStream := adminJS(t, url) stream, err := adminJetStream.CreateStream(ctxFor(t), jetstream.StreamConfig{ - Name: "MAIL_EVENTS", Subjects: []string{"mesh.mod.mesh-mail.event.>"}, Duplicates: time.Minute, + Name: "MAIL_EVENTS", Subjects: []string{"mesh.mod.mesh-mail.event.>", "mesh.seat.mail-sender.event.>"}, Duplicates: time.Minute, }) if err != nil { t.Fatalf("the admin could not make the events stream: %v", err) @@ -37,12 +37,20 @@ func TestEmitTwiceWithTheSameIDIsKeptToOneWithTheSDKsHeaders(t *testing.T) { } } + // A seat's event goes on mesh.seat..event. with the same headers (design 47 G A-10). + if err := conn.EmitSeat("mail-sender", "sent", map[string]any{"send": "s1"}, stdio.EmitOptions{MessageID: "seat-s1"}); err != nil { + t.Fatalf("EmitSeat: %v", err) + } + if m, err := stream.GetLastMsgForSubject(ctxFor(t), "mesh.seat.mail-sender.event.sent"); err != nil || m.Header.Get("x-event-id") != "seat-s1" || m.Header.Get("x-source") != "mesh-mail" { + t.Fatalf("the seat's event: %v, %v", m, err) + } + info, err := stream.Info(ctxFor(t)) if err != nil { t.Fatal(err) } - if info.State.Msgs != 1 { - t.Fatalf("the stream holds %d messages, want exactly 1 (the second publish is a duplicate)", info.State.Msgs) + if info.State.Msgs != 2 { + t.Fatalf("the stream holds %d messages, want 2: one of the doubled publish, one of the seat's event", info.State.Msgs) } msg, err := stream.GetMsg(ctxFor(t), 1) diff --git a/go/process/fixture_integration_test.go b/go/process/fixture_integration_test.go index 91ed0a3f..ff69c644 100644 --- a/go/process/fixture_integration_test.go +++ b/go/process/fixture_integration_test.go @@ -39,6 +39,7 @@ var sendGrants = []string{ "mesh.mod.mesh-mail.event.mail.send.failed", "mesh.mod.mesh-mail.event.mail.send.sent", "mesh.mod.mesh-mail.event.states", + "mesh.seat.mail-sender.event.sent", // a held seat's event, said through Conn.EmitSeat (design 47 G A-10) "mesh.seat.operator-channel.call.tell.*.novox~mesh-mail~send", "mesh.seat.operator-channel.call.tell.novox~mesh-mail~send", "mesh.seat.operator-channel.tool.tell",