jschoubben
aa74bd86ca
Task 3.9's other half and 1.4's missing client. The derivation is pure and unit-tested; only "does the server accept this" needs one running, behind MESH_TEST_NATS so the ordinary suite stays offline. A seat's work queue is created at registration, not assignment, so work queues until a holder appears — a stream created at assignment would make "the holder is not here yet" mean "your messages are gone". Named after the seat, because the holder can change and the queued work must not care. A holder's worker uses a queue group even though the seat guarantees one holder: the seat is authority, the queue group is delivery, and tying them together means the day somebody allows two holders every message is processed twice with nothing reporting it. One consumer per module carrying every filter, because its ack permission is derived from its name. And a real bug the live server caught: a durable name may not contain a dot, but an ack subject is $JS.ACK.<stream>.<consumer>, so the single string that read correctly inside the permission was rejected as a consumer name. Split in two, beside the permission that has to match. Unfixed, the symptom would have been every message redelivered forever with a permission list that looks right — which is the failure design 25 §4 warns about.
72 lines
2.1 KiB
Go
72 lines
2.1 KiB
Go
package broker
|
|
|
|
import (
|
|
"os"
|
|
"testing"
|
|
)
|
|
|
|
// Against a real server, because the questions here are all "does the server accept this" —
|
|
// which a mock would answer by agreeing with whatever this file already believes.
|
|
//
|
|
// Skipped unless MESH_TEST_NATS names one, so the ordinary suite stays fast and offline:
|
|
//
|
|
// docker run -d --rm --name t -p 14222:4222 nats:2.10-alpine -js
|
|
// MESH_TEST_NATS=nats://127.0.0.1:14222 go test ./internal/broker/ -run TestAgainstARealServer
|
|
func TestAgainstARealServer(t *testing.T) {
|
|
url := os.Getenv("MESH_TEST_NATS")
|
|
if url == "" {
|
|
t.Skip("MESH_TEST_NATS unset")
|
|
}
|
|
js, err := Dial(url)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer js.Close()
|
|
|
|
t.Run("the mesh's own streams are accepted", func(t *testing.T) {
|
|
if err := AssertMeshStreams(js); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
})
|
|
|
|
t.Run("asserting again changes nothing and fails nothing", func(t *testing.T) {
|
|
if err := AssertMeshStreams(js); err != nil {
|
|
t.Fatalf("the second assertion failed, so the controller cannot restart: %v", err)
|
|
}
|
|
})
|
|
|
|
t.Run("a seat's work queue is accepted beside them", func(t *testing.T) {
|
|
seats := []DeclaredSeat{{Name: "telegram-sender", Accepts: []string{"send"}}}
|
|
for _, s := range SeatStreams(seats) {
|
|
if err := js.EnsureStream(s); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
if c := AllOverlaps(seats); len(c) != 0 {
|
|
t.Fatalf("overlaps the server would refuse: %v", c)
|
|
}
|
|
})
|
|
|
|
t.Run("a module's consumer is accepted and is idempotent", func(t *testing.T) {
|
|
c, ok := ConsumerFor(Principal{Kind: KindModule, Node: "one", Module: "audit",
|
|
Consumes: []string{"shop.order.placed", "billing.invoice.sent"}, PasswordHash: "x"})
|
|
if !ok {
|
|
t.Fatal("no consumer derived")
|
|
}
|
|
if err := js.EnsureConsumer(c); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := js.EnsureConsumer(c); err != nil {
|
|
t.Fatalf("the second assertion failed: %v", err)
|
|
}
|
|
})
|
|
|
|
t.Run("a holder's worker is accepted with its queue group", func(t *testing.T) {
|
|
c, _ := HolderConsumerFor("one", "telegram",
|
|
DeclaredSeat{Name: "telegram-sender", Accepts: []string{"send"}})
|
|
if err := js.EnsureConsumer(c); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
})
|
|
}
|