The node seat node-engine served on the engine's link: its verbs with their proven caller, its acts kept in its own state, a hand-over or access act asked of the operator as node-engine-<node> and performed only on the warrant (issue 774, ADR 0563 §1-§3)
This commit is contained in:
@@ -0,0 +1,260 @@
|
||||
package link
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||
"git.novox.be/novox/mesh-sdk/go/asks"
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
"github.com/nats-io/nats.go"
|
||||
"github.com/nats-io/nats.go/micro"
|
||||
)
|
||||
|
||||
// The node seat node-engine, served on the engine's own link (novox/hq ADR 0563 §1).
|
||||
//
|
||||
// Each verb is answered on this node's subject alone, `mesh.seat.node-engine.tool.<verb>.<node>`, and on the same
|
||||
// address naming its caller, `mesh.seat.node-engine.call.<verb>.<node>.<caller>`, whose last token the bus lets each user
|
||||
// publish with its own name alone (novox/hq issue 365): the caller is that token, and a call on the subject naming none is
|
||||
// from an unproven caller, recorded so. The body is the call's arguments; the answer is {result, error, node}, as every
|
||||
// tool's (mesh-tools node-tools). One service, `node-engine`, answers discovery with each verb's seat, tool and node.
|
||||
|
||||
// EngineSeat is the seat.
|
||||
const EngineSeat = catalogue.EngineSeat
|
||||
|
||||
// EngineToolSubject is where a verb of this node's seat is asked naming no caller.
|
||||
func EngineToolSubject(verb, node string) string {
|
||||
return "mesh.seat." + EngineSeat + ".tool." + verb + "." + node
|
||||
}
|
||||
|
||||
// EngineCallPattern is where a verb of this node's seat is asked naming its caller, any caller.
|
||||
func EngineCallPattern(verb, node string) string {
|
||||
return "mesh.seat." + EngineSeat + ".call." + verb + "." + node + ".*"
|
||||
}
|
||||
|
||||
// EngineEventSubject is where the seat says one of its events about this node.
|
||||
func EngineEventSubject(event, node string) string {
|
||||
return "mesh.seat." + EngineSeat + ".event." + event + "." + node
|
||||
}
|
||||
|
||||
// EngineCall is one call of the seat: its verb, its caller as the bus proved it, and its arguments.
|
||||
type EngineCall struct {
|
||||
Verb string
|
||||
Caller string
|
||||
Args map[string]any
|
||||
}
|
||||
|
||||
// EngineFunc answers one call: its result, or why it is refused.
|
||||
type EngineFunc func(EngineCall) (any, error)
|
||||
|
||||
// CallerOfSubject is the caller a call's subject names for this node's verb: the last token of a caller-named subject,
|
||||
// each `~` a dot, or stdio.Unproven on the subject that names none or on any other.
|
||||
func CallerOfSubject(subject, verb, node string) string {
|
||||
token, ok := strings.CutPrefix(subject, "mesh.seat."+EngineSeat+".call."+verb+"."+node+".")
|
||||
if !ok || token == "" || strings.ContainsAny(token, ".*> ") {
|
||||
return stdio.Unproven
|
||||
}
|
||||
return strings.ReplaceAll(token, "~", ".")
|
||||
}
|
||||
|
||||
// MaxEngineCall bounds a call's arguments: a path, an access and a few words.
|
||||
const MaxEngineCall = 16 << 10
|
||||
|
||||
// AnswerEngineCall is the envelope one call is answered with: the verb's result, or its refusal in words.
|
||||
func AnswerEngineCall(handle EngineFunc, verb, node, subject string, body []byte) []byte {
|
||||
envelope := map[string]any{"node": node}
|
||||
say := func() []byte {
|
||||
raw, err := json.Marshal(envelope)
|
||||
if err != nil {
|
||||
raw, _ = json.Marshal(map[string]any{"node": node, "error": "the answer could not be written: " + err.Error()})
|
||||
}
|
||||
return raw
|
||||
}
|
||||
args := map[string]any{}
|
||||
switch {
|
||||
case handle == nil:
|
||||
envelope["error"] = "this engine serves no verb of " + EngineSeat
|
||||
return say()
|
||||
case len(body) > MaxEngineCall:
|
||||
envelope["error"] = fmt.Sprintf("the call is %d bytes, more than a call of %s takes", len(body), EngineSeat)
|
||||
return say()
|
||||
case len(strings.TrimSpace(string(body))) > 0:
|
||||
if err := json.Unmarshal(body, &args); err != nil {
|
||||
envelope["error"] = "the call's arguments are not an object: " + err.Error()
|
||||
return say()
|
||||
}
|
||||
}
|
||||
result, err := handle(EngineCall{Verb: verb, Caller: CallerOfSubject(subject, verb, node), Args: args})
|
||||
if err != nil {
|
||||
envelope["error"] = err.Error()
|
||||
} else {
|
||||
envelope["result"] = result
|
||||
}
|
||||
return say()
|
||||
}
|
||||
|
||||
// EngineSeats is a link that serves this node's seat. Optional, as Asked is.
|
||||
type EngineSeats interface {
|
||||
ServeEngineSeat(EngineFunc)
|
||||
}
|
||||
|
||||
// ServeEngineSeat answers every verb of the seat on this node's subjects, as the service node-engine, which says each
|
||||
// verb's seat, tool and node to discovery (novox/hq ADR 0197). A bus whose user list is older than the grant refuses the
|
||||
// subscriptions, said by the client; a caller then hears nobody answer.
|
||||
func (l *natsLink) ServeEngineSeat(handle EngineFunc) {
|
||||
if handle == nil || l.conn == nil {
|
||||
return
|
||||
}
|
||||
svc, err := micro.AddService(l.conn, micro.Config{Name: EngineSeat, Version: "1.0.0",
|
||||
Description: "the node seat node-engine of " + l.node + " (novox/hq ADR 0563)", QueueGroupDisabled: true})
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
for _, v := range catalogue.EngineVerbs() {
|
||||
verb := v.Name
|
||||
answer := micro.HandlerFunc(func(r micro.Request) {
|
||||
_ = r.Respond(AnswerEngineCall(handle, verb, l.node, r.Subject(), r.Data()))
|
||||
})
|
||||
meta := map[string]string{"seat": EngineSeat, "tool": verb, "node": l.node}
|
||||
_ = svc.AddEndpoint(verb, answer, micro.WithEndpointSubject(EngineToolSubject(verb, l.node)),
|
||||
micro.WithEndpointMetadata(meta))
|
||||
_ = svc.AddEndpoint(verb+"-called", answer, micro.WithEndpointSubject(EngineCallPattern(verb, l.node)),
|
||||
micro.WithEndpointMetadata(meta))
|
||||
}
|
||||
l.engine = svc
|
||||
}
|
||||
|
||||
// SayEngine says one of the seat's events about this node, on the events stream.
|
||||
func (l *natsLink) SayEngine(event string, body any) error {
|
||||
raw, err := json.Marshal(body)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
||||
defer cancel()
|
||||
_, err = l.js.Publish(EngineEventSubject(event, l.node), raw, nats.Context(ctx))
|
||||
return err
|
||||
}
|
||||
|
||||
// The engine asks the operator under its own asker name (asks.EngineAsker) on the router's seat (ADR 0259 §3, as ADR
|
||||
// 0563 §3 changes it): its ask and cancel on the seat's work stream, its warrants heard on its own decided subject, and
|
||||
// its asks' state read from the router's records under its own name.
|
||||
|
||||
// RouterRecords is the router's bucket of asks, which each asker reads under its own name: the state `asks` of the
|
||||
// operator-channel's holder, the messenger (mesh-catalog modules/messenger).
|
||||
const RouterRecords = "messenger_asks"
|
||||
|
||||
// OperatorAsks is a link that asks the operator as this node's engine. Optional.
|
||||
type OperatorAsks interface {
|
||||
PublishAsk(subject string, body any, id string) error
|
||||
RouterRecord(key string) (*stdio.StateEntry, error)
|
||||
HearWarrants(asker string, heard func(asks.Warrant)) error
|
||||
}
|
||||
|
||||
// PublishAsk publishes an ask or a cancel on the router's seat, once per id within the stream's duplicate window.
|
||||
func (l *natsLink) PublishAsk(subject string, body any, id string) error {
|
||||
raw, err := json.Marshal(body)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
||||
defer cancel()
|
||||
_, err = l.js.Publish(subject, raw, nats.MsgId(id), nats.Context(ctx))
|
||||
return err
|
||||
}
|
||||
|
||||
// RouterRecord reads one key of the router's records, `<asker>.<id>`; nil when it holds none.
|
||||
func (l *natsLink) RouterRecord(key string) (*stdio.StateEntry, error) {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
||||
defer cancel()
|
||||
msg, err := l.conn.RequestWithContext(ctx, "$JS.API.DIRECT.GET.KV_"+RouterRecords+".$KV."+RouterRecords+"."+key, nil)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if status := msg.Header.Get("Status"); status == "404" {
|
||||
return nil, nil
|
||||
} else if status != "" {
|
||||
return nil, fmt.Errorf("the router's record %s could not be read: %s %s", key, status, msg.Header.Get("Description"))
|
||||
}
|
||||
return &stdio.StateEntry{Key: key, Value: msg.Data}, nil
|
||||
}
|
||||
|
||||
// HearWarrants hears the warrants for asker on its decided subject, each handed to heard as it arrives.
|
||||
func (l *natsLink) HearWarrants(asker string, heard func(asks.Warrant)) error {
|
||||
sub, err := l.conn.Subscribe(asks.DecidedSubject(asker), func(msg *nats.Msg) {
|
||||
var w asks.Warrant
|
||||
if json.Unmarshal(msg.Data, &w) == nil {
|
||||
heard(w)
|
||||
}
|
||||
})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
l.warrants = sub
|
||||
return nil
|
||||
}
|
||||
|
||||
// errNoAsking is a link that cannot ask the operator.
|
||||
var errNoAsking = errors.New("the link open now cannot ask the operator")
|
||||
|
||||
// discoveryQuiet is how long the engine waits after the last answer to the bus's discovery, and discoveryMost the
|
||||
// longest it waits in all.
|
||||
const (
|
||||
discoveryQuiet = 300 * time.Millisecond
|
||||
discoveryMost = 2 * time.Second
|
||||
)
|
||||
|
||||
// HeardServed asks the bus's discovery (novox/hq ADR 0197) whether a seat's verb is answered on this machine: an
|
||||
// endpoint naming the seat, the verb as its tool and this node.
|
||||
func (l *natsLink) HeardServed(seat, verb string) (bool, error) {
|
||||
inbox := l.conn.NewRespInbox()
|
||||
sub, err := l.conn.SubscribeSync(inbox)
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
defer func() { _ = sub.Unsubscribe() }()
|
||||
if err := l.conn.PublishRequest("$SRV.INFO", inbox, nil); err != nil {
|
||||
return false, fmt.Errorf("asking the bus who serves what: %w", err)
|
||||
}
|
||||
heard := false
|
||||
deadline := time.Now().Add(discoveryMost)
|
||||
for time.Now().Before(deadline) {
|
||||
msg, err := sub.NextMsg(discoveryQuiet)
|
||||
if err != nil {
|
||||
break
|
||||
}
|
||||
var info micro.Info
|
||||
if json.Unmarshal(msg.Data, &info) != nil {
|
||||
continue
|
||||
}
|
||||
for _, e := range info.Endpoints {
|
||||
node := e.Metadata["node"]
|
||||
if node == "" {
|
||||
node = info.ID
|
||||
}
|
||||
if e.Metadata["seat"] == seat && e.Metadata["tool"] == verb && node == l.node {
|
||||
heard = true
|
||||
}
|
||||
}
|
||||
}
|
||||
return heard, nil
|
||||
}
|
||||
|
||||
// HeardServed is the link open now asking the bus's discovery whether a seat's verb is answered on this machine.
|
||||
func (q *Session) HeardServed(seat, verb string) (bool, error) {
|
||||
q.init()
|
||||
q.mu.Lock()
|
||||
bus := q.bus
|
||||
q.mu.Unlock()
|
||||
asker, ok := bus.(interface {
|
||||
HeardServed(string, string) (bool, error)
|
||||
})
|
||||
if bus == nil || !ok {
|
||||
return false, fmt.Errorf("%w: this machine has no link to the bus open now", ErrNotAsked)
|
||||
}
|
||||
return asker.HeardServed(seat, verb)
|
||||
}
|
||||
Reference in New Issue
Block a user