diff --git a/go/views/stdio_writer.go b/go/views/stdio_writer.go index e966ab88..a56a437c 100644 --- a/go/views/stdio_writer.go +++ b/go/views/stdio_writer.go @@ -10,13 +10,9 @@ import ( stdio "git.novox.be/novox/mesh-sdk/go" ) -// FindingsAskedEvent is the event `findings-asked`, always the seat mesh-conditions' event, named with that seat (design 47 G -// A-17); a stdio module's writer subscribes to it by this name, and no module supplies it. -const FindingsAskedEvent = "mesh-conditions.findings-asked" - // NewStdioWriter makes the finding writer of a stdio module (design 47 G A-16, §461, §485, §503): it fills `Publish` from // the module's own event publish (stdio.Emit, answered once the bus acknowledged it) and `Listen` from its subscription -// (stdio.SubscribeNew of FindingsAskedEvent, only the asks published after it is made), so no module writes that adapter. A `Publish` or `Listen` +// (stdio.SubscribeNew of the seat event the config's Asked subject names, only the asks published after it is made), so no module writes that adapter. A `Publish` or `Listen` // the config already holds is kept. The runtime publishes and subscribes by the event's name on the subject the module's // membership names for it, so each of the writer's four subjects (FindingSubjects) is told by its event. func NewStdioWriter[F, C, S, A any](cfg WriterConfig[F, C, S, A]) (*Writer[F, C, S, A], error) { @@ -36,8 +32,11 @@ func newStdioWriter[F, C, S, A any](cfg WriterConfig[F, C, S, A], } } if cfg.Listen == nil { - cfg.Listen = func(_ context.Context, _ string, handle func([]byte)) (func(), error) { - event := FindingsAskedEvent + cfg.Listen = func(_ context.Context, subject string, handle func([]byte)) (func(), error) { + event, err := seatEventOfSubject(subject) + if err != nil { + return nil, err + } var stopped atomic.Bool // the runtime keeps a subscription until the process ends: a stopped listener hands on nothing if err := subscribe(event, func(env stdio.Envelope) error { @@ -64,3 +63,13 @@ func eventOfSubject(subject string) (string, error) { } return "", fmt.Errorf("the subject %q is not of the form mesh.stream..event..", subject) } + +// seatEventOfSubject is `.` of a seat event's subject `mesh.seat..event.`, the name a runtime +// subscribes a seat's event by. +func seatEventOfSubject(subject string) (string, error) { + p := strings.Split(subject, ".") + if len(p) == 5 && p[0] == "mesh" && p[1] == "seat" && p[3] == "event" { + return p[2] + "." + p[4], nil + } + return "", fmt.Errorf("the subject %q is not of the form mesh.seat..event.", subject) +} diff --git a/go/views/stdio_writer_test.go b/go/views/stdio_writer_test.go index d1893bd0..b74a7272 100644 --- a/go/views/stdio_writer_test.go +++ b/go/views/stdio_writer_test.go @@ -48,7 +48,7 @@ func TestStdioWriterPublishesAndListensThroughTheModulesOwnPlumbing(t *testing.T var subscribed []string var deliver func(stdio.Envelope) error w, err := newStdioWriter(WriterConfig[swFinding, swCleared, swStated, swAsked]{ - Module: "m", Node: "n1", State: nil, + Module: "m", Node: "n1", State: nil, Asked: "mesh.seat.s.event.findings-asked", Observe: func(context.Context) (Observed[swFinding], error) { return Observed[swFinding]{}, nil }, }, func(event string, body any, o ...stdio.EmitOptions) error { raw, _ := json.Marshal(body) @@ -63,7 +63,7 @@ func TestStdioWriterPublishesAndListensThroughTheModulesOwnPlumbing(t *testing.T t.Fatalf("the writer still needs its state: %v", err) } w, err = newStdioWriter(WriterConfig[swFinding, swCleared, swStated, swAsked]{ - Module: "m", Node: "n1", State: noState{}, + Module: "m", Node: "n1", State: noState{}, Asked: "mesh.seat.s.event.findings-asked", Observe: func(context.Context) (Observed[swFinding], error) { return Observed[swFinding]{}, nil }, }, func(event string, body any, o ...stdio.EmitOptions) error { raw, _ := json.Marshal(body) @@ -84,8 +84,8 @@ func TestStdioWriterPublishesAndListensThroughTheModulesOwnPlumbing(t *testing.T t.Fatalf("the publish: %+v", out) } var heard []string - stop, err := w.cfg.Listen(context.Background(), w.subj.Asked, func(b []byte) { heard = append(heard, string(b)) }) - if err != nil || len(subscribed) != 1 || subscribed[0] != "mesh-conditions.findings-asked" { + stop, err := w.cfg.Listen(context.Background(), w.cfg.Asked, func(b []byte) { heard = append(heard, string(b)) }) + if err != nil || len(subscribed) != 1 || subscribed[0] != "s.findings-asked" { t.Fatalf("the listen: %v %v", subscribed, err) } _ = deliver(stdio.Envelope{Body: json.RawMessage(`{"ask":"x"}`)}) diff --git a/go/views/writer.go b/go/views/writer.go index fe1b9a60..965556ac 100644 --- a/go/views/writer.go +++ b/go/views/writer.go @@ -111,21 +111,17 @@ type FindingState interface { Keys() ([]string, error) } -// WriterSubjects are the subjects of the four events of an observer's module and node. +// WriterSubjects are the subjects of the three events an observer's module says on its own stream. type WriterSubjects struct { - Written, Cleared, Stated, Asked string + Written, Cleared, Stated string } -// FindingsAskedSubject is the subject the writer hears an ask for findings on: the seat mesh-conditions' event -// `findings-asked` (design 47 G §387, §390, A-17), never a stream of the part's own. -const FindingsAskedSubject = "mesh.seat.mesh-conditions.event.findings-asked" - // FindingSubjects are the subjects of the module's three own events on node, in §383's form -// `mesh.stream..event..` (design 47 G A-13, §461, §439, §383), and the seat's subject of the ask it hears -// (FindingsAskedSubject). The writer forms them itself; a module passes none. +// `mesh.stream..event..` (design 47 G A-13, §461, §439, §383). The writer forms them itself; a module +// passes none. The fourth, the subject of the ask it hears, is the seat's and is given (WriterConfig.Asked, A-17). func FindingSubjects(module, node string) WriterSubjects { of := func(event string) string { return "mesh.stream." + module + ".event." + event + "." + node } - return WriterSubjects{Written: of("finding-written"), Cleared: of("finding-cleared"), Stated: of("findings-stated"), Asked: FindingsAskedSubject} + return WriterSubjects{Written: of("finding-written"), Cleared: of("finding-cleared"), Stated: of("findings-stated")} } // WriterConfig makes a Writer. @@ -141,6 +137,9 @@ type WriterConfig[F, C, S, A any] struct { // JS is the bus connection's JetStream; Conn, when given, lets the writer hear the connection made again. JS jetstream.JetStream Conn *nats.Conn + // Asked is the subject the writer hears `findings-asked` on: the seat's event, which the module fills from the generated + // contract of the seat that defines it (design 47 G A-17); mesh-sdk names no seat. + Asked string // Node is the machine the module runs on: it forms the subjects of the four events (FindingSubjects) and a views // reader's findings name it in their subject (design 47 G §687, A-13); none given, MESH_NODE. Node string @@ -186,6 +185,8 @@ func NewWriter[F, C, S, A any](cfg WriterConfig[F, C, S, A]) (*Writer[F, C, S, A return nil, errors.New("a finding writer needs the state `findings`") case cfg.Observe == nil: return nil, errors.New("a finding writer needs the observe function") + case cfg.Asked == "": + return nil, errors.New("a finding writer needs the subject of findings-asked, from the generated contract of its seat") case cfg.JS == nil && (cfg.Publish == nil || cfg.Listen == nil): return nil, errors.New("a finding writer needs the bus connection's JetStream") } @@ -309,7 +310,7 @@ func (w *Writer[F, C, S, A]) Close() { } func (w *Writer[F, C, S, A]) listen() error { - stop, err := w.cfg.Listen(w.run, w.subj.Asked, w.heard) + stop, err := w.cfg.Listen(w.run, w.cfg.Asked, w.heard) if err != nil { return err } @@ -329,7 +330,7 @@ func (w *Writer[F, C, S, A]) Reconnected() { return } if err := w.listen(); err != nil { - w.cfg.Logf("the finding writer could not make its consumer of %s again: %v", w.subj.Asked, err) + w.cfg.Logf("the finding writer could not make its consumer of %s again: %v", w.cfg.Asked, err) } w.asks, w.want = append(w.asks, ""), true w.signal() diff --git a/go/views/writer_integration_test.go b/go/views/writer_integration_test.go index f697dffc..1a9a309d 100644 --- a/go/views/writer_integration_test.go +++ b/go/views/writer_integration_test.go @@ -31,12 +31,12 @@ func TestWriterSaysItsEventsAndHearsAsksOnTheBus(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() subj := views.FindingSubjects("m", "n1") - for name, subjects := range map[string][]string{"EVENTS": {subj.Written, subj.Cleared, subj.Stated}, "ASKS": {subj.Asked}} { + for name, subjects := range map[string][]string{"EVENTS": {subj.Written, subj.Cleared, subj.Stated}, "ASKS": {asked}} { if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: name, Subjects: subjects}); err != nil { t.Fatal(err) } } - if _, err := js.Publish(ctx, subj.Asked, []byte(`{"ask":"old","at":"2026-11-11T12:00:00Z","by":"keeper"}`)); err != nil { + if _, err := js.Publish(ctx, asked, []byte(`{"ask":"old","at":"2026-11-11T12:00:00Z","by":"keeper"}`)); err != nil { t.Fatal(err) } listen, err := js.CreateConsumer(ctx, "EVENTS", jetstream.ConsumerConfig{AckPolicy: jetstream.AckNonePolicy}) @@ -44,7 +44,7 @@ func TestWriterSaysItsEventsAndHearsAsksOnTheBus(t *testing.T) { t.Fatal(err) } state := &fakeState{m: map[string][]byte{}} - w, err := views.NewWriter(testConfig{Module: "m", State: state, JS: js, Conn: nc, Node: "n1", + w, err := views.NewWriter(testConfig{Module: "m", State: state, JS: js, Conn: nc, Node: "n1", Asked: asked, Observe: func(context.Context) (testObserved, error) { return testObserved{Findings: []testFinding{finding("a", "now")}}, nil }, Logf: t.Logf}) @@ -82,7 +82,7 @@ func TestWriterSaysItsEventsAndHearsAsksOnTheBus(t *testing.T) { for _, ask := range []string{"new", "again"} { w.Reconnected() next(subj.Stated) // the statement after the connection was made again - if _, err := js.Publish(ctx, subj.Asked, []byte(`{"ask":"`+ask+`","at":"2026-11-11T12:00:00Z","by":"keeper"}`)); err != nil { + if _, err := js.Publish(ctx, asked, []byte(`{"ask":"`+ask+`","at":"2026-11-11T12:00:00Z","by":"keeper"}`)); err != nil { t.Fatal(err) } if st := next(subj.Stated); st.Asked != ask { diff --git a/go/views/writer_test.go b/go/views/writer_test.go index 45b5ee53..e91af2e2 100644 --- a/go/views/writer_test.go +++ b/go/views/writer_test.go @@ -78,10 +78,13 @@ type rig struct { // subjects are the four subjects the writer forms itself in §383's form (design 47 G A-13); the literal is the form. var subjects = views.FindingSubjects("m", "n1") +// asked is the subject the module gives the writer for findings-asked, as its seat's generated contract has it (design 47 G A-17). +const asked = "mesh.seat.s.event.findings-asked" + func TestFindingSubjectsAreTheEventsStreamForm(t *testing.T) { want := views.WriterSubjects{ Written: "mesh.stream.m.event.finding-written.n1", Cleared: "mesh.stream.m.event.finding-cleared.n1", - Stated: "mesh.stream.m.event.findings-stated.n1", Asked: "mesh.seat.mesh-conditions.event.findings-asked", + Stated: "mesh.stream.m.event.findings-stated.n1", } if subjects != want { t.Fatalf("the subjects: %+v, want %+v", subjects, want) @@ -94,7 +97,7 @@ func newRigKinds(t *testing.T, kinds ...string) *rig { r := &rig{t: t, state: &fakeState{m: map[string][]byte{}}, now: time.Date(2026, 11, 11, 12, 0, 0, 0, time.UTC), ch: make(chan struct{}, 100)} r.observe = func(context.Context) (testObserved, error) { return testObserved{}, nil } w, err := views.NewWriter(testConfig{ - Module: "m", State: r.state, Kinds: kinds, Node: "n1", + Module: "m", State: r.state, Kinds: kinds, Node: "n1", Asked: asked, Observe: func(ctx context.Context) (testObserved, error) { return r.observe(ctx) }, Logf: func(f string, a ...any) { r.mu.Lock() @@ -123,7 +126,7 @@ func newRigKinds(t *testing.T, kinds ...string) *rig { Listen: func(_ context.Context, subject string, handle func([]byte)) (func(), error) { r.mu.Lock() defer r.mu.Unlock() - if subject != subjects.Asked { + if subject != asked { t.Errorf("listening on %s", subject) } r.listens++ @@ -444,11 +447,19 @@ func TestWriterRefusesABodyWithoutAContractField(t *testing.T) { At time.Time `json:"at"` Subject testSubject } - _, err := views.NewWriter(views.WriterConfig[noSince, testCleared, testStated, testAsked]{Module: "m", State: &fakeState{m: map[string][]byte{}}, Node: "n1", + _, err := views.NewWriter(views.WriterConfig[noSince, testCleared, testStated, testAsked]{Module: "m", State: &fakeState{m: map[string][]byte{}}, Node: "n1", Asked: asked, Observe: func(context.Context) (views.Observed[noSince], error) { return views.Observed[noSince]{}, nil }, Publish: func(context.Context, string, []byte, string) error { return nil }, Listen: func(context.Context, string, func([]byte)) (func(), error) { return func() {}, nil }}) if err == nil || !strings.Contains(err.Error(), `"kind"`) { t.Fatalf("a body of other wire names: %v", err) } + // the subject of findings-asked is the seat's and the module's to give; none given is refused (design 47 G A-17) + _, err = views.NewWriter(testConfig{Module: "m", State: &fakeState{m: map[string][]byte{}}, Node: "n1", + Observe: func(context.Context) (testObserved, error) { return testObserved{}, nil }, + Publish: func(context.Context, string, []byte, string) error { return nil }, + Listen: func(context.Context, string, func([]byte)) (func(), error) { return func() {}, nil }}) + if err == nil || !strings.Contains(err.Error(), "findings-asked") { + t.Fatalf("a writer with no subject for findings-asked: %v", err) + } } diff --git a/src/views/writer.ts b/src/views/writer.ts index cfcb018f..2ab42a74 100644 --- a/src/views/writer.ts +++ b/src/views/writer.ts @@ -3,13 +3,11 @@ // working on their wire form, with no layout of a finding of its own (design 47 G A-3, §640). It keeps an observer's findings in its own state // `findings` and says them as events (§383, §384). It names the four events itself (design 47 G A-13, A-17): it says // `finding-written`, `finding-cleared` and `findings-stated` by their names, which the runtime publishes on the subject the -// module's membership names for each, and hears `findings-asked` as the seat mesh-conditions' event; a module passes none. +// module's membership names for each; the event `findings-asked` is the seat's and is given as `asked` (A-17: mesh-sdk names no seat). // Its acts, `write`, `clear` and a statement, run one at a time in the order they are asked, on one chain of promises. import type { State } from "../state/index.js"; -/** The event `findings-asked` is always the seat mesh-conditions' (design 47 G A-17), named with that seat. */ -export const FINDINGS_ASKED_EVENT = "mesh-conditions.findings-asked"; /** The events the writer says, by their names (design 47 G §383, A-13). */ export const FINDING_WRITTEN_EVENT = "finding-written"; export const FINDING_CLEARED_EVENT = "finding-cleared"; @@ -94,6 +92,8 @@ export interface WriterOptions; + /** The event the writer hears `findings-asked` as: the seat's, which the module fills from the seat's generated contract (design 47 G A-17). */ + asked: string; /** The finding kinds this writer owns (design 47 G §580): a write or clear of another kind is refused, naming it, and a statement carries them as `kinds` and deletes only keys of them. None: every kind. The views reader's kinds are always its own besides. */ kinds?: string[]; /** The findings the observer observes now and its pending keys. It is called inside the writer's one chain: it must not call `write` or `clear`. */ @@ -174,7 +174,7 @@ export class Writer { - const stop = await this.o.listen(FINDINGS_ASKED_EVENT, (body) => this.heard(body)); + const stop = await this.o.listen(this.o.asked, (body) => this.heard(body)); this.stop?.(); this.stop = stop; } diff --git a/test/sdk.test.ts b/test/sdk.test.ts index 3db384a9..0f7ac149 100644 --- a/test/sdk.test.ts +++ b/test/sdk.test.ts @@ -860,6 +860,7 @@ test("the finding writer writes, clears and states findings as the Go writer doe const logs: string[] = []; const w = new Writer({ module: "m", + asked: "s.findings-asked", state: st, observe: async () => { @@ -870,7 +871,7 @@ test("the finding writer writes, clears and states findings as the Go writer doe if (refuse > 0) { refuse--; throw new Error("no ack"); } sent.push({ subject, body }); }, - listen: async (s, h) => { assert.equal(s, "mesh-conditions.findings-asked"); consumers++; ask = h; return () => {}; }, + listen: async (s, h) => { assert.equal(s, "s.findings-asked"); consumers++; ask = h; return () => {}; }, log: (l) => logs.push(l), now: () => new Date(clock), wait: async (ms) => { waits.push(ms); clock += ms; return true; }, @@ -945,7 +946,7 @@ test("the finding writer owns its kinds, keeps a views reader's findings in its }; const sent: { subject: string; body: any }[] = []; const w = new Writer({ - module: "m", state: st, kinds: ["k"], + module: "m", asked: "s.findings-asked", state: st, kinds: ["k"], observe: async () => ({ findings: [], pending: [] }), publish: async (subject, body) => void sent.push({ subject, body }), listen: async () => () => {},