COND9: the keeper takes a part whose finding stream the bus module made as a source (the moved-stream half stops on a question)
Builds: design 47 G §515, §663 Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ
This commit is contained in:
@@ -0,0 +1,73 @@
|
||||
package main
|
||||
|
||||
// The keeper's sources are the parts whose finding stream the bus module made (design 47 G §515, §663): it takes a source
|
||||
// when it hears the seat mesh-broker's `created` of a part's `finding-events` entry, and at its start reads the entries
|
||||
// the bus module holds by `list`. No part is named here (§7).
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
broker "git.novox.be/novox/mesh-contracts/go/seats/meshbroker"
|
||||
)
|
||||
|
||||
// FindingEventsTemplate is the template of a part's finding stream in its `requires` (design 47 G §529).
|
||||
const FindingEventsTemplate = "finding-events"
|
||||
|
||||
// BrokerCreated takes one `created` of the bus module: a `finding-events` entry makes its module a source, asked once with
|
||||
// `findings-asked` naming it. An entry of another template, or a module already a source, changes nothing.
|
||||
func (k *Keeper) BrokerCreated(body []byte) error {
|
||||
var said broker.Said
|
||||
if err := json.Unmarshal(body, &said); err != nil || said.Module == "" {
|
||||
k.Logf("a created of the bus module is not readable: %v", err)
|
||||
return nil
|
||||
}
|
||||
if said.Template != FindingEventsTemplate {
|
||||
return nil
|
||||
}
|
||||
return k.takeSource(said.Module)
|
||||
}
|
||||
|
||||
// takeSource makes the module a source the bus module made a stream for, and asks it once.
|
||||
func (k *Keeper) takeSource(module string) error {
|
||||
k.mu.Lock()
|
||||
k.init()
|
||||
if k.created == nil {
|
||||
k.created = map[string]bool{}
|
||||
}
|
||||
if k.created[module] {
|
||||
k.mu.Unlock()
|
||||
return nil
|
||||
}
|
||||
k.created[module] = true
|
||||
k.mu.Unlock()
|
||||
if _, err := k.Ask(Module, module); err != nil {
|
||||
return fmt.Errorf("asking %s for its findings: %w", module, err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// ListBound reads the entries the bus module holds at the keeper's start: every module with a `finding-events` entry is a
|
||||
// source. A call that fails is asked again with a growing wait, the bus module being delivered after the keeper at times.
|
||||
func ListBound(k *Keeper, call func(key string, body any) (json.RawMessage, error)) {
|
||||
for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) {
|
||||
raw, err := call("seat:"+broker.Seat+"."+broker.VerbList, broker.ListInput{})
|
||||
var answer broker.ListAnswer
|
||||
if err == nil {
|
||||
err = json.Unmarshal(raw, &answer)
|
||||
}
|
||||
if err == nil {
|
||||
for _, e := range answer.Entries {
|
||||
if e.Template == FindingEventsTemplate {
|
||||
if err := k.takeSource(e.Module); err != nil {
|
||||
k.Logf("%v", err)
|
||||
}
|
||||
}
|
||||
}
|
||||
return
|
||||
}
|
||||
k.Logf("the entries of the bus module are not read yet (%v); asked again in %s", err, wait)
|
||||
time.Sleep(wait)
|
||||
}
|
||||
}
|
||||
@@ -113,6 +113,10 @@ func (k *Keeper) isSource(observer string) bool {
|
||||
k.mu.Lock()
|
||||
defer k.mu.Unlock()
|
||||
k.init()
|
||||
// A part whose finding stream the bus module made is a source (design 47 G §663).
|
||||
if k.created[observer] {
|
||||
return true
|
||||
}
|
||||
// A source the keeper kept in `findings-views` is heard before its first view of the catalog half (design 47 G §388).
|
||||
for _, v := range k.views {
|
||||
if v.observer == observer && v.kept {
|
||||
|
||||
@@ -14,6 +14,7 @@ import (
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
broker "git.novox.be/novox/mesh-contracts/go/seats/meshbroker"
|
||||
engine "git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
|
||||
cc "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient"
|
||||
seat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions"
|
||||
@@ -51,6 +52,9 @@ func (k *Keeper) FindingEvent(e stdio.Envelope) error {
|
||||
if e.Key == engine.Seat+"."+engine.EventRunning {
|
||||
return k.EngineRunning(e.Node, e.Body)
|
||||
}
|
||||
if e.Key == broker.Seat+"."+broker.EventCreated {
|
||||
return k.BrokerCreated(e.Body)
|
||||
}
|
||||
if observer, ok := emitterOf(e.Key, EventFindingsStated); ok {
|
||||
return k.FindingsStated(observer, e.Node, e.Body)
|
||||
}
|
||||
@@ -202,7 +206,7 @@ func (k *Keeper) MachinePresence(body []byte, away bool) error {
|
||||
func hear(k *Keeper) {
|
||||
patterns := []string{"*." + EventFindingWritten, "*." + EventFindingCleared, "*." + EventFindingsStated,
|
||||
watchdogs.Seat + "." + watchdogs.EventMachineAway, watchdogs.Seat + "." + watchdogs.EventMachineBack,
|
||||
engine.Seat + "." + engine.EventRunning}
|
||||
engine.Seat + "." + engine.EventRunning, broker.Seat + "." + broker.EventCreated}
|
||||
for _, pattern := range patterns {
|
||||
go func() {
|
||||
<-stdio.Served()
|
||||
|
||||
@@ -147,6 +147,8 @@ type Keeper struct {
|
||||
placed map[string]bool
|
||||
placedHeld bool
|
||||
stopped map[string]bool
|
||||
// created are the modules whose `finding-events` entry the bus module made (bound.go).
|
||||
created map[string]bool
|
||||
}
|
||||
|
||||
func (k *Keeper) init() {
|
||||
|
||||
@@ -617,6 +617,24 @@ func TestTheKeeperSetsItsViewFromWhatAnObserverStates(t *testing.T) {
|
||||
{"the second branch dropped takes its source away", func() error {
|
||||
return r.k.CatalogDropped("feature")
|
||||
}, []string{"machine.d.silent", "machine.g.silent"}, whyLeftCatalog},
|
||||
// Design 47 G §663: a part whose finding-events entry the bus module made is a source, asked once.
|
||||
{"a created finding-events entry makes its part a source, asked once", func() error {
|
||||
before := len(asked)
|
||||
created := mustJSON(t, map[string]any{"module": "newpart", "template": FindingEventsTemplate, "bus-name": "newpart_finding-events", "local-name": "finding-events"})
|
||||
other := mustJSON(t, map[string]any{"module": "otherpart", "template": "state", "bus-name": "otherpart_x", "local-name": "x"})
|
||||
for _, body := range [][]byte{created, other, created} {
|
||||
if err := r.k.FindingEvent(stdio.Envelope{Key: "mesh-broker.created", Body: body}); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
if len(asked) != before+1 || asked[before]["module"] != "newpart" {
|
||||
t.Errorf("asked %v, want newpart once", asked[before:])
|
||||
}
|
||||
if err := r.k.FindingWritten("otherpart", "ace", mustJSON(t, f("otherpart", "o"))); err != nil {
|
||||
return err
|
||||
}
|
||||
return r.k.FindingWritten("newpart", "ace", mustJSON(t, f("newpart", "w")))
|
||||
}, []string{"machine.d.silent", "machine.g.silent", "machine.w.silent"}, ""},
|
||||
}
|
||||
for _, s := range steps {
|
||||
if err := s.do(); err != nil {
|
||||
|
||||
@@ -130,6 +130,7 @@ func Keep(k *Keeper, v *views.Reader) {
|
||||
return said(k.Attempt(c.Key, c.Value, c.Op != "put"))
|
||||
})
|
||||
hear(k)
|
||||
go func() { <-stdio.Served(); ListBound(k, stdio.Ask) }()
|
||||
watch(k, v, BoundsState, "", func(c stdio.StateChange) error {
|
||||
if c.Op == stdio.OpExpire {
|
||||
return said(k.BoundPassed(c.Key))
|
||||
|
||||
@@ -37,9 +37,9 @@ func TestTheManifestSaysWhatTheCodeDoes(t *testing.T) {
|
||||
// the observers' findings are heard as events and no `findings` is read (design 47 G §383).
|
||||
reads := []string{HavesRead, AttemptsRead}
|
||||
if !slices.Equal(m.Invokes, []string{WantsTool, AssignmentsTool, "seat:mesh-hand-acts.hand-act", "seat:mesh-wants.revision",
|
||||
"seat:mesh-catalog.registrations", "seat:mesh-catalog.revision"}) ||
|
||||
"seat:mesh-catalog.registrations", "seat:mesh-catalog.revision", "seat:mesh-broker.list"}) ||
|
||||
!slices.Equal(m.Consumes, []string{"mesh-delivery-assigner.views.wants", "mesh-delivery-assigner.views.assignments",
|
||||
"*.finding-written", "*.finding-cleared", "*.findings-stated", "mesh-watchdogs.machine-away", "mesh-watchdogs.machine-back", "mesh-wants.assignment-revision", "mesh-catalog.registration-changed", "mesh-catalog.catalog-revision", "node-engine.running"}) {
|
||||
"*.finding-written", "*.finding-cleared", "*.findings-stated", "mesh-watchdogs.machine-away", "mesh-watchdogs.machine-back", "mesh-wants.assignment-revision", "mesh-catalog.registration-changed", "mesh-catalog.catalog-revision", "mesh-broker.created", "node-engine.running"}) {
|
||||
t.Errorf("invokes %v, consumes %v", m.Invokes, m.Consumes)
|
||||
}
|
||||
for _, r := range m.Reads {
|
||||
|
||||
@@ -5,6 +5,7 @@ go 1.26.0
|
||||
require (
|
||||
git.novox.be/novox/mesh-contracts/go/clients/meshmoduleassignerclient v0.0.0-00010101000000-000000000000
|
||||
git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient v0.0.0-00010101000000-000000000000
|
||||
git.novox.be/novox/mesh-contracts/go/seats/meshbroker v0.0.0-00010101000000-000000000000
|
||||
git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner v0.0.0-00010101000000-000000000000
|
||||
git.novox.be/novox/mesh-contracts/go/seats/nodeengine v0.0.0-00010101000000-000000000000
|
||||
git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient v0.0.0-00010101000000-000000000000
|
||||
|
||||
@@ -23,7 +23,8 @@
|
||||
"seat:mesh-hand-acts.hand-act",
|
||||
"seat:mesh-wants.revision",
|
||||
"seat:mesh-catalog.registrations",
|
||||
"seat:mesh-catalog.revision"
|
||||
"seat:mesh-catalog.revision",
|
||||
"seat:mesh-broker.list"
|
||||
],
|
||||
"consumes": [
|
||||
"mesh-delivery-assigner.views.wants",
|
||||
@@ -36,6 +37,7 @@
|
||||
"mesh-wants.assignment-revision",
|
||||
"mesh-catalog.registration-changed",
|
||||
"mesh-catalog.catalog-revision",
|
||||
"mesh-broker.created",
|
||||
"node-engine.running"
|
||||
],
|
||||
"state": [
|
||||
|
||||
Reference in New Issue
Block a user