diff --git a/go/catalogue/streams.go b/go/catalogue/streams.go index b9f9565c..5377c9a4 100644 --- a/go/catalogue/streams.go +++ b/go/catalogue/streams.go @@ -287,14 +287,45 @@ type HeardPart struct { // stream, granted or not, and the `consumes` left for EVENTS — those naming something the provider emits // there. // -// Only an entry naming its emitter literally reaches a module's stream: a wildcard emitter, `**` among them, -// is EVENTS' alone, so nothing that hears every event hears a restricted stream by saying so. +// Only an entry naming its emitter literally reaches a restricted stream: a wildcard emitter, `**` among them, +// never does, so nothing that hears every event hears a restricted stream by saying so. A `*` emitter before one +// event reaches each declaring module's unrestricted stream (design 47 G §386). func Hearing(consumer Manifest, shelf map[string]Manifest, machine string, values func(provider, setting string) []any) (parts []HeardPart, touches bool, onEvents []string) { byStream := map[string]*HeardPart{} var order []string for _, c := range consumer.Consumes { emitter, rest, named := strings.Cut(c, ".") + if named && emitter == OneName && rest != "" && !strings.ContainsAny(rest, "*>") && rest != TheRest { + // A `*` emitter (design 47 G §386): the event from every module that declares it on a stream, never a + // restricted one; the entry stays EVENTS' too, for the modules that say it there. + onEvents = append(onEvents, c) + names := make([]string, 0, len(shelf)) + for name := range shelf { + names = append(names, name) + } + slices.Sort(names) + for _, name := range names { + hearing, _ := shelf[name].StreamsHearing(rest) + for _, s := range hearing { + if s.Grant != nil { + continue + } + touches = true + id := name + "/" + s.BusName() + if byStream[id] == nil { + byStream[id] = &HeardPart{Emitter: name, Stream: s} + order = append(order, id) + } + for _, p := range s.Events { + if x, ok := narrow(rest, p); ok { + byStream[id].Patterns = append(byStream[id].Patterns, x) + } + } + } + } + continue + } provider, known := shelf[emitter] if !named || emitter == OneName || emitter == TheRest || !known || len(provider.Streams) == 0 { onEvents = append(onEvents, c) diff --git a/go/catalogue/streams_test.go b/go/catalogue/streams_test.go index 6a5cd52b..03466509 100644 --- a/go/catalogue/streams_test.go +++ b/go/catalogue/streams_test.go @@ -152,3 +152,26 @@ func TestNarrowingTwoPatterns(t *testing.T) { t.Fatalf("uncovered: %v", got) } } + +// Design 47 G §386: `*.` reaches the unrestricted stream of each module that declares the event, and never a +// restricted one; a wildcard before anything wider still reaches none. +func TestAStarEmitterHearsEachDeclaringModulesUnrestrictedStream(t *testing.T) { + mail, err := decodeForTest([]byte(mailManifest(t, mailStream, `"emits":["account.added"],`))) + if err != nil { + t.Fatal(err) + } + a := Manifest{Module: "a", Streams: []StreamDeclaration{{Name: "a-findings", Events: []string{"finding-written"}}}} + b := Manifest{Module: "b", Streams: []StreamDeclaration{{Name: "b-findings", Events: []string{"finding-written", "finding-cleared"}}}} + other := Manifest{Module: "c", Streams: []StreamDeclaration{{Name: "c-stuff", Events: []string{"other"}}}} + shelf := map[string]Manifest{"a": a, "b": b, "c": other, "mesh-mail": mail} + keeper := Manifest{Module: "keeper", Consumes: []string{"*.finding-written"}} + parts, touches, onEvents := Hearing(keeper, shelf, "ace", func(string, string) []any { return nil }) + if !touches || len(parts) != 2 || parts[0].Emitter != "a" || parts[1].Emitter != "b" || + !slices.Equal(parts[1].Patterns, []string{"finding-written"}) || !slices.Equal(onEvents, []string{"*.finding-written"}) { + t.Fatalf("heard %+v (touches %v, on EVENTS %v)", parts, touches, onEvents) + } + restricted := Manifest{Module: "audit", Consumes: []string{"*.mail.mesh.message.added", "*.>"}} + if parts, _, _ := Hearing(restricted, shelf, "ace", func(string, string) []any { return nil }); len(parts) != 0 { + t.Fatalf("a wildcard reached %+v", parts) + } +}