Files
mesh-controller/internal/link/advisories_test.go
T

42 lines
1.9 KiB
Go

package link
import (
"testing"
)
// **A reader's own consumer gone is not a consumer lost**: every bucket watch and stream read-back
// makes and deletes one, many a minute, and the bus says so each time (found running the controller
// against a real bus: five a tick).
func TestOnlyTheMeshsOwnConsumersAreSaidLost(t *testing.T) {
deleted := func(stream, consumer string) bool {
_, ok := ReadAdvisory("$JS.EVENT.ADVISORY.CONSUMER.DELETED."+stream+"."+consumer,
[]byte(`{"type":"io.nats.jetstream.advisory.v1.consumer_action","stream":"`+stream+`","consumer":"`+consumer+`","action":"delete"}`))
return ok
}
for _, c := range [][2]string{{"EVENTS", "controller"}, {"CONTROL", "controller"}, {"NODES", "anchor"},
{"EVENTS", "anchor_shop"}, {"SEAT_NODE_BUILD_AGENT", "SEAT_NODE_BUILD_AGENT_worker"}} {
if !deleted(c[0], c[1]) {
t.Errorf("%s on %s deleted is not said", c[1], c[0])
}
}
for _, c := range [][2]string{{"KV_mesh-controller_conditions", "381UWW5Y"}, {"EVENTS", "E7vVYoe6"},
{"SEAT_NODE_BUILD_AGENT", "Rcnt6jla"}, {"ASSIGNMENTS", "x"}} {
if deleted(c[0], c[1]) {
t.Errorf("a reader's own consumer %s on %s is said lost", c[1], c[0])
}
}
a, ok := ReadAdvisory("$JS.EVENT.ADVISORY.CONSUMER.MAX_DELIVERIES.EVENTS.anchor_shop",
[]byte(`{"stream":"EVENTS","consumer":"anchor_shop","stream_seq":7,"deliveries":5}`))
if !ok || a.Kind != AdvisoryMaxDeliveries || a.ID != "EVENTS.anchor_shop" ||
a.Said != "how shop on anchor hears what it consumes handed message 7 over 5 times and gave up on it: "+
"what it asked for was not done, and the controller keeps it in DEAD_LETTERS until it is delivered "+
"again or dropped" {
t.Fatalf("%+v", a)
}
}
// **A question the bus refuses is answered as refused at once**, not after its timeout: against a
// server whose only user may not publish where it asks.
// The refusal reaches whoever waits on that subject, and only them.