jschoubben
aecac5bda2
The mesh runs on the seat's bus alone (novox/hq ADR 0131, design 28 task 5.5). The old transport's consume loop, build request, tool ask, management API and account scoping are deleted, and the bus switch with them; the controller connects to the broker seat and to nothing else. The store-window tests keep their assertions on a bus-less fake, and the tests that only made sense for the old transport's in-memory holding go with it.
59 lines
2.3 KiB
Go
59 lines
2.3 KiB
Go
package link
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"time"
|
|
|
|
"github.com/nats-io/nats.go"
|
|
)
|
|
|
|
// Answer is what a module's tool replies: one of the two, never both.
|
|
type Answer struct {
|
|
Result json.RawMessage `json:"result,omitempty"`
|
|
Error string `json:"error,omitempty"`
|
|
}
|
|
|
|
// Ask calls one of a module's tools over the broker and waits for its answer.
|
|
//
|
|
// **The control plane is the way in** (novox/hq 04-ISSUES/049, ADR 0095). A module's broker
|
|
// account is scoped to what it emits, consumes and serves, and a tool call needs a reply queue the
|
|
// caller creates and a publish to the serving module's request key — which no module's scope
|
|
// grants, and should not. The control plane already holds a connection that may, so a person or
|
|
// an agent asks through it, and every question passes one process where an audit belongs.
|
|
//
|
|
// The answer comes back on the asker's own inbox, which only the asker may read; the serving
|
|
// module answers there and nowhere else. The bus refuses a request nothing serves at once, so a
|
|
// module that is down or a tool that does not exist is said now rather than after the whole wait.
|
|
func Ask(ctx context.Context, bus Bus, module, tool string, args json.RawMessage,
|
|
timeout time.Duration) (Answer, error) {
|
|
|
|
if len(args) == 0 {
|
|
args = json.RawMessage(`{}`)
|
|
}
|
|
key := module + "." + tool
|
|
reply, err := bus.AskTool(ctx, module, tool, args, timeout)
|
|
if err != nil {
|
|
if errors.Is(err, nats.ErrNoResponders) {
|
|
return Answer{}, fmt.Errorf(
|
|
"nothing serves %s: no runtime has bound %q on the bus. The module is not "+
|
|
"assigned, its runtime is not up, or it serves no such tool — `status` says "+
|
|
"whether the machine carrying it has applied", module, key)
|
|
}
|
|
if errors.Is(err, context.DeadlineExceeded) || errors.Is(err, nats.ErrTimeout) {
|
|
return Answer{}, fmt.Errorf(
|
|
"%s did not answer within %s. Its runtime serves %q when it is up and has bound "+
|
|
"the bus — `status` says whether the machine carrying it has applied",
|
|
module, timeout, key)
|
|
}
|
|
return Answer{}, fmt.Errorf("cannot ask %s: %w", key, err)
|
|
}
|
|
var answer Answer
|
|
if err := json.Unmarshal(reply, &answer); err != nil {
|
|
return Answer{}, fmt.Errorf("%s answered with something unreadable: %w", key, err)
|
|
}
|
|
return answer, nil
|
|
}
|