diff --git a/go/views/stdio_writer.go b/go/views/stdio_writer.go new file mode 100644 index 00000000..d8b2aa4c --- /dev/null +++ b/go/views/stdio_writer.go @@ -0,0 +1,65 @@ +package views + +import ( + "context" + "encoding/json" + "fmt" + "strings" + "sync/atomic" + + stdio "git.novox.be/novox/mesh-sdk/go" +) + +// 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, 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) { + return newStdioWriter(cfg, stdio.Emit, stdio.SubscribeNew) +} + +func newStdioWriter[F, C, S, A any](cfg WriterConfig[F, C, S, A], + emit func(string, any, ...stdio.EmitOptions) error, + subscribe func(string, func(stdio.Envelope) error) error) (*Writer[F, C, S, A], error) { + if cfg.Publish == nil { + cfg.Publish = func(_ context.Context, subject string, data []byte, id string) error { + event, err := eventOfSubject(subject) + if err != nil { + return err + } + return emit(event, json.RawMessage(data), stdio.EmitOptions{MessageID: id, Ack: true}) + } + } + if cfg.Listen == nil { + cfg.Listen = func(_ context.Context, subject string, handle func([]byte)) (func(), error) { + event, err := eventOfSubject(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 { + if !stopped.Load() { + handle(env.Body) + } + return nil + }); err != nil { + return nil, fmt.Errorf("subscribing to %s: %w", event, err) + } + return func() { stopped.Store(true) }, nil + } + } + return NewWriter(cfg) +} + +// eventOfSubject is the event of a subject of FindingSubjects: `mesh.stream..event..`. +func eventOfSubject(subject string) (string, error) { + p := strings.Split(subject, ".") + for i, s := range p { + if s == "event" && i+2 == len(p)-1 && i > 0 { + return p[i+1], nil + } + } + return "", fmt.Errorf("the subject %q is not of the form mesh.stream..event..", subject) +} diff --git a/go/views/stdio_writer_test.go b/go/views/stdio_writer_test.go new file mode 100644 index 00000000..c0262d9d --- /dev/null +++ b/go/views/stdio_writer_test.go @@ -0,0 +1,107 @@ +package views + +import ( + "context" + "encoding/json" + "strings" + "testing" + + stdio "git.novox.be/novox/mesh-sdk/go" +) + +type swFinding struct { + Kind string `json:"kind"` + Scope string `json:"scope"` + ID string `json:"id"` + Observer string `json:"observer"` + At string `json:"at"` + Since string `json:"since"` + Subject struct { + Scope string `json:"scope"` + } `json:"subject"` +} +type swCleared struct { + Kind string `json:"kind"` + Scope string `json:"scope"` + ID string `json:"id"` + At string `json:"at"` + Why string `json:"why"` +} +type swStated struct { + At string `json:"at"` + Findings []swFinding `json:"findings"` + Pending []PendingKey `json:"pending"` +} +type swAsked struct { + Ask string `json:"ask"` +} + +// A stdio module's writer says each of its events through the module's own publish, acknowledged, under the writer's +// message id, and hears the asks through the module's subscription (design 47 G A-16). +func TestStdioWriterPublishesAndListensThroughTheModulesOwnPlumbing(t *testing.T) { + type said struct { + event string + body string + o stdio.EmitOptions + } + var out []said + var subscribed []string + var deliver func(stdio.Envelope) error + w, err := newStdioWriter(WriterConfig[swFinding, swCleared, swStated, swAsked]{ + Module: "m", Node: "n1", State: nil, + Observe: func(context.Context) (Observed[swFinding], error) { return Observed[swFinding]{}, nil }, + }, func(event string, body any, o ...stdio.EmitOptions) error { + raw, _ := json.Marshal(body) + out = append(out, said{event, string(raw), o[0]}) + return nil + }, func(pattern string, h func(stdio.Envelope) error) error { + subscribed = append(subscribed, pattern) + deliver = h + return nil + }) + if err == nil || !strings.Contains(err.Error(), "state") { + t.Fatalf("the writer still needs its state: %v", err) + } + w, err = newStdioWriter(WriterConfig[swFinding, swCleared, swStated, swAsked]{ + Module: "m", Node: "n1", State: noState{}, + Observe: func(context.Context) (Observed[swFinding], error) { return Observed[swFinding]{}, nil }, + }, func(event string, body any, o ...stdio.EmitOptions) error { + raw, _ := json.Marshal(body) + out = append(out, said{event, string(raw), o[0]}) + return nil + }, func(pattern string, h func(stdio.Envelope) error) error { + subscribed = append(subscribed, pattern) + deliver = h + return nil + }) + if err != nil { + t.Fatal(err) + } + if err := w.cfg.Publish(context.Background(), w.subj.Written, []byte(`{"a":1}`), "id-1"); err != nil { + t.Fatal(err) + } + if len(out) != 1 || out[0].event != "finding-written" || out[0].body != `{"a":1}` || out[0].o.MessageID != "id-1" || !out[0].o.Ack { + 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] != "findings-asked" { + t.Fatalf("the listen: %v %v", subscribed, err) + } + _ = deliver(stdio.Envelope{Body: json.RawMessage(`{"ask":"x"}`)}) + stop() + _ = deliver(stdio.Envelope{Body: json.RawMessage(`{"ask":"y"}`)}) + if len(heard) != 1 || heard[0] != `{"ask":"x"}` { + t.Fatalf("heard %v: a stopped listener hands on nothing", heard) + } + if err := w.cfg.Publish(context.Background(), "obs.event", nil, ""); err == nil { + t.Fatal("a subject of no event form is refused") + } +} + +type noState struct{} + +func (noState) Get(string) (*stdio.StateEntry, error) { return nil, nil } +func (noState) Put(string, any) (uint64, error) { return 0, nil } +func (noState) Delete(string) error { return nil } +func (noState) Keys() ([]string, error) { return nil, nil }