Seat list reader: the seat definitions read from the bus, followed on seat-definition-changed (ADR 0635 §1, §3; issue 1966)
Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig
This commit is contained in:
@@ -0,0 +1,57 @@
|
||||
package catalogue
|
||||
|
||||
import (
|
||||
"sort"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/seatdef"
|
||||
)
|
||||
|
||||
// SeatFromDefinition is a seat in this package's types from its definition (novox/hq ADR 0608 §1, ADR 0635 §1): what
|
||||
// the seat list once held, now read from the definition the indexer keeps on the bus. The definition is resolved with
|
||||
// its parts (seatdef.Resolve) as the indexer keeps it. A verb's input and answer have their named types in place, and a
|
||||
// contribution kind given as data gets its checks (withShapeCode), which are code and never a definition's.
|
||||
func SeatFromDefinition(d seatdef.Definition) Seat {
|
||||
s := Seat{Name: d.Seat, Scope: d.Scope, Replicated: d.Replicated, Delivers: d.Delivers,
|
||||
Accepts: seatdef.MessageNames(d.Accepts), Emits: seatdef.MessageNames(d.Events), PulledByKind: d.PulledByKind,
|
||||
Yields: d.Yields, HeldByTheEngine: d.HeldByTheEngine, Decision: d.Decision}
|
||||
for _, v := range d.Verbs {
|
||||
s.Serves = append(s.Serves, Verb{Name: v.Name, Description: v.Description,
|
||||
Input: seatdef.ResolveRefs(v.Input, d.Defs), Output: seatdef.ResolveRefs(v.Answer, d.Defs),
|
||||
Optional: v.Optional, Replaces: v.Replaces})
|
||||
}
|
||||
for _, r := range d.Receives {
|
||||
rv := Receivable{Kind: r.Kind, Comment: r.Comment, Dirs: r.Dirs, Offered: r.Offered}
|
||||
if r.Shape != nil {
|
||||
sh := &Shape{Order: r.Shape.Order, Examples: r.Shape.Examples, Text: r.Shape.Text}
|
||||
for _, f := range r.Shape.Places {
|
||||
sh.Places = append(sh.Places, PlaceField{Field: f.Field, Values: f.Values})
|
||||
}
|
||||
withShapeCode(d.Seat, r.Kind, sh)
|
||||
rv.Shape = sh
|
||||
}
|
||||
s.Receives = append(s.Receives, rv)
|
||||
}
|
||||
for _, p := range d.Paths {
|
||||
s.Paths = append(s.Paths, SeatPath{Path: p.Path, StepsBackFor: p.StepsBackFor})
|
||||
}
|
||||
return s
|
||||
}
|
||||
|
||||
// SeatsFromDefinitions are the seats of a set of definitions, in reading order: the mesh's own, then a node's, each by
|
||||
// name. A seat its holder declares itself (HolderDeclares) is not in the list, as the compiled list left it out.
|
||||
func SeatsFromDefinitions(defs []seatdef.Definition) []Seat {
|
||||
out := make([]Seat, 0, len(defs))
|
||||
for _, d := range defs {
|
||||
if d.HolderDeclares {
|
||||
continue
|
||||
}
|
||||
out = append(out, SeatFromDefinition(d))
|
||||
}
|
||||
sort.SliceStable(out, func(i, j int) bool {
|
||||
if (out[i].Scope == seatdef.ScopeMesh) != (out[j].Scope == seatdef.ScopeMesh) {
|
||||
return out[i].Scope == seatdef.ScopeMesh
|
||||
}
|
||||
return out[i].Name < out[j].Name
|
||||
})
|
||||
return out
|
||||
}
|
||||
@@ -0,0 +1,40 @@
|
||||
package seatdef
|
||||
|
||||
import "strings"
|
||||
|
||||
// ResolveRefs is a schema with every `{"$ref": "#/defs/<name>"}` replaced by the definition's named type, as the seat
|
||||
// list's readers need it (a verb's input and answer are one schema each, not a schema and its types). A reference to a
|
||||
// name the definition does not define stays as it is; nesting deeper than 32 levels stops there.
|
||||
func ResolveRefs(schema map[string]any, defs map[string]map[string]any) map[string]any {
|
||||
if schema == nil {
|
||||
return nil
|
||||
}
|
||||
m, _ := resolveValue(schema, defs, 0).(map[string]any)
|
||||
return m
|
||||
}
|
||||
|
||||
func resolveValue(v any, defs map[string]map[string]any, depth int) any {
|
||||
if depth > 32 {
|
||||
return v
|
||||
}
|
||||
switch x := v.(type) {
|
||||
case map[string]any:
|
||||
if r, ok := x["$ref"].(string); ok {
|
||||
if d, ok := defs[strings.TrimPrefix(r, "#/defs/")]; ok {
|
||||
return resolveValue(d, defs, depth+1)
|
||||
}
|
||||
}
|
||||
out := make(map[string]any, len(x))
|
||||
for k, e := range x {
|
||||
out[k] = resolveValue(e, defs, depth+1)
|
||||
}
|
||||
return out
|
||||
case []any:
|
||||
out := make([]any, len(x))
|
||||
for i, e := range x {
|
||||
out[i] = resolveValue(e, defs, depth+1)
|
||||
}
|
||||
return out
|
||||
}
|
||||
return v
|
||||
}
|
||||
@@ -0,0 +1,31 @@
|
||||
package seatdef
|
||||
|
||||
import (
|
||||
"reflect"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// A verb's schema has its named types in place (the seat list's readers take one schema, not a schema and its types).
|
||||
func TestResolveRefsPutsNamedTypesInPlace(t *testing.T) {
|
||||
defs := map[string]map[string]any{
|
||||
"id": {"type": "string"},
|
||||
"row": {"type": "object", "properties": map[string]any{"id": map[string]any{"$ref": "#/defs/id"}}},
|
||||
}
|
||||
cases := []struct {
|
||||
name string
|
||||
schema, want map[string]any
|
||||
}{
|
||||
{"nothing", nil, nil},
|
||||
{"a type nested in a type", map[string]any{"type": "array", "items": map[string]any{"$ref": "#/defs/row"}},
|
||||
map[string]any{"type": "array", "items": map[string]any{"type": "object",
|
||||
"properties": map[string]any{"id": map[string]any{"type": "string"}}}}},
|
||||
{"a name it does not define stays", map[string]any{"$ref": "#/defs/gone"}, map[string]any{"$ref": "#/defs/gone"}},
|
||||
{"a list of schemas", map[string]any{"oneOf": []any{map[string]any{"$ref": "#/defs/id"}}},
|
||||
map[string]any{"oneOf": []any{map[string]any{"type": "string"}}}},
|
||||
}
|
||||
for _, c := range cases {
|
||||
if got := ResolveRefs(c.schema, defs); !reflect.DeepEqual(got, c.want) {
|
||||
t.Errorf("%s: %v, want %v", c.name, got, c.want)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,227 @@
|
||||
// Package seatlist is the runtime's reader of the seat list (novox/hq ADR 0635 §1, §3): the seat definitions the
|
||||
// indexer keeps on the bus, asked once through the owner's tool and followed on its event `seat-definition-changed`.
|
||||
// The manifest validator, the node-engine, the tool runner, claude-code, the grants part and every other caller of
|
||||
// `catalogue` get their seats here and give them to `catalogue` as an input; mesh-sdk compiles no seat list and imports
|
||||
// no contracts module.
|
||||
//
|
||||
// **What it asks and hears.** The owner's tool Tool, with no seat for every trunk definition (the answer's shape is
|
||||
// Listing) and with one seat for that seat's (the answer is the owner's own, read as Answer); the owner's event
|
||||
// ChangedEvent, whose body names the seat and its source (Changed). Only trunk definitions are held: a pull request's are
|
||||
// the indexer's to judge. Nothing polls and nothing waits on a timer: a reader that could not ask says so in its
|
||||
// Status and reads again on the next event or on Reload, which the host module offers as a tool (steering it).
|
||||
//
|
||||
// **What it names.** The tool's and the event's names, which ADR 0635 §1 gives the owner. It names no module and no
|
||||
// seat: the owner's module is the caller's (Config.Owner), the one its manifest's `invokes` lists.
|
||||
package seatlist
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"strings"
|
||||
"sync"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
"git.novox.be/novox/mesh-sdk/go/seatdef"
|
||||
)
|
||||
|
||||
const (
|
||||
// Tool is the owner's tool that answers a seat definition.
|
||||
Tool = "indexer_seat_definition"
|
||||
// ChangedEvent is the owner's event that says a definition moved.
|
||||
ChangedEvent = "seat-definition-changed"
|
||||
// SourceTrunk is the only source this reader holds.
|
||||
SourceTrunk = "trunk"
|
||||
)
|
||||
|
||||
// Bus is what the reader needs of its host: a call of a tool by its key (`<module>.<tool>`) and a subscription to an
|
||||
// event pattern. StdioBus is the one a bundle has; the node-engine and the tool runner give theirs.
|
||||
type Bus interface {
|
||||
Ask(key string, body any) (json.RawMessage, error)
|
||||
Subscribe(pattern string, handle func(key string, body json.RawMessage) error) error
|
||||
}
|
||||
|
||||
// Entry is one definition as the owner answers it: the seat, its source, the digest of the file, and the definition
|
||||
// resolved with the parts it includes. It is the owner's SeatDefinitionEntry, read here by the fields it needs.
|
||||
type Entry struct {
|
||||
Seat string `json:"seat"`
|
||||
Source string `json:"source"`
|
||||
Digest string `json:"digest,omitempty"`
|
||||
Definition json.RawMessage `json:"definition,omitempty"`
|
||||
Unparsed string `json:"unparsed,omitempty"`
|
||||
}
|
||||
|
||||
// Listing is the owner's answer to the tool with no seat: every trunk definition and the revision of its state.
|
||||
type Listing struct {
|
||||
Revision uint64 `json:"revision"`
|
||||
Definitions []Entry `json:"definitions"`
|
||||
}
|
||||
|
||||
// Answer is the owner's answer to the tool with one seat.
|
||||
type Answer struct {
|
||||
Seat string `json:"seat"`
|
||||
Found bool `json:"found"`
|
||||
Entry *Entry `json:"entry,omitempty"`
|
||||
Revision uint64 `json:"revision"`
|
||||
}
|
||||
|
||||
// Changed is the event's body.
|
||||
type Changed struct {
|
||||
Seat string `json:"seat"`
|
||||
Source string `json:"source"`
|
||||
Digest string `json:"digest"`
|
||||
}
|
||||
|
||||
// Config is what a host gives Start.
|
||||
type Config struct {
|
||||
// Owner is the module that keeps the definitions, as the host's manifest `invokes` names it.
|
||||
Owner string
|
||||
// Bus is the host's way to the owner; StdioBus() for a bundle.
|
||||
Bus Bus
|
||||
// Bootstrap are the definitions known before the owner is heard (ADR 0635 §3); Bootstrapped() is the foundation's.
|
||||
Bootstrap []seatdef.Definition
|
||||
// OnChange, when set, is called after the held set changed, with the seats in reading order. It runs on the
|
||||
// goroutine that heard the change: it returns quickly.
|
||||
OnChange func([]catalogue.Seat)
|
||||
}
|
||||
|
||||
// Reader holds the seat list.
|
||||
type Reader struct {
|
||||
cfg Config
|
||||
|
||||
mu sync.Mutex
|
||||
defs map[string]seatdef.Definition // by seat
|
||||
boot map[string]bool // seats held from the bootstrap, not yet from the owner
|
||||
}
|
||||
|
||||
// Start returns a reader holding the bootstrap, asks the owner for every definition and follows its event. A first
|
||||
// ask the owner did not answer is no failure: the reader holds the bootstrap and reads again on the owner's next
|
||||
// event or on Reload. Start returns the subscription's error alone, which the host may ask again by calling Start again
|
||||
// on a new reader.
|
||||
func Start(cfg Config) (*Reader, error) {
|
||||
if cfg.Bus == nil {
|
||||
return nil, errors.New("seatlist: no bus to read the seat list through")
|
||||
}
|
||||
if cfg.Owner == "" || strings.ContainsAny(cfg.Owner, ". \t") {
|
||||
return nil, fmt.Errorf("seatlist: the owner %q is not a module name", cfg.Owner)
|
||||
}
|
||||
r := &Reader{cfg: cfg, defs: map[string]seatdef.Definition{}, boot: map[string]bool{}}
|
||||
for _, d := range cfg.Bootstrap {
|
||||
r.defs[d.Seat] = d
|
||||
r.boot[d.Seat] = true
|
||||
}
|
||||
if err := cfg.Bus.Subscribe("*."+ChangedEvent, r.heard); err != nil {
|
||||
return r, fmt.Errorf("seatlist: following %s: %w", ChangedEvent, err)
|
||||
}
|
||||
_ = r.Reload()
|
||||
return r, nil
|
||||
}
|
||||
|
||||
// Seats are the held seats in reading order, as catalogue takes them.
|
||||
func (r *Reader) Seats() []catalogue.Seat {
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
return r.seatsLocked()
|
||||
}
|
||||
|
||||
// Definitions are the held definitions, by seat name.
|
||||
func (r *Reader) Definitions() map[string]seatdef.Definition {
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
out := make(map[string]seatdef.Definition, len(r.defs))
|
||||
for k, v := range r.defs {
|
||||
out[k] = v
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func (r *Reader) seatsLocked() []catalogue.Seat {
|
||||
ds := make([]seatdef.Definition, 0, len(r.defs))
|
||||
for _, d := range r.defs {
|
||||
ds = append(ds, d)
|
||||
}
|
||||
return catalogue.SeatsFromDefinitions(ds)
|
||||
}
|
||||
|
||||
// Reload asks the owner for every trunk definition and holds what it answers, a seat the owner holds replacing the
|
||||
// bootstrap's. An answer that cannot be read is returned and changes nothing.
|
||||
func (r *Reader) Reload() error {
|
||||
raw, err := r.cfg.Bus.Ask(r.cfg.Owner+"."+Tool, map[string]any{})
|
||||
if err != nil {
|
||||
return fmt.Errorf("seatlist: asking %s.%s: %w", r.cfg.Owner, Tool, err)
|
||||
}
|
||||
var l Listing
|
||||
if err := json.Unmarshal(raw, &l); err != nil {
|
||||
return fmt.Errorf("seatlist: %s.%s answered what is not a listing: %w", r.cfg.Owner, Tool, err)
|
||||
}
|
||||
parsed := map[string]seatdef.Definition{}
|
||||
for _, e := range l.Definitions {
|
||||
if e.Source != SourceTrunk || e.Unparsed != "" || len(e.Definition) == 0 {
|
||||
continue
|
||||
}
|
||||
d, err := seatdef.Parse(e.Definition)
|
||||
if err != nil {
|
||||
return fmt.Errorf("seatlist: the definition of %s does not parse: %w", e.Seat, err)
|
||||
}
|
||||
parsed[d.Seat] = d
|
||||
}
|
||||
r.mu.Lock()
|
||||
for seat := range r.defs {
|
||||
if !r.boot[seat] {
|
||||
delete(r.defs, seat)
|
||||
}
|
||||
}
|
||||
for seat, d := range parsed {
|
||||
r.defs[seat] = d
|
||||
delete(r.boot, seat)
|
||||
}
|
||||
r.mu.Unlock()
|
||||
r.changed()
|
||||
return nil
|
||||
}
|
||||
|
||||
// heard takes the owner's event: a trunk definition moved, so that seat is asked again. An error is returned so the
|
||||
// bus offers the event again.
|
||||
func (r *Reader) heard(_ string, body json.RawMessage) error {
|
||||
var c Changed
|
||||
if err := json.Unmarshal(body, &c); err != nil || c.Seat == "" {
|
||||
return nil
|
||||
}
|
||||
if c.Source != SourceTrunk {
|
||||
return nil
|
||||
}
|
||||
raw, err := r.cfg.Bus.Ask(r.cfg.Owner+"."+Tool, map[string]any{"seat": c.Seat})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
var a Answer
|
||||
if err := json.Unmarshal(raw, &a); err != nil {
|
||||
return err
|
||||
}
|
||||
r.mu.Lock()
|
||||
switch {
|
||||
case !a.Found || a.Entry == nil || a.Entry.Source != SourceTrunk:
|
||||
if !r.boot[c.Seat] {
|
||||
delete(r.defs, c.Seat)
|
||||
}
|
||||
case a.Entry.Unparsed != "" || len(a.Entry.Definition) == 0:
|
||||
// the definition that stands stays: a trunk's that does not parse is never written over it
|
||||
default:
|
||||
d, err := seatdef.Parse(a.Entry.Definition)
|
||||
if err != nil {
|
||||
r.mu.Unlock()
|
||||
return fmt.Errorf("seatlist: the definition of %s does not parse: %w", c.Seat, err)
|
||||
}
|
||||
r.defs[c.Seat] = d
|
||||
delete(r.boot, c.Seat)
|
||||
}
|
||||
r.mu.Unlock()
|
||||
r.changed()
|
||||
return nil
|
||||
}
|
||||
|
||||
func (r *Reader) changed() {
|
||||
if r.cfg.OnChange != nil {
|
||||
r.cfg.OnChange(r.Seats())
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,129 @@
|
||||
package seatlist
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
"git.novox.be/novox/mesh-sdk/go/seatdef"
|
||||
)
|
||||
|
||||
// fakeOwner is the definitions' owner: the tool answers from its table and the event is delivered by the test.
|
||||
type fakeOwner struct {
|
||||
table map[string]string // seat -> definition JSON, trunk
|
||||
down bool
|
||||
asked []string
|
||||
handler func(string, json.RawMessage) error
|
||||
pattern string
|
||||
}
|
||||
|
||||
func (f *fakeOwner) Ask(key string, body any) (json.RawMessage, error) {
|
||||
f.asked = append(f.asked, key)
|
||||
if f.down {
|
||||
return nil, errors.New("no answer")
|
||||
}
|
||||
args, _ := body.(map[string]any)
|
||||
if seat, _ := args["seat"].(string); seat != "" {
|
||||
raw, ok := f.table[seat]
|
||||
if !ok {
|
||||
return json.Marshal(Answer{Seat: seat})
|
||||
}
|
||||
return json.Marshal(Answer{Seat: seat, Found: true, Entry: &Entry{Seat: seat, Source: SourceTrunk, Definition: json.RawMessage(raw)}})
|
||||
}
|
||||
l := Listing{Revision: 7}
|
||||
for seat, raw := range f.table {
|
||||
l.Definitions = append(l.Definitions, Entry{Seat: seat, Source: SourceTrunk, Definition: json.RawMessage(raw)})
|
||||
}
|
||||
return json.Marshal(l)
|
||||
}
|
||||
|
||||
func (f *fakeOwner) Subscribe(pattern string, h func(string, json.RawMessage) error) error {
|
||||
f.pattern, f.handler = pattern, h
|
||||
return nil
|
||||
}
|
||||
|
||||
func (f *fakeOwner) say(seat, source string) error {
|
||||
b, _ := json.Marshal(Changed{Seat: seat, Source: source, Digest: "d"})
|
||||
return f.handler("mesh-owner."+ChangedEvent, b)
|
||||
}
|
||||
|
||||
func def(seat, scope, verb string) string {
|
||||
return `{"format":1,"seat":"` + seat + `","scope":"` + scope + `","grain":"seat-list","decision":"ADR 1",` +
|
||||
`"verbs":[{"name":"` + verb + `","input":{"type":"object"}}]}`
|
||||
}
|
||||
|
||||
func names(r *Reader) string {
|
||||
var out []string
|
||||
for _, s := range r.Seats() {
|
||||
out = append(out, s.Name)
|
||||
}
|
||||
return strings.Join(out, ",")
|
||||
}
|
||||
|
||||
// ADR 0635 §1, §3: the seat list comes from the owner on the bus, starting from the bootstrap, and follows its event.
|
||||
func TestTheSeatListIsReadFromTheOwnerAndFollowsItsEvent(t *testing.T) {
|
||||
boot, err := seatdef.Parse([]byte(def("boot-seat", "mesh", "go")))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
owner := &fakeOwner{down: true, table: map[string]string{
|
||||
"node-b": def("node-b", "node", "put"), "mesh-a": def("mesh-a", "mesh", "get"), "boot-seat": def("boot-seat", "mesh", "other")}}
|
||||
var told [][]catalogue.Seat
|
||||
r, err := Start(Config{Owner: "mesh-owner", Bus: owner, Bootstrap: []seatdef.Definition{boot},
|
||||
OnChange: func(s []catalogue.Seat) { told = append(told, s) }})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if owner.pattern != "*."+ChangedEvent {
|
||||
t.Fatalf("subscribed to %q", owner.pattern)
|
||||
}
|
||||
steps := []struct {
|
||||
name string
|
||||
act func() error
|
||||
want string
|
||||
err string
|
||||
}{
|
||||
{"the owner has not answered: the bootstrap stands", func() error { return nil }, "boot-seat", ""},
|
||||
{"an event while the owner is down is offered again", func() error { return owner.say("mesh-a", SourceTrunk) }, "boot-seat", "no answer"},
|
||||
{"the owner answers: its seats in reading order, mesh seats first, the owner's definition replacing the bootstrap's",
|
||||
func() error { owner.down = false; return r.Reload() }, "boot-seat,mesh-a,node-b", ""},
|
||||
{"a moved definition is asked again", func() error {
|
||||
owner.table["mesh-a"] = def("mesh-a", "mesh", "renamed")
|
||||
return owner.say("mesh-a", SourceTrunk)
|
||||
}, "boot-seat,mesh-a,node-b", ""},
|
||||
{"a seat the owner holds no more goes", func() error { delete(owner.table, "node-b"); return owner.say("node-b", SourceTrunk) }, "boot-seat,mesh-a", ""},
|
||||
{"a pull request's definition is not held", func() error { owner.asked = nil; return owner.say("mesh-z", "pull") }, "boot-seat,mesh-a", ""},
|
||||
}
|
||||
for _, s := range steps {
|
||||
err := s.act()
|
||||
if (err == nil) != (s.err == "") || (err != nil && !strings.Contains(err.Error(), s.err)) {
|
||||
t.Fatalf("%s: error %v, want %q", s.name, err, s.err)
|
||||
}
|
||||
if got := names(r); got != s.want {
|
||||
t.Fatalf("%s: seats %q, want %q", s.name, got, s.want)
|
||||
}
|
||||
}
|
||||
if len(owner.asked) != 0 {
|
||||
t.Fatalf("a pull request's event asked the owner: %v", owner.asked)
|
||||
}
|
||||
if v := r.Seats()[1].Serves[0]; v.Name != "renamed" || v.Input["type"] != "object" {
|
||||
t.Fatalf("the moved definition's verb is %+v, want renamed ", v)
|
||||
}
|
||||
if len(told) != 3 {
|
||||
t.Fatalf("OnChange was told %d times", len(told))
|
||||
}
|
||||
}
|
||||
|
||||
func TestStartRefusesWhatCannotRead(t *testing.T) {
|
||||
for name, cfg := range map[string]Config{
|
||||
"no bus": {Owner: "m"},
|
||||
"an owner with a dot": {Owner: "m.tool", Bus: &fakeOwner{}},
|
||||
"no owner": {Bus: &fakeOwner{}},
|
||||
} {
|
||||
if _, err := Start(cfg); err == nil {
|
||||
t.Fatalf("%s: started", name)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,18 @@
|
||||
package seatlist
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
|
||||
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||
)
|
||||
|
||||
type stdioBus struct{}
|
||||
|
||||
// StdioBus is the bus a bundle has: the runtime's ask and subscribe over its pipe. Call Start once Serve runs.
|
||||
func StdioBus() Bus { return stdioBus{} }
|
||||
|
||||
func (stdioBus) Ask(key string, body any) (json.RawMessage, error) { return stdio.Ask(key, body) }
|
||||
|
||||
func (stdioBus) Subscribe(pattern string, handle func(key string, body json.RawMessage) error) error {
|
||||
return stdio.Subscribe(pattern, func(e stdio.Envelope) error { return handle(e.Key, e.Body) })
|
||||
}
|
||||
Reference in New Issue
Block a user