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.
Summary
Functions
Block for the stream's terminal RETURN/ERROR, discarding (but
crediting) any inbound chunks still queued for the owner.
Cancel the stream in both directions, sending CANCEL and releasing
every piece of local state for it. Idempotent.
Close only the local sending direction. The peer may keep sending until it half-closes its own direction. Idempotent.
Receive the next inbound event for this stream in the owning process.
Send one chunk on the outbound direction, blocking until the peer has granted enough byte credit (and frame allowance) to accept it.
Outbound Enumerable API: push every element of enumerable onto the
stream, respecting credit, then half-close the outbound direction.
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.
Types
@type recv_result() :: {:ok, term()} | :half_closed | {:closed, {:ok, term()} | {:error, Portals.Error.t()}} | {:error, :timeout}
@type t() :: %Portals.Stream{connection: pid(), id: non_neg_integer()}
Functions
@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.
@spec cancel(t()) :: :ok
Cancel the stream in both directions, sending CANCEL and releasing
every piece of local state for it. Idempotent.
@spec half_close(t()) :: :ok
Close only the local sending direction. The peer may keep sending until it half-closes its own direction. Idempotent.
@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.
@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.
@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.
@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.