novox/mesh-delivery
19
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

DLV-G1, DLV-W1: read the bus module's events by the mesh-broker contract's field names, on its own stream or the seat's subject

Browse Source 
The seat's definition (mesh-contracts meshbroker) names local-name and storage-type and puts
the events on mesh.seat.mesh-broker.event.<event>; design 25 § 12.3 keeps them on the bus
module's own stream. Both are read until that is ruled.

Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig
This commit is contained in:
jschoubben
2026-11-04 12:34:50 +01:00
parent fc150986b7
commit 745c30e068
5 changed files with 47 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/consumes.go
+9 -2
View File
@@ -8,6 +8,7 @@ import (
"github.com/nats-io/nats.go/jetstream"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-sdk/go/membership"
"git.novox.be/novox/mesh-sdk/go/process"
@@ -39,13 +40,13 @@ func consumed(w *watch.Watchman, subject string, body []byte, logf func(string,
return
}
w.Accepted(a)
case strings.HasPrefix(subject, "mesh.stream.") && strings.HasSuffix(subject, ".event."+eventCreated):
case brokerSubject(subject) && strings.HasSuffix(subject, ".event."+eventCreated):
if s, err := watch.ReadBrokerSaid(body); err != nil {
logf("%v; left", err)
} else {
w.BrokerCreated(s)
}
case strings.HasPrefix(subject, "mesh.stream.") && strings.HasSuffix(subject, ".event."+eventRefused):
case brokerSubject(subject) && strings.HasSuffix(subject, ".event."+eventRefused):
if s, err := watch.ReadBrokerSaid(body); err != nil {
logf("%v; left", err)
} else {
@@ -108,3 +109,9 @@ func hearConsumed(ctx context.Context, js jetstream.JetStream, cred process.Acco
}
}()
}
// brokerSubject is a subject the bus module's created or refused arrives on: its own stream's, or the seat mesh-broker's
// (where the seat's definition puts it; only its holder publishes there).
func brokerSubject(subject string) bool {
return strings.HasPrefix(subject, "mesh.stream.") || strings.HasPrefix(subject, "mesh.seat."+catalogue.BrokerSeat+".event.")
}
modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main_test.go
+1 -1
View File
@@ -61,7 +61,7 @@ func TestRunNeedsItsCredential(t *testing.T) { // 1135/AC6
t.Fatalf("an accepted manifest no longer asking for l left %v", out.findings)
}
consumed(w, "mesh.stream.nats.event.refused", []byte(`{"module":"m","local":"l"}`), say)
consumed(w, "mesh.stream.nats.event.created", []byte(`{"module":"m","local":"l"}`), say)
consumed(w, "mesh.seat.mesh-broker.event.created", []byte(`{"module":"m","local-name":"l"}`), say)
if len(out.findings) != 0 {
t.Fatalf("created left %v", out.findings)
}
modules/mesh-delivery-watchman/internal/watch/busstate.go
+7 -2
View File
@@ -35,8 +35,10 @@ const (
// create was refused. It stands until the seat mesh-broker's definition is generated in mesh-contracts.
type BrokerSaid struct {
Module string `json:"module"`
Local string `json:"local"`
Why string `json:"why,omitempty"`
// Local is the entry: the contract's `local-name` (mesh-contracts meshbroker.Said), or `local` where a body names it so.
Local string `json:"local,omitempty"`
LocalName string `json:"local-name,omitempty"`
Why string `json:"why,omitempty"`
}
// busStateID is the signal's and the findings' id of one entry: `<module>.<local>`.
@@ -145,6 +147,9 @@ func ReadBrokerSaid(body []byte) (BrokerSaid, error) {
if err := json.Unmarshal(body, &s); err != nil {
return BrokerSaid{}, fmt.Errorf("the bus module's word cannot be read: %w", err)
}
if s.Local == "" {
s.Local = s.LocalName
}
if s.Module == "" || s.Local == "" {
return BrokerSaid{}, fmt.Errorf("the bus module's word names no module or no entry")
}
modules/mesh-grants/grants_test.go
+2 -1
View File
@@ -2261,7 +2261,8 @@ func TestGrants(t *testing.T) {
if got := bindingsOn("b.mod-x"); len(got) != 1 || got[0].Bucket != "mod-x_records-2" {
t.Errorf("after recreated: %+v", got)
}
if !hear("mesh.stream.nats.event.dropped", membership.BusBinding{Local: "records"}) {
// The seat's own subject, and the contract's field names (mesh-contracts meshbroker.Said), are read alike.
if changed, err := p.HearBroker("mesh.seat.mesh-broker.event.dropped", []byte(`{"module":"mod-x","local-name":"records"}`), p.Brokers()); err != nil || !changed {
t.Fatal("dropped changed nothing")
}
p.Compose()
modules/mesh-grants/internal/grants/bindings.go
+28 -10
View File
@@ -6,6 +6,7 @@ import (
"sort"
"strings"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"git.novox.be/novox/mesh-sdk/go/membership"
"mesh-grants/internal/buscompose"
)
@@ -39,11 +40,15 @@ func BoundKey(module string) string { return BoundPrefix + module }
// It stands until the seat mesh-broker's definition is generated in mesh-contracts (ADR 0613 §1, §5; design 26): this
// part then reads the generated type, with the same fields.
type BrokerEvent struct {
V int `json:"v"`
Module string `json:"module"`
Local string `json:"local"`
Storage string `json:"storage"`
Template string `json:"template,omitempty"`
V int `json:"v,omitempty"`
Module string `json:"module"`
// LocalName and StorageType are the seat's contract's names (mesh-contracts meshbroker.Said); Local and Storage the
// same read where an older body names them so.
LocalName string `json:"local-name,omitempty"`
Local string `json:"local,omitempty"`
StorageType string `json:"storage-type,omitempty"`
Storage string `json:"storage,omitempty"`
Template string `json:"template,omitempty"`
// Binding is the entry's names on the bus: whole on `created` and `recreated`; on `dropped` only its local name counts.
Binding membership.BusBinding `json:"binding"`
}
@@ -61,6 +66,12 @@ type Bound struct {
func brokerEvent(event string, brokers map[string]bool) (string, bool) {
emitter, name := "", ""
switch {
case strings.HasPrefix(event, "mesh.seat."+catalogue.BrokerSeat+".event."):
// The seat's own subject: only its holder publishes there, so the emitter is the holder.
name = strings.TrimPrefix(event, "mesh.seat."+catalogue.BrokerSeat+".event.")
for b := range brokers {
emitter = b
}
case buscompose.OnAModuleStream(event):
// mesh.stream.<module>.event.<event>
rest := strings.TrimPrefix(event, "mesh.stream.")
@@ -93,10 +104,8 @@ func (p *Part) HearBroker(event string, body []byte, brokers map[string]bool) (b
if err := json.Unmarshal(body, &e); err != nil {
return false, fmt.Errorf("the bus module's %s does not read: %w", name, err)
}
local := e.Local
if local == "" {
local = e.Binding.Local
}
local := firstOf(e.LocalName, e.Local, e.Binding.Local)
storage := firstOf(e.StorageType, e.Storage)
if e.Module == "" || local == "" || strings.Contains(e.Module, ".") {
return false, fmt.Errorf("the bus module's %s names no module or no entry (%q, %q)", name, e.Module, local)
}
@@ -115,7 +124,7 @@ func (p *Part) HearBroker(event string, body []byte, brokers map[string]bool) (b
bind := e.Binding
bind.Local = local
if bind.Storage == "" {
bind.Storage = e.Storage
bind.Storage = storage
}
if bind.Template == "" {
bind.Template = e.Template
@@ -245,3 +254,12 @@ func bindingStreams(b membership.BusBinding) []string {
}
return out
}
func firstOf(values ...string) string {
for _, v := range values {
if v != "" {
return v
}
}
return ""
}