Owned, bounded, structured streaming of session command logs.
Where ExDaytona.Session.stream_logs/4 delivers merged raw chunks over
HTTP, a LogStream speaks the websocket protocol Daytona multiplexes
command output on, and demultiplexes it back into separate :stdout
and :stderr events in provider arrival order.
Channel separation depends on the daemon
Separation requires the daemon to emit the channel-marker protocol.
Daemons that stream unlabeled output (the production daemon at the
time of this release — verified live) yield {:output, bytes}
events carrying the merged stream instead; consumers should handle
all three event shapes.
{:ok, stream} =
ExDaytona.Session.open_log_stream(session, cmd_id,
max_buffer_bytes: 1_048_576,
idle_timeout: 60_000
)
case ExDaytona.LogStream.next(stream, 5_000) do
{:ok, {:stdout, bytes}} -> IO.write(bytes)
{:ok, {:stderr, bytes}} -> IO.write(:stderr, bytes)
{:closed, :normal} -> :done
{:closed, {:error, %ExDaytona.Error{}}} -> :failed
end
:ok = ExDaytona.LogStream.close(stream)Contract
- Pull-based: events are buffered inside the stream process and
handed out one
next/2at a time — no unbounded delivery into an ordinary mailbox, and nothing the consumer does can block the socket receive loop. - Owned: the stream monitors its owner (default: the opener) and
shuts the connection down when the owner dies.
close/1is idempotent. - Bounded:
max_buffer_bytes/max_framescap undelivered output andmax_frame_bytescaps a single websocket frame; exceeding a bound closes the connection with an explicit overflow error. - Deadlines:
idle_timeout(between frames) andoverall_timeout(whole stream) close the stream with an explicit error. - No hidden recovery: a dropped connection is reported as closed; the stream never silently reconnects or replays (the provider offers no cursor to resume from).
The final exit status is not part of the stream — after
{:closed, :normal} use ExDaytona.Session.command/2 or await/3.
Summary
Functions
Returns a specification to start this module under a supervisor.
Close the stream (idempotent). Buffered undelivered events are discarded.
Collect events until the stream closes, returning
{:ok, %{stdout: binary, stderr: binary, closed: reason}}. Convenience
for short commands; long-lived follows should loop next/2.
The next event, waiting up to timeout ms.
Open a log stream over an established websocket URL. Usually called
through ExDaytona.Session.open_log_stream/3.
Types
@type close_reason() :: :normal | {:error, ExDaytona.Error.t()}
Functions
Returns a specification to start this module under a supervisor.
See Supervisor.
@spec close(pid()) :: :ok
Close the stream (idempotent). Buffered undelivered events are discarded.
@spec collect(pid(), timeout()) :: {:ok, %{ stdout: binary(), stderr: binary(), output: binary(), closed: close_reason() }} | {:error, ExDaytona.Error.t()}
Collect events until the stream closes, returning
{:ok, %{stdout: binary, stderr: binary, closed: reason}}. Convenience
for short commands; long-lived follows should loop next/2.
@spec next(pid(), timeout()) :: {:ok, event()} | {:closed, close_reason()} | {:error, ExDaytona.Error.t()}
The next event, waiting up to timeout ms.
Returns {:ok, {:stdout | :stderr | :output, binary}} (:output =
unlabeled bytes from daemons without channel marking),
{:closed, reason} once
the stream has ended and the buffer is drained (reason is :normal
or {:error, %ExDaytona.Error{}} for overflow/timeout/transport
failures), or {:error, %ExDaytona.Error{}} when timeout elapses
with the stream still live.
@spec open(String.t(), String.t(), keyword()) :: {:ok, pid()} | {:error, ExDaytona.Error.t()}
Open a log stream over an established websocket URL. Usually called
through ExDaytona.Session.open_log_stream/3.
Options
:owner— monitored owner pid (default: the caller):ws_mod— websocket transport module (defaultExDaytona.WebSocket):max_buffer_bytes— cap on undelivered bytes (default1_048_576):max_frames— cap on undelivered events (default10_000):max_frame_bytes— cap on one websocket frame (default1_048_576):idle_timeout— ms without any frame before erroring (default60_000):overall_timeout— ms for the whole stream (default:infinity):connect_timeout— ms for the websocket upgrade (default15_000)