jschoubben
adb0acdfae
Writer[F, C, S, A any] marshals a body, reads and sets kind, scope, id, subject, at, since and observer by their contract names; the interfaces FindingBody, ClearedBody, StatedBody, AskedBody and the constructors go. A body whose wire form lacks a field the contract names is refused when the writer is made. views.CheckFinding(any) judges the wire form. WriterFindings takes only the writer. Builds: design 47 G §686, §640, §687, §689, A-3 (.work/47-G/dev2-ledger.md) Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ
93 lines
2.9 KiB
Go
93 lines
2.9 KiB
Go
//go:build integration
|
|
|
|
package views_test
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/nats-io/nats.go"
|
|
"github.com/nats-io/nats.go/jetstream"
|
|
|
|
"git.novox.be/novox/mesh-sdk/go/testbus"
|
|
"git.novox.be/novox/mesh-sdk/go/views"
|
|
)
|
|
|
|
// Against a real bus: the writer's events are acknowledged by their stream, a start states what the observer observes, an
|
|
// ask published before the writer's consumer existed is never answered, one published after is, and after the connection
|
|
// is made again (Reconnected) the consumer is made again from the newest ask (design 47 G §383, §390, §485).
|
|
func TestWriterSaysItsEventsAndHearsAsksOnTheBus(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.WithCancel(context.Background())
|
|
defer cancel()
|
|
for name, subj := range map[string]string{"EVENTS": "obs.event.>", "ASKS": "obs.ask"} {
|
|
if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: name, Subjects: []string{subj}}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
if _, err := js.Publish(ctx, "obs.ask", []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})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
state := &fakeState{m: map[string][]byte{}}
|
|
w, err := views.NewWriter(testConfig{Module: "m", State: state, JS: js, Conn: nc, Node: "n1",
|
|
Subjects: views.WriterSubjects{Written: "obs.event.written", Cleared: "obs.event.cleared", Stated: "obs.event.stated", Asked: "obs.ask"},
|
|
Observe: func(context.Context) (testObserved, error) {
|
|
return testObserved{Findings: []testFinding{finding("a", "now")}}, nil
|
|
}, Logf: t.Logf})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := w.Start(ctx); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer w.Close()
|
|
next := func(want string) testStated {
|
|
t.Helper()
|
|
m, err := listen.Next(jetstream.FetchMaxWait(10 * time.Second))
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if m.Subject() != want {
|
|
t.Fatalf("said %s, want %s", m.Subject(), want)
|
|
}
|
|
var st testStated
|
|
if want == "obs.event.stated" {
|
|
if err := json.Unmarshal(m.Data(), &st); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
return st
|
|
}
|
|
if st := next("obs.event.stated"); st.Asked != "" || len(st.Findings) != 1 {
|
|
t.Fatalf("start statement: %+v", st)
|
|
}
|
|
if err := w.Write(ctx, finding("b", "x")); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
next("obs.event.written")
|
|
for _, ask := range []string{"new", "again"} {
|
|
w.Reconnected()
|
|
next("obs.event.stated") // the statement after the connection was made again
|
|
if _, err := js.Publish(ctx, "obs.ask", []byte(`{"ask":"`+ask+`","at":"2026-11-11T12:00:00Z","by":"keeper"}`)); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if st := next("obs.event.stated"); st.Asked != ask {
|
|
t.Fatalf("answered %q, want %q", st.Asked, ask)
|
|
}
|
|
}
|
|
}
|