Documentation
¶
Overview ¶
Package eventstream maps magus's internal producers onto the single types.StreamEvent envelope external integrations subscribe to.
It is an ADAPTER layer: the producers it reads keep their own on-disk schemas, and nothing here rewrites them.
The run journal is the only source today, and it covers the whole taxonomy. It is a stdlib-only leaf, so adapting it costs this package no dependency on the engine. The report writer, the attention store and the trail hold facts a subscriber would want, but each is a different file in a different directory rather than one bus; docs/guides/integrations/editor/design.md prices them.
Index ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Follower ¶
type Follower struct {
// contains filtered or unexported fields
}
Follower turns the workspace's run-log directory into a live event stream.
The directory IS the bus, and that is the design rather than a fallback. Every magus process in a workspace already appends its invocation to <cacheDir>/runs/<inv>.jsonl, so a follower reading that directory sees runs started from any terminal, any editor, and the daemon alike - with no daemon required, no socket to discover, and no token to provision. A subscriber that wants lower latency than a poll can take the daemon socket instead; it learns the same events.
Two properties bound what a follower sees, and both are deliberate:
- Output lines lag. journal.FileHandler flushes every kind EXCEPT output, so lifecycle and result events land immediately while output arrives when a bufio page fills or the run ends. A subscriber asking for output is opting into chunked delivery.
- A partially written line is never emitted. An output-triggered page flush can split a line, so reads stop at the last newline and resume there.
A Follower is not safe for concurrent use; one goroutine drives it.
func NewFollower ¶
NewFollower returns a Follower over runsDir, stamping workspace on every event it produces. It touches no files until Replay or Follow is called, so a workspace that has never run anything is not an error.
func (*Follower) Follow ¶
func (f *Follower) Follow(ctx context.Context, interval time.Duration, emit func(types.StreamEvent) error) error
Follow polls the run-log directory every interval and emits what appears, blocking until ctx is cancelled. It returns nil on cancellation: a follower stopping because it was asked to is not a failure.
An emit error stops the follow and is returned - that is the subscriber's pipe closing, and continuing to read a directory nobody is listening to is waste. A read error on one log is skipped rather than fatal, because a log being written concurrently is the normal case.
func (*Follower) Replay ¶
Replay emits the events already on disk from the newest limit invocations, oldest event first, and leaves the Follower positioned at the end of each file so a subsequent Follow does not repeat them.
A limit of 0 or less replays every retained invocation. Ordering is by file modification time rather than by the timestamps inside, because an invocation still running has no final timestamp to sort on.
func (*Follower) Skip ¶
Skip positions the Follower at the end of every log currently on disk without emitting anything, so Follow reports only what happens next.
This is what `--follow` without a replay window needs: attaching to a workspace with months of retained runs must not deliver months of history first.
type Writer ¶
type Writer struct {
// contains filtered or unexported fields
}
Writer serializes stream events as JSONL onto an io.Writer, dropping what the filter excludes.
It is safe for concurrent use: a run fans out one goroutine per project, so events arrive from several at once and an unsynchronized write would interleave two JSON objects on one line.
func NewWriter ¶
func NewWriter(dst io.Writer, filter types.StreamFilter) *Writer
NewWriter returns a Writer emitting to dst under filter. The caller owns dst and must call Writer.Close to flush; a Writer that is never closed loses whatever is still buffered.
func (*Writer) Close ¶
Close flushes any buffered line. It does not close the underlying writer, which the caller owns.
func (*Writer) Emit ¶
func (w *Writer) Emit(e types.StreamEvent) error
Emit writes one event, or does nothing when the filter excludes its type.
It flushes per event rather than relying on the buffer filling: a subscriber is usually a pipe being read live, and a status bar that learns about a failed build once 4KB has accumulated is worse than no status bar.
Source Files