# `Portals.Stream`
[🔗](https://github.com/Ziinc/portals/blob/v0.1.0/lib/portals/stream.ex#L1)

A handle on one bidirectional stream, plus the demand/credit and
terminal-state machine behind it (FR-5, Phase 6).

A stream is opened by the BEAM with `Portals.open_stream/5`, which
dispatches an ordinary `CALL` frame and then uses that call's
`request_id` as the stream's `stream_id` (see `protocol/v1.md` §9).
Both directions then carry `STREAM_DATA`, and either peer may
`HALF_CLOSE` its own sending direction independently. The call's
`RETURN`/`ERROR` is the stream's terminal frame.

Backpressure is an encoded-byte credit window, independent in each
direction, plus a hard cap on the number of queued (sent but not yet
consumed) frames so a flood of tiny frames cannot evade the byte
window. Credits are charged against the *complete* encoded frame size,
including the 4-byte length prefix.

## Explicit stream-handle API

    {:ok, stream} = Portals.open_stream(conn, "bench_worker", "echo_stream", [])
    :ok = Portals.Stream.send(stream, "chunk-1")
    :ok = Portals.Stream.half_close(stream)
    {:ok, "chunk-1"} = Portals.Stream.recv(stream)
    :half_closed = Portals.Stream.recv(stream)
    {:ok, "done"} = Portals.Stream.await(stream)

## Enumerable API

    {:ok, stream} = Portals.open_stream(conn, "bench_worker", "echo_stream", [])
    :ok = Portals.Stream.send_enumerable(stream, ["a", "b", "c"])
    chunks = stream |> Portals.Stream.to_enumerable() |> Enum.to_list()

`Portals.Stream.send/3` blocks the calling process while the peer's
credit window is exhausted, so a slow consumer applies backpressure to
the producer instead of growing BEAM memory.

# `recv_result`

```elixir
@type recv_result() ::
  {:ok, term()}
  | :half_closed
  | {:closed, {:ok, term()} | {:error, Portals.Error.t()}}
  | {:error, :timeout}
```

# `t`

```elixir
@type t() :: %Portals.Stream{connection: pid(), id: non_neg_integer()}
```

# `await`

```elixir
@spec await(t(), timeout()) :: {:ok, term()} | {:error, Portals.Error.t()}
```

Block for the stream's terminal `RETURN`/`ERROR`, discarding (but
crediting) any inbound chunks still queued for the owner.

# `cancel`

```elixir
@spec cancel(t()) :: :ok
```

Cancel the stream in both directions, sending `CANCEL` and releasing
every piece of local state for it. Idempotent.

# `half_close`

```elixir
@spec half_close(t()) :: :ok
```

Close *only* the local sending direction. The peer may keep sending
until it half-closes its own direction. Idempotent.

# `recv`

```elixir
@spec recv(t(), timeout()) :: recv_result()
```

Receive the next inbound event for this stream in the owning process.

Consuming a chunk is what releases credit back to the peer, so a
process that stops calling `recv/2` stops the peer's producer.

# `send`

```elixir
@spec send(t(), term(), timeout()) :: :ok | {:error, Portals.Error.t()}
```

Send one chunk on the outbound direction, blocking until the peer has
granted enough byte credit (and frame allowance) to accept it.

Returns `{:error, %Portals.Error{}}` if the outbound direction is
already half-closed, the stream was cancelled or terminated, or the
chunk cannot be encoded within the connection's frame limits.

Note that a `send/3` that gives up on `timeout` only stops *waiting*:
the chunk stays queued behind the credit window and is still delivered
if credit arrives later. Cancel the stream if you need it not to be.

# `send_enumerable`

```elixir
@spec send_enumerable(t(), Enumerable.t(), keyword()) ::
  :ok | {:error, Portals.Error.t()}
```

Outbound `Enumerable` API: push every element of `enumerable` onto the
stream, respecting credit, then half-close the outbound direction.

Pass `half_close: false` to keep the outbound direction open.

# `to_enumerable`

```elixir
@spec to_enumerable(t(), timeout()) :: Enumerable.t()
```

The inbound direction as a lazy `Enumerable` of chunks. The enumeration
halts when the peer half-closes or the stream terminates; a terminal
`ERROR` raises `Portals.CallError`.

---

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