macula_stream_sink behaviour (macula v9.3.0)
View SourceBehaviour for supervised streaming RPC consumers.
call_stream/5 hands back a raw stream pid; a real consumer has to hand-write a recv/2 loop around it — the provider side already gets this for free via advertise_stream/5's callback handler, this is the missing consumer-side half. macula_stream_sink opens the stream for you, drives the recv/2 loop in a linked reader process (so a slow or stuck recv never blocks your gen_server's own mailbox), and calls Module:handle_chunk/2 once per item against state your module owns, Module:handle_close/2 when the stream ends or errors.
This is the general-purpose RPC streaming feature (call_stream/5, e.g. a logs.tail_v1-style procedure) — unrelated to content sharing's own chunked-transfer protocol; see macula_feeder / macula_download for that.
Publishes streaming.started_v1 / streaming.completed_v1 mesh facts around the stream's lifetime, from the consumer's own perspective — the provider side (macula_streamer) publishes its own copy from its perspective; the two are not deduplicated, mirroring how macula_feeder / macula_download each announce their own side of a content transfer.
Example
-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_v1">>, []).
Summary
Functions
Start a sink. Opens a stream to Procedure on (Realm) via Pool and passes Args to Module:init/1.
As start_link/5, with CallArgs passed to call_stream/5 as the RPC argument payload.
Callbacks
Functions
-spec start_link(module(), macula:pool(), macula:realm(), macula:procedure(), term()) -> {ok, pid()} | {error, term()}.
Start a sink. Opens a stream to Procedure on (Realm) via Pool and passes Args to Module:init/1.
-spec start_link(module(), macula:pool(), macula:realm(), macula:procedure(), term(), term()) -> {ok, pid()} | {error, term()}.
As start_link/5, with CallArgs passed to call_stream/5 as the RPC argument payload.