macula_stream (macula v11.1.0)

View Source

Macula streaming RPC — single-stream state machine.

Owns one streaming RPC's state. Each call_stream, open_stream, or server-side handler invocation gets its own macula_stream gen_server. The state machine itself is carrier-agnostic; the peer shape ({local, _} or {remote_via_link, _, _}) decides how chunks reach the wire.

Two carriers route through forward_to_peer/2:

  • {local, Pid} — in-process pairing for unit tests and macula_stream_local dispatch.
  • {remote_via_link, Link, Sid}: frames over a peering connection through macula_station_link. The stream signs and numbers its own frames with the node identity key its start loader returns, from 0 across STREAM_DATA, STREAM_END, STREAM_ERROR and STREAM_REPLY, and hands the link their bytes. It verifies each frame from the peer against its STREAM_OPEN before the frame takes effect, and reports a refused frame to its peering connection.

Renamed from macula_stream_v1 in 3.17.0; the V1 mesh_client carrier ({remote, _, _}) was retired alongside the rest of the V1 surface in the same release. The module now spans the LOCAL carrier and the V2 station_link carrier only.

Summary

Types

chunk/0
encoding/0
mode/0
peer/0
result/0
role/0
stream_id/0
stream_io/0

Functions

abort(Pid, Code, Message)

Abort the stream with a STREAM_ERROR frame. Both sides close; any pending recv/await_reply waiters receive {error, {Code, Message}}. A code a STREAM_ERROR cannot carry, over 64 bytes or not UTF-8, still aborts the stream, with the code aborted, and the caller gets the code's refusal by name, so a call to abort always stops the stream.

attach_to_link(StreamPid, LinkPid, StreamId)

Attach a macula_station_link peer to this stream. The stream hands the link the bytes of each frame it signs, and the link forwards the peer's STREAM_* frames into it through deliver_frame/2.

await_reply(Pid)

Wait for the terminal reply (client-stream / bidi).

await_reply(Pid, Timeout)
close(Pid)

Close both sides. Idempotent.

close_send(Pid)

Half-close the write side. Recv side stays open.

controlling_process(Pid, NewOwner)

Hand the stream to NewOwner. A stream ends when its owner ends; after this it ends when NewOwner does, and NewOwner is told when the session ends, as {macula_stream, ended, Stream, How}, or at once if it already has. Only the stream's current owner can hand it over; any other caller gets {error, not_owner} and the stream stays with its owner.

deliver_chunk(Pid, Encoding, Body)

Deliver a chunk frame from the peer.

deliver_end(Pid, Role)

Deliver a STREAM_END frame from the peer.

deliver_error(Pid, Code, Message)

Deliver a STREAM_ERROR frame from the peer.

deliver_frame(Pid, Frame)

Deliver a STREAM_DATA, STREAM_END, STREAM_ERROR or STREAM_REPLY frame of a link-carried stream from the peer. The stream verifies it against its STREAM_OPEN before it takes effect. A refused frame is dropped and reported to the stream's peering connection, and the stream carries on. A frame of a type that belongs on the control stream rejects the connection with malformed_frame and ends the stream, in either profile.

deliver_reply(Pid, Result)

Deliver a STREAM_REPLY frame from the peer.

format_status(Status)
handle_call(Msg, From, State)
handle_cast(Msg, State)
handle_info(Msg, State)
info(Pid)

Inspect stream state (debugging).

init(Opts)
pair(A, B)

Pair two stream processes as peers (Phase 1 local dispatch).

recv(Pid)

Receive the next chunk (blocks indefinitely).

recv(Pid, Timeout)
send(Pid, Bin)

Send a binary chunk on the stream.

send(Pid, Body, _)
set_error(Pid, Reason)

Server-side: emit a terminal error as the reply value.

set_reply(Pid, Result)

Server-side: emit the terminal reply.

start_link(Opts)

Start a stream gen_server.

stream_io(Defaults, Given)

The stream functions a wrapper runs on. Defaults are the functions the wrapper calls, by key, and Given the stream_io its caller gave, or undefined for none, which gives Defaults. A given set has every key in Defaults, each function at the arity its key takes, and may carry other stream_io() functions; any other is refused with function_clause, in the calling process.

terminate(Reason, State)

Types

chunk/0

-type chunk() :: binary() | {raw, binary()} | {term, term()}.

encoding/0

-type encoding() :: raw | msgpack.

mode/0

-type mode() :: server_stream | client_stream | bidi.

peer/0

-type peer() :: undefined | {local, pid()} | {remote_via_link, pid(), stream_id()}.

result/0

-type result() :: {ok, term()} | {error, term()}.

role/0

-type role() :: client | server.

stream_id/0

-type stream_id() :: binary().

stream_io/0

-type stream_io() ::
          #{call_stream =>
                fun((macula:pool(), macula:realm(), macula:procedure(), term(), map()) ->
                        {ok, pid()} | {error, term()}),
            recv =>
                fun((pid(), timeout()) -> {chunk, binary()} | {data, term()} | eof | {error, term()}),
            send => fun((pid(), binary() | term(), encoding()) -> ok | {error, term()}),
            close_send => fun((pid()) -> term()),
            close => fun((pid()) -> term()),
            close_stream => fun((pid()) -> term()),
            abort => fun((pid(), binary(), binary()) -> term()),
            set_reply => fun((pid(), term()) -> term()),
            set_error => fun((pid(), term()) -> term()),
            await_reply => fun((pid()) -> result()),
            controlling_process => fun((pid(), pid()) -> ok | {error, not_owner})}.

Functions

abort(Pid, Code, Message)

-spec abort(pid(), binary(), binary()) -> ok | {error, {text_too_long | invalid_text, code}}.

Abort the stream with a STREAM_ERROR frame. Both sides close; any pending recv/await_reply waiters receive {error, {Code, Message}}. A code a STREAM_ERROR cannot carry, over 64 bytes or not UTF-8, still aborts the stream, with the code aborted, and the caller gets the code's refusal by name, so a call to abort always stops the stream.

attach_to_link(StreamPid, LinkPid, StreamId)

-spec attach_to_link(pid(), pid(), stream_id()) -> ok.

Attach a macula_station_link peer to this stream. The stream hands the link the bytes of each frame it signs, and the link forwards the peer's STREAM_* frames into it through deliver_frame/2.

await_reply(Pid)

-spec await_reply(pid()) -> result().

Wait for the terminal reply (client-stream / bidi).

await_reply(Pid, Timeout)

-spec await_reply(pid(), timeout()) -> result() | {error, timeout}.

close(Pid)

-spec close(pid()) -> ok.

Close both sides. Idempotent.

close_send(Pid)

-spec close_send(pid()) -> ok.

Half-close the write side. Recv side stays open.

controlling_process(Pid, NewOwner)

-spec controlling_process(pid(), pid()) -> ok | {error, not_owner}.

Hand the stream to NewOwner. A stream ends when its owner ends; after this it ends when NewOwner does, and NewOwner is told when the session ends, as {macula_stream, ended, Stream, How}, or at once if it already has. Only the stream's current owner can hand it over; any other caller gets {error, not_owner} and the stream stays with its owner.

deliver_chunk(Pid, Encoding, Body)

-spec deliver_chunk(pid(), encoding(), term()) -> ok.

Deliver a chunk frame from the peer.

deliver_end(Pid, Role)

-spec deliver_end(pid(), send | both) -> ok.

Deliver a STREAM_END frame from the peer.

deliver_error(Pid, Code, Message)

-spec deliver_error(pid(), binary(), binary()) -> ok.

Deliver a STREAM_ERROR frame from the peer.

deliver_frame(Pid, Frame)

-spec deliver_frame(pid(), macula_frame:frame()) -> ok.

Deliver a STREAM_DATA, STREAM_END, STREAM_ERROR or STREAM_REPLY frame of a link-carried stream from the peer. The stream verifies it against its STREAM_OPEN before it takes effect. A refused frame is dropped and reported to the stream's peering connection, and the stream carries on. A frame of a type that belongs on the control stream rejects the connection with malformed_frame and ends the stream, in either profile.

deliver_reply(Pid, Result)

-spec deliver_reply(pid(), result()) -> ok.

Deliver a STREAM_REPLY frame from the peer.

format_status(Status)

handle_call(Msg, From, State)

handle_cast(Msg, State)

handle_info(Msg, State)

info(Pid)

-spec info(pid()) -> map().

Inspect stream state (debugging).

init(Opts)

pair(A, B)

-spec pair(pid(), pid()) -> ok.

Pair two stream processes as peers (Phase 1 local dispatch).

recv(Pid)

-spec recv(pid()) -> {chunk, binary()} | {data, term()} | eof | {error, term()}.

Receive the next chunk (blocks indefinitely).

recv(Pid, Timeout)

-spec recv(pid(), timeout()) -> {chunk, binary()} | {data, term()} | eof | {error, term()}.

send(Pid, Bin)

-spec send(pid(), binary()) -> ok | {error, term()}.

Send a binary chunk on the stream.

send(Pid, Body, _)

-spec send(pid(), binary() | term(), encoding()) -> ok | {error, term()}.

set_error(Pid, Reason)

-spec set_error(pid(), term()) -> ok.

Server-side: emit a terminal error as the reply value.

set_reply(Pid, Result)

-spec set_reply(pid(), term()) -> ok.

Server-side: emit the terminal reply.

start_link(Opts)

-spec start_link(map()) -> {ok, pid()} | {error, term()}.

Start a stream gen_server.

Required opts: id, role, mode, owner. Optional: max_inbox_bytes, the bytes of chunks no reader has taken that the stream keeps; a chunk past them ends the session (default 16 MiB). A stream carried by a macula_station_link also takes key (a function that returns the node identity key it signs with, called each time it signs, so the stream never holds the key), open (the verified STREAM_OPEN), conn (the peering connection that carries it) and profile. A stream given the key itself as key does not start, and start_link returns {error, {key, not_a_loader}}.

stream_io(Defaults, Given)

-spec stream_io(stream_io(), stream_io() | undefined) -> stream_io().

The stream functions a wrapper runs on. Defaults are the functions the wrapper calls, by key, and Given the stream_io its caller gave, or undefined for none, which gives Defaults. A given set has every key in Defaults, each function at the arity its key takes, and may carry other stream_io() functions; any other is refused with function_clause, in the calling process.

terminate(Reason, State)