# `RTMP.ChunkStream`
[🔗](https://github.com/Adeyeye93/rtmp/blob/v0.1.0/lib/rtmp/chunk_stream.ex#L1)

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.

# `t`

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

# `chunk_size`

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

The chunk size currently used for incoming chunks.

# `encode`

```elixir
@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`

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

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

# `parse`

```elixir
@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.

---

*Consult [api-reference.md](api-reference.md) for complete listing*