Macula SDK — Streaming Guide

Streaming RPC over the mesh: server-push, client-push, and bidirectional.

Audience: applications that need more than one request/response — a live feed of frames, an upload of many chunks, or a long-lived two-way exchange. Available since SDK 3.17.

Overview

A plain macula:call/5 is one request and one reply. A streaming RPC keeps the channel open so either side (or both) can send a sequence of chunks before a final result. Each session gets its own dedicated QUIC stream, opened for that session alone — not multiplexed onto the connection's shared control stream the way an ordinary CALL or a PUBLISH is — so QUIC's own per-stream flow control paces the sender to the receiver, and a slow or stalled stream cannot head-of-line-block other RPC/PubSub traffic on the same connection.

Two ways to open one, mirroring unary RPC:

• call_stream/5 — opens on the pool's own healthy link; the station routes the STREAM_OPEN to whichever connection advertised the procedure. Good when you don't know or care which station serves it.
• call_stream_station/6 (direct-dial) — dials a specific station and opens the stream there in one hop, exactly like call_station/6 for unary RPC. Use it after resolving a provider's procedure_advertisement and station_endpoint in the DHT (see the RPC Guide), so a stream reaches its provider the same way a unary call does.

{ok, Stream} = macula:call_stream_station(Pool, StationUrl, Realm, Procedure,
                                          Args, #{}).

Opts may set dial_timeout_ms (default 10_000) for the dial + handshake.

There are three modes:

A stream is a pid(). It is sticky to one link: if that link dies you get {error, peer_down} and re-open.

Consumer side

server_stream — read a feed

{ok, Stream} = macula:call_stream(Pool, Realm, <<"live.feed">>, Request, #{}),
loop(Stream).

loop(Stream) ->
    case macula:recv(Stream) of
        {chunk, Bin}  -> handle(Bin), loop(Stream);   %% raw bytes
        {data, Term}  -> handle(Term), loop(Stream);  %% decoded (msgpack)
        eof           -> ok;                           %% source stopped
        {error, R}    -> {error, R}                    %% e.g. peer_down -> re-open
    end.

recv/1 blocks for the next chunk; recv/2 takes a timeout. eof means the source closed the stream cleanly.

client_stream — push then await the result

{ok, Stream} = macula:call_stream(Pool, Realm, <<"bulk.ingest">>, Meta,
                                  #{mode => client_stream}),
[ok = macula:send(Stream, Chunk) || Chunk <- Chunks],
ok = macula:close_send(Stream),               %% signal "no more input"
{ok, Result} = macula:await_reply(Stream).    %% the provider's final reply

send/2 sends raw bytes; send/3 takes an encoding (raw | msgpack). close_send/1 half-closes your direction; await_reply/1,2 returns the provider's single final result.

Provider side

Advertise a streaming procedure with a mode and a fun(Stream, Args) handler. The handler drives the stream with the same send / recv primitives, and ends it with set_reply (a final result) or abort (an error).

%% server_stream: push N chunks, then CLOSE — that is what produces `eof'
%% for a consumer looping on `recv'. `set_reply' is for client_stream /
%% bidi (see below); a pure push-only server_stream does not use it.
ok = macula:advertise_stream(
       Pool, Realm, <<"live.feed">>, server_stream,
       fun(Stream, _Args) ->
           lists:foreach(fun(Frame) -> macula:send(Stream, Frame) end, frames()),
           macula:close_stream(Stream)
       end),

%% client_stream: drain the consumer's chunks, then reply
ok = macula:advertise_stream(
       Pool, Realm, <<"bulk.ingest">>, client_stream,
       fun(Stream, _Args) ->
           N = drain(Stream, 0),
           macula:set_reply(Stream, #{ingested => N})
       end),

drain(Stream, N) ->
    case macula:recv(Stream) of
        {chunk, Bin} -> store(Bin), drain(Stream, N + 1);
        eof          -> N;
        {error, _}   -> N
    end.

close_stream vs. set_reply — do not mix them for server_stream. close_stream/1 is what makes a consumer's recv loop see eof. set_reply/2 only resolves await_reply/1,2; it does not close the stream. A server_stream handler that calls set_reply without also closing leaves a consumer's recv-until-eof loop waiting forever — use close_stream for a pure push, and reserve set_reply + await_reply for client_stream / bidi, where the consumer already knows to stop sending and ask for the result instead of draining chunks.

Abort with a BOLT#4-style code and message when something goes wrong:

macula:abort(Stream, <<"0F">>, <<"source unavailable">>).

Supervised wrappers: macula_streamer / macula_stream_sink

The bare handler fun above runs in a transient process per inbound STREAM_OPEN, and a consumer has to hand-write its own recv/2 loop around call_stream/5's raw stream pid. macula_streamer and macula_stream_sink wrap the same two primitives as proper OTP behaviours, and each publishes its own streaming.started_v1 / streaming.completed_v1 mesh fact — from its own side's perspective, not deduplicated against the other side's copy.

Provider side — each inbound stream starts one supervised child under a factory supervisor this module owns. Sending is push-based: once handle_open/2 has stashed self() somewhere discoverable (a registry, an ETS table, a message to a known process), any code holding that pid can call macula_streamer:send/2,3 and close/1 on it from outside:

-module(log_tailer_provider).
-behaviour(macula_streamer).
-export([init/1, handle_open/2]).

init(Registry) -> {ok, Registry}.

handle_open(#{<<"topic">> := Topic}, Registry) ->
    Registry ! {tailer_ready, Topic, self()},
    {ok, Registry}.

{ok, _Sup} = macula_streamer:advertise(Pool, Realm, <<"logs.tail">>,
                                       log_tailer_provider, self()).

%% elsewhere, once the provider has announced its pid:
ok = macula_streamer:send(TailerPid, <<"a log line\n">>).

Consumer side — start_link/5,6 opens the stream and drives the recv/2 loop in a linked reader process, delivering each chunk to Module:handle_chunk/2:

-module(log_tailer).
-behaviour(macula_stream_sink).
-export([init/1, handle_chunk/2, handle_close/2]).

init(_Args) -> {ok, []}.

handle_chunk(Line, Lines) ->
    io:format("~s", [Line]),
    {noreply, [Line | Lines]}.

handle_close(_Reason, _Lines) -> ok.

{ok, Pid} = macula_stream_sink:start_link(log_tailer, Pool, Realm,
                                          <<"logs.tail">>, []).

Content streaming

"Content streaming" is the server_stream mode applied to a live source: the provider advertises the stream procedure and its procedure_advertisement in the DHT; a viewer resolves it, dials the serving station directly with call_stream_station (as the diagram shows), and reads frames until the source stops. Unlike content sharing there is no fixed size or chunk_count — the stream is open-ended and ordered, riding its own dedicated QUIC stream (see Overview above), and QUIC's per-stream flow control paces the source to the viewer's consumption without contending with anything else on the connection.

Freshness is not optional. A live source can go away. Treat a recv stall or {error, peer_down} as a signal to re-resolve the source and re-open, exactly as a direct-dial caller re-resolves on a dial failure.

Local (in-process) streams

macula:open_stream/3,4, macula:advertise_stream/2,3, and call_stream/2,3 drive streams inside one BEAM (no mesh), backed by macula_stream_local. They are for unit tests and same-node dispatch. The pool forms (call_stream/5, advertise_stream/5) are the ones that go over the mesh.

Reference