jochen
743e6478a7
mesh/merge-gate head 743e6478a76c merged into trunk c2ec0a601acb — pass: builds build-agent, mesh-controller, route-proxy → ace, g14, novox, shanks; no bus…
mesh/repo-check head 743e6478a76c merged into trunk c2ec0a601acb — pass: its merge-check.sh passed
mesh/delivery delivered
69 lines
2.5 KiB
Go
69 lines
2.5 KiB
Go
package broker
|
|
|
|
import (
|
|
"slices"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/novox/mesh-controller/internal/testbus"
|
|
)
|
|
|
|
// On a real bus: a module's stream is made beside EVENTS without a clash, keeps what the module publishes on
|
|
// its own subjects, a granted consumer hears only its part, a narrowed grant narrows it, and a withdrawn one
|
|
// is removed (novox/hq ADR 0363).
|
|
//
|
|
// go test ./internal/broker/ -run TestAModuleStreamOnARealBus
|
|
func TestAModuleStreamOnARealBus(t *testing.T) {
|
|
js, err := Dial(testbus.URL(t))
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer js.Close()
|
|
for _, s := range []Stream{MeshStreams()[3], mailStream.Stream()} {
|
|
_ = js.js.DeleteStream(s.Name)
|
|
if err := js.EnsureStream(s); err != nil {
|
|
t.Fatalf("%s: %v", s.Name, err)
|
|
}
|
|
defer func(name string) { _ = js.js.DeleteStream(name) }(s.Name)
|
|
}
|
|
for _, account := range []string{"mesh", "work"} {
|
|
if _, err := js.js.Publish(ModuleStreamSubject("mesh-mail", "mail."+account+".message.added"), []byte("{}")); err != nil {
|
|
t.Fatalf("publishing %s's mail: %v", account, err)
|
|
}
|
|
}
|
|
if info, _ := js.js.StreamInfo(EventsStream); info.State.Msgs != 0 {
|
|
t.Fatalf("EVENTS kept %d of the module stream's events", info.State.Msgs)
|
|
}
|
|
|
|
reader := Principal{Kind: KindModule, Node: "ace", Module: "reader", Hears: []Heard{mailHeard}}
|
|
c := heardConsumer(reader, mailHeard)
|
|
if err := js.EnsureConsumer(c); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
time.Sleep(50 * time.Millisecond)
|
|
if info, _ := js.js.ConsumerInfo("MAIL", c.Name); info.NumPending != 1 {
|
|
t.Fatalf("the reader granted `mesh` has %d pending, want 1", info.NumPending)
|
|
}
|
|
narrowed := c
|
|
narrowed.Filters = []string{"mesh.stream.mesh-mail.event.mail.mesh.message.changed"}
|
|
if err := EnsureHeard(js, narrowed, true); err != nil {
|
|
t.Fatalf("narrowing the grant: %v", err)
|
|
}
|
|
if info, _ := js.js.ConsumerInfo("MAIL", c.Name); info.Config.FilterSubject != narrowed.Filters[0] || len(info.Config.FilterSubjects) != 0 {
|
|
t.Fatalf("narrowed to %q %v, want only %s", info.Config.FilterSubject, info.Config.FilterSubjects, narrowed.Filters[0])
|
|
}
|
|
if info, _ := js.js.ConsumerInfo("MAIL", c.Name); info.NumPending != 0 {
|
|
t.Fatalf("after narrowing, %d of the reader's former grant are still pending", info.NumPending)
|
|
}
|
|
removed, err := PruneHeard(js, []ModuleStream{mailStream}, nil)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if !slices.Equal(removed, []string{"MAIL/ace_reader"}) {
|
|
t.Fatalf("removed %v", removed)
|
|
}
|
|
if names, _ := js.ConsumerNames("MAIL"); len(names) != 0 {
|
|
t.Fatalf("still on MAIL: %v", names)
|
|
}
|
|
}
|