tool runner: mesh/stream.publish says an event on the calling bundle's own bound stream (A-42)

Builds: design 42 zfs-owns-zfs §25, §24, lead answer A-42
This commit is contained in:
2026-11-13 01:49:02 +01:00
parent 447a8d0359
commit a775b42e4d
6 changed files with 97 additions and 2 deletions
+35
View File
@@ -275,6 +275,41 @@ func (c *Conn) StreamFill(module string, r buscalls.StreamRequest) (buscalls.Str
return c.streamFill(b.Stream)
}
// StreamPublish says one event on the module's own bound stream, by an acknowledged publish on the binding's subject plus
// "." plus the event, the id as the message id (design 42 zfs-owns-zfs, A-42). The module's own binding alone: another
// module's, and a binding that is not a stream, are refused by Binding.
func (c *Conn) StreamPublish(module string, r buscalls.StreamPublishRequest) (buscalls.StreamPublished, error) {
b, err := c.streamBinding(module, r.Stream)
if err != nil {
return buscalls.StreamPublished{}, err
}
if b.Subject == "" {
return buscalls.StreamPublished{}, fmt.Errorf("%s's stream %q has no subject to say an event on", module, r.Stream)
}
if r.Event == "" || strings.ContainsAny(r.Event, " *>\t\r\n") || strings.HasPrefix(r.Event, ".") || strings.HasSuffix(r.Event, ".") {
return buscalls.StreamPublished{}, fmt.Errorf("the event %q is not a name to say on %s's stream %q", r.Event, module, r.Stream)
}
body := []byte(r.Body)
if len(body) == 0 {
body = []byte("null")
}
ctx, cancel := context.WithTimeout(context.Background(), StateTimeout)
defer cancel()
js, err := c.jetStream()
if err != nil {
return buscalls.StreamPublished{}, err
}
var opts []jetstream.PublishOpt
if r.ID != "" {
opts = append(opts, jetstream.WithMsgID(r.ID))
}
ack, err := js.Publish(ctx, b.Subject+"."+r.Event, body, opts...)
if err != nil {
return buscalls.StreamPublished{}, fmt.Errorf("the event %s was not acknowledged on %s's stream %q: %w", r.Event, module, r.Stream, err)
}
return buscalls.StreamPublished{Sequence: ack.Sequence}, nil
}
// StreamReadMost is the most messages one read answers.
const StreamReadMost = 1000