jschoubben
3327f8679d
A process publishes on mesh.seat.<seat>.accept.<kind>.<caller>, its module the caller, as Emit publishes an event, so the gate's channel over a connection can ask and cancel through the seat operator-channel. Builds: design 47 G §677, §700; lead answers A-272, A-273 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
74 lines
2.8 KiB
Go
74 lines
2.8 KiB
Go
//go:build integration
|
|
|
|
package process
|
|
|
|
import (
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/nats-io/nats.go/jetstream"
|
|
|
|
stdio "git.novox.be/novox/mesh-sdk/go"
|
|
)
|
|
|
|
// 1188/AC3: given a stream accepting mail.send.sent, when the process emits it twice with the same
|
|
// message id inside the duplicate window, then the stream holds one message; with the headers stdio's
|
|
// Emit sets (stdio.go:498-509).
|
|
func TestEmitTwiceWithTheSameIDIsKeptToOneWithTheSDKsHeaders(t *testing.T) {
|
|
url, account := grantedSendProcess(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.>", "mesh.seat.mail-sender.event.>", "mesh.seat.mail-sender.accept.>"}, Duplicates: time.Minute,
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("the admin could not make the events stream: %v", err)
|
|
}
|
|
|
|
conn, err := Connect(account)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer conn.Close()
|
|
|
|
for i := 0; i < 2; i++ {
|
|
if err := conn.Emit("mail.send.sent", map[string]any{"send": "s1"}, stdio.EmitOptions{MessageID: "send-s1"}); err != nil {
|
|
t.Fatalf("emit %d: %v", i, err)
|
|
}
|
|
}
|
|
|
|
// A seat's event goes on mesh.seat.<seat>.event.<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)
|
|
}
|
|
|
|
// An accepted kind goes on mesh.seat.<seat>.accept.<kind>.<caller>, the caller this process's module (A-273).
|
|
if err := conn.Accept("mail-sender", "send", map[string]any{"send": "s2"}, stdio.EmitOptions{MessageID: "accept-s2"}); err != nil {
|
|
t.Fatalf("Accept: %v", err)
|
|
}
|
|
if m, err := stream.GetLastMsgForSubject(ctxFor(t), "mesh.seat.mail-sender.accept.send.mesh-mail"); err != nil || m.Header.Get("x-event-id") != "accept-s2" || m.Header.Get("x-source") != "mesh-mail" {
|
|
t.Fatalf("the accepted kind: %v, %v", m, err)
|
|
}
|
|
|
|
info, err := stream.Info(ctxFor(t))
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if info.State.Msgs != 3 {
|
|
t.Fatalf("the stream holds %d messages, want 3: one of the doubled publish, one of the seat's event, one accepted", info.State.Msgs)
|
|
}
|
|
|
|
msg, err := stream.GetMsg(ctxFor(t), 1)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if msg.Header.Get("x-event-id") != "send-s1" || msg.Header.Get("x-source") != "mesh-mail" ||
|
|
msg.Header.Get("x-node") != "novox" || msg.Header.Get("content-type") != "application/json" ||
|
|
msg.Header.Get("x-time") == "" {
|
|
t.Errorf("the stored message's headers are not the SDK's: %v", msg.Header)
|
|
}
|
|
}
|