G A-16: views.NewStdioWriter fills Publish and Listen from the stdio module's own event publish and subscription

Builds: design 47 G A-16, §461, §485, §503 Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ
2026-11-11 21:43:32 +01:00
parent 68670789d1
commit 43960cc925
2 changed files with 172 additions and 0 deletions
@@ -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.<module>.event.<event>.<node>`.
+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.<module>.event.<event>.<node>", subject)
+}
@@ -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 }