macula_streamer behaviour (macula v9.3.1)

View Source

Behaviour for supervised streaming RPC providers.

advertise_stream/5 on the raw SDK takes a bare handler fun invoked as Handler(StreamPid, Args) in a transient process spawned per inbound STREAM_OPEN (see the internal macula_station_link advertise_stream/5 — "this link spawns a server-side macula_stream and dispatches Handler(StreamPid, Args) in a transient process"). This is the provider-side counterpart to macula_stream_sink: each inbound stream starts one supervised macula_streamer child (under a simple_one_for_one factory this module owns), threading state through Module:init/1 and Module:handle_open/3, and publishing streaming.started_v1 / streaming.completed_v1 mesh facts around the stream's lifetime.

Sending is push-based and driven from outside the callback: once Module:handle_open/3 has done whatever registration it needs (e.g. stashing self() in a registry keyed by some connection id), any process holding this streamer's pid can call send/2,3 / close/1 on it. This module does not prescribe the discovery mechanism.

This is the general-purpose RPC streaming feature (call_stream/5, advertise_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.

Example

   -module(log_tailer_provider).
   -behaviour(macula_streamer).
   -export([init/1, handle_open/3]).
  
   init(Registry) -> {ok, Registry}.
  
   handle_open(#{topic := Topic}, Registry, State) ->
       Registry ! {tailer_ready, Topic, self()},
       {ok, State}.
   {ok, _Sup} = macula_streamer:advertise(Pool, Realm,
       <<"logs.tail_v1">>, log_tailer_provider, self()).
  
   %% elsewhere, once the provider has announced its pid:
   ok = macula_streamer:send(TailerPid, <<"a log line\n">>).

Summary

Functions

Advertise Procedure on Pool/Realm. Starts a private factory supervisor for per-stream provider children and registers a dispatch handler with macula:advertise_stream/5. Returns the supervisor pid so the caller can supervise it (or ignore it).

As advertise/5. Opts may include announce (default true) and mode (default server_stream).

Close the send side of the stream.

Send a chunk out on the stream this streamer owns.

As send/2, with an explicit encoding.

Stop advertising. Does not stop the factory supervisor returned by advertise/5,6 — callers that want to tear it down should exit(Sup, shutdown) themselves.

Callbacks

handle_open/2

-callback handle_open(StreamArgs :: term(), State :: term()) ->
                         {ok, NewState :: term()} | {stop, Reason :: term(), NewState :: term()}.

init/1

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

terminate/2

(optional)
-callback terminate(Reason :: term(), State :: term()) -> any().

Functions

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

-spec advertise(macula:pool(), macula:realm(), macula:procedure(), module(), term()) ->
                   {ok, pid()} | {error, term()}.

Advertise Procedure on Pool/Realm. Starts a private factory supervisor for per-stream provider children and registers a dispatch handler with macula:advertise_stream/5. Returns the supervisor pid so the caller can supervise it (or ignore it).

advertise(Pool, Realm, Procedure, Module, Args, Opts)

-spec advertise(macula:pool(), macula:realm(), macula:procedure(), module(), term(), map()) ->
                   {ok, pid()} | {error, term()}.

As advertise/5. Opts may include announce (default true) and mode (default server_stream).

close(Pid)

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

Close the send side of the stream.

send(Pid, Chunk)

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

Send a chunk out on the stream this streamer owns.

send(Pid, Chunk, Encoding)

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

As send/2, with an explicit encoding.

unadvertise(Pool, Realm, Procedure)

-spec unadvertise(macula:pool(), macula:realm(), macula:procedure()) -> ok.

Stop advertising. Does not stop the factory supervisor returned by advertise/5,6 — callers that want to tear it down should exit(Sup, shutdown) themselves.