macula_stream_sink behaviour (macula v9.3.1)

View Source

Behaviour 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

handle_chunk/2

-callback handle_chunk(Chunk :: term(), State :: term()) ->
                          {noreply, NewState :: term()} | {stop, Reason :: term(), NewState :: term()}.

handle_close/2

(optional)
-callback handle_close(Reason :: normal | term(), State :: term()) -> any().

init/1

-callback init(Args :: term()) -> {ok, State :: term()} | {stop, Reason :: term()}.

Functions

start_link(Module, Pool, Realm, Procedure, Args)

-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.

start_link(Module, Pool, Realm, Procedure, Args, CallArgs)

-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.