Portals.Stream.State (portals v0.1.0)

The per-stream credit and terminal-state machine, owned functionally by Portals.Connection (like the pending-request table) rather than by a process of its own.

Two completely independent windows:

send window — send_credit bytes the peer has granted us and send_frames frames we have sent that the peer has not yet acknowledged, capped by max_frames;
receive window — recv_credit bytes we have granted the peer and recv_frames frames it has sent that our owner has not yet consumed, capped by max_frames.

Every byte figure is the complete encoded frame size, including the 4-byte length prefix.

Summary

Types

direction_state()

t()

Functions

add_send_credit(state, bytes, frames)

Apply a CREDIT frame received from the peer.

charge_send(state, bytes)

Charge one outbound frame of bytes encoded bytes against the send window. Never sends: the caller must only put the frame on the wire when this returns {:ok, state}.

consume_inbound(state, bytes, frames)

The owner consumed frames inbound frames totalling bytes encoded bytes. Returns the replenished state and the number of bytes to grant back to the peer in a CREDIT frame.

fully_half_closed?(state)

True when both directions are half-closed; the terminal frame may still be outstanding.

grant(state)

The initial (or replenishing) grant this side owes the peer: the unused portion of our receive window. Returns {state, bytes}; bytes is 0 when there is nothing new to grant.

half_close_inbound(state)

Record the peer's HALF_CLOSE of its sending direction. Idempotent.

half_close_outbound(state)

Close only this side's sending direction. Idempotent.

new(id, owner, opts)

A fresh stream with no credit in either direction.

park_sender(state, waiter)

Park a blocked sender until credit arrives.

pop_sender(state)

Pop the next parked sender, if any.

record_inbound(state, bytes)

Record one inbound STREAM_DATA frame of bytes encoded bytes. Rejects anything the peer was not entitled to send.

take_senders(state)

Every parked sender, in order, leaving the queue empty.

Types

direction_state()

@type direction_state() :: :open | :closed

t()

@type t() :: %Portals.Stream.State{
  id: non_neg_integer(),
  inbound: direction_state(),
  max_credit: pos_integer(),
  max_frames: pos_integer(),
  outbound: direction_state(),
  owner: pid(),
  owner_ref: reference() | nil,
  recv_bytes: non_neg_integer(),
  recv_credit: non_neg_integer(),
  recv_frames: non_neg_integer(),
  send_credit: non_neg_integer(),
  send_frames: non_neg_integer(),
  send_waiters: :queue.queue()
}

Functions

add_send_credit(state, bytes, frames)

@spec add_send_credit(t(), non_neg_integer(), non_neg_integer()) ::
  {:ok, t()} | {:error, :credit_limit_exceeded}

Apply a CREDIT frame received from the peer.

charge_send(state, bytes)

@spec charge_send(t(), pos_integer()) ::
  {:ok, t()} | {:error, :outbound_closed | :insufficient_credit | :frame_limit}

Charge one outbound frame of bytes encoded bytes against the send window. Never sends: the caller must only put the frame on the wire when this returns {:ok, state}.

consume_inbound(state, bytes, frames)

@spec consume_inbound(t(), non_neg_integer(), non_neg_integer()) ::
  {t(), non_neg_integer()}

The owner consumed frames inbound frames totalling bytes encoded bytes. Returns the replenished state and the number of bytes to grant back to the peer in a CREDIT frame.

fully_half_closed?(state)

@spec fully_half_closed?(t()) :: boolean()

True when both directions are half-closed; the terminal frame may still be outstanding.

grant(state)

@spec grant(t()) :: {t(), non_neg_integer()}

The initial (or replenishing) grant this side owes the peer: the unused portion of our receive window. Returns {state, bytes}; bytes is 0 when there is nothing new to grant.

half_close_inbound(state)

@spec half_close_inbound(t()) :: {:ok, t()} | :already_closed

Record the peer's HALF_CLOSE of its sending direction. Idempotent.

half_close_outbound(state)

@spec half_close_outbound(t()) :: {:ok, t()} | :already_closed

Close only this side's sending direction. Idempotent.

new(id, owner, opts)

@spec new(non_neg_integer(), pid(), keyword()) :: t()

A fresh stream with no credit in either direction.

park_sender(state, waiter)

@spec park_sender(t(), term()) :: t()

Park a blocked sender until credit arrives.

pop_sender(state)

@spec pop_sender(t()) :: {:ok, term(), t()} | :empty

Pop the next parked sender, if any.

record_inbound(state, bytes)

@spec record_inbound(t(), pos_integer()) ::
  {:ok, t()} | {:error, :inbound_closed | :credit_exceeded | :frame_limit}

Record one inbound STREAM_DATA frame of bytes encoded bytes. Rejects anything the peer was not entitled to send.

take_senders(state)

@spec take_senders(t()) :: {[term()], t()}

Every parked sender, in order, leaving the queue empty.