CrowdControl.Backend.Docker.Demux (crowd_control v0.1.1)

Copy Markdown View Source

Docker multiplexed-stream framing.

When a container is created without a TTY, the Engine API interleaves stdout and stderr on one connection using an 8-byte header per frame:

<<stream_type::8, 0::24, length::32-big, payload::binary-size(length)>>

where stream_type is 0 stdin, 1 stdout, 2 stderr.

Why this is resumable

Frames do not align with HTTP chunk boundaries. A header can arrive split across two chunks — three bytes in one, five in the next — and a payload can span several. So this is a feed/state machine rather than a parser over a complete binary: feed/2 returns whatever complete payloads it can and carries the remainder forward. It deliberately mirrors the pure-function shape of CrowdControl.Protocol.split_lines/1.

Only stdout survives

Stream types other than 1 are dropped. CrowdControl controls the exec command, so stdout carries the CLI's stream-json and nothing else; stderr is diagnostic noise that would corrupt the JSON line stream if merged into it. Stderr frames really do occur in practice — a tail against a missing file produces them — so this is load-bearing, not defensive.

iex> alias CrowdControl.Backend.Docker.Demux
iex> frame = <<1, 0, 0, 0, 0, 0, 0, 5, "hello">>
iex> {payloads, _state} = Demux.feed(Demux.new(), frame)
iex> payloads
["hello"]

Summary

Types

t()

Opaque demux state; carries the bytes of an incomplete frame.

Functions

Feed bytes in, get complete stdout payloads out.

A demux state with nothing buffered.

Bytes currently held for an incomplete frame.

Types

t()

@opaque t()

Opaque demux state; carries the bytes of an incomplete frame.

Functions

feed(demux, data)

@spec feed(t(), binary()) :: {[binary()], t()}

Feed bytes in, get complete stdout payloads out.

Returns {payloads, state}. Pass state to the next call.

iex> alias CrowdControl.Backend.Docker.Demux
iex> # a header split across two feeds
iex> {[], s} = Demux.feed(Demux.new(), <<1, 0, 0>>)
iex> {payloads, _s} = Demux.feed(s, <<0, 0, 0, 0, 3, "abc">>)
iex> payloads
["abc"]

iex> alias CrowdControl.Backend.Docker.Demux
iex> # stderr (type 2) is dropped, stdout (type 1) survives
iex> err = <<2, 0, 0, 0, 0, 0, 0, 3, "err">>
iex> out = <<1, 0, 0, 0, 0, 0, 0, 2, "ok">>
iex> {payloads, _s} = Demux.feed(Demux.new(), err <> out)
iex> payloads
["ok"]

new()

@spec new() :: t()

A demux state with nothing buffered.

pending(demux)

@spec pending(t()) :: non_neg_integer()

Bytes currently held for an incomplete frame.

Non-zero at end of stream means the stream was cut mid-frame.