60 lines
2.1 KiB
Go
60 lines
2.1 KiB
Go
package stdio
|
|
|
|
import (
|
|
"encoding/json"
|
|
"fmt"
|
|
)
|
|
|
|
// A module's own stream read back (novox/hq ADR 0363): every module may read the streams it declared under `streams`,
|
|
// and no other's, by the runtime's stream.read, a batch of direct gets with no consumer, as a log is read. A module's
|
|
// tools read with it what its stream holds (a build step's log, issue: the full failure output), within the stream's age.
|
|
|
|
// StreamEntry is one event of a module's own stream: its sequence, its event's name (the subject after
|
|
// `mesh.stream.<module>.event.`), its body and when the bus took it.
|
|
type StreamEntry struct {
|
|
Sequence uint64 `json:"sequence"`
|
|
Event string `json:"event"`
|
|
Body json.RawMessage `json:"body"`
|
|
At string `json:"at"`
|
|
}
|
|
|
|
// StreamPage is one page of a read: its entries in order, the sequence to read on from (0 when nothing follows) and the
|
|
// stream's last sequence when it was read.
|
|
type StreamPage struct {
|
|
Entries []StreamEntry `json:"entries"`
|
|
Next uint64 `json:"next"`
|
|
Last uint64 `json:"last"`
|
|
}
|
|
|
|
// KeptStream is one stream this module declared.
|
|
type KeptStream struct{ name string }
|
|
|
|
// Stream is a stream this module declared, by its local name.
|
|
func Stream(name string) KeptStream { return KeptStream{name: name} }
|
|
|
|
// Read is the events from the sequence from on (0 is the first), whose event names match events (`*` for one name,
|
|
// `>` for the rest, last; empty is every event the stream carries), oldest first, at most limit (0 is the runtime's
|
|
// page, 100; at most 1000). Read on from the page's Next until it is 0.
|
|
func (s KeptStream) Read(events string, from uint64, limit int) (*StreamPage, error) {
|
|
if limit < 0 {
|
|
return nil, fmt.Errorf("a read of %s asks for at least no entries: limit %d", s.name, limit)
|
|
}
|
|
c, err := live()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
params := map[string]any{"stream": s.name, "from": from}
|
|
if events != "" {
|
|
params["subject"] = events
|
|
}
|
|
if limit > 0 {
|
|
params["limit"] = limit
|
|
}
|
|
raw, err := c.ask("mesh/stream.read", params, logTimeout)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
var p StreamPage
|
|
return &p, json.Unmarshal(raw, &p)
|
|
}
|