RTMP.ChunkStream (rtmp v0.1.0)

The RTMP chunk stream layer: splits a byte stream into chunks and reassembles them into messages (parse/2), and does the reverse for outgoing messages (encode/6).

Handles every chunk header format (fmt 0–3), all three basic-header widths (chunk stream IDs 2–65599), extended timestamps, and chunks from different chunk streams arriving interleaved — encoders routinely put an audio message between two chunks of a video message.

The two protocol control messages that change how chunks are framed are applied here, the moment they're parsed, so they take effect for the very next chunk in the same buffer:

Set Chunk Size (type 1)
Abort Message (type 2)

They are not returned to the caller. Every other message is.

Timestamps

Message timestamps are RTMP's 32-bit millisecond timestamps and wrap at 2^32 (about 49.7 days). A fmt 3 header that starts a new message reuses the previous header's timestamp field as its delta, matching how FFmpeg and librtmp write such headers.

Extended timestamps on continuation chunks

The spec says a fmt 3 continuation chunk repeats the 4-byte extended timestamp when the message's header used one; some older encoders don't. Continuation chunks are handled the way SRS and nginx-rtmp do it: the 4 bytes are consumed only if they equal the timestamp from the message's header. A new message started with fmt 3 always carries them.

Limits

Enforces max_chunk_size, max_message_size, max_chunk_streams and max_buffered_bytes from RTMP.Limits. Any violation — or any header that can't be valid — returns {:error, reason}; nothing here raises on network input.

Summary

Types

t()

Functions

chunk_size(chunk_stream)

The chunk size currently used for incoming chunks.

encode(csid, type_id, stream_id, timestamp, payload, chunk_size)

Encodes one message as chunks: a fmt 0 header for the first chunk, fmt 3 for every continuation.

new(limits \\ %Limits{})

Creates a parser with the default chunk size of 128 bytes.

parse(cs, buffer)

Parses as many complete chunks as buffer holds.

Types

t()

@type t() :: %RTMP.ChunkStream{
  buffered: non_neg_integer(),
  chunk_size: pos_integer(),
  limits: RTMP.Limits.t(),
  streams: %{optional(pos_integer()) => map()}
}

Functions

chunk_size(chunk_stream)

@spec chunk_size(t()) :: pos_integer()

The chunk size currently used for incoming chunks.

encode(csid, type_id, stream_id, timestamp, payload, chunk_size)

@spec encode(
  csid :: pos_integer(),
  type_id :: byte(),
  stream_id :: non_neg_integer(),
  timestamp :: non_neg_integer(),
  payload :: iodata(),
  chunk_size :: pos_integer()
) :: iodata()

Encodes one message as chunks: a fmt 0 header for the first chunk, fmt 3 for every continuation.

csid must be in 2..65599. Returns iodata ready to write to the socket.

new(limits \\ %Limits{})

@spec new(RTMP.Limits.t()) :: t()

Creates a parser with the default chunk size of 128 bytes.

parse(cs, buffer)

@spec parse(t(), binary()) ::
  {:ok, [RTMP.ChunkStream.Message.t()], t(), binary()}
  | {:error, term(), [RTMP.ChunkStream.Message.t()]}

Parses as many complete chunks as buffer holds.

Returns {:ok, messages, state, rest} where messages are the messages completed by those chunks (in order) and rest is the unconsumed tail — an incomplete chunk to be retried once more data arrives.

Returns {:error, reason, messages} on a protocol violation or limit breach, where messages are those completed before the offending chunk. They're returned so the outcome doesn't depend on how TCP happened to split the bytes: a frame that fully arrived is delivered whether or not the garbage after it shared its read. The stream can't be resynchronised after an error and the connection should be closed.