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/2, 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/2 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.

For client_stream mode — a consumer pushing chunks INTO the provider, e.g. a batch upload — export the optional handle_chunk/2 callback (mirroring macula_stream_sink's consumer-side callback exactly) and this module drives the same linked-reader recv/2 loop for you, on the provider side. A server_stream-mode module that doesn't export it is unaffected.

A client_stream provider that also needs to hand the consumer a terminal result (not just accept chunks) exports the optional handle_eof/1 callback: called once, when the consumer's own close_send/1 surfaces here as end-of-stream, in place of the default unconditional {stop, normal, State}. Returning {reply, Result, NewState} sets the stream's terminal reply (macula_stream:set_reply/2 for {ok, Value}, set_error/2 for {error, Reason}) so the consumer's own macula:await_reply/1,2 unblocks with it, before stopping. A module that doesn't export handle_eof/1 keeps the exact prior behavior — no reply is ever set, eof just stops the stream.

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.

Cancel

Stopping this gen_server for any non-normal reason (a crash, the underlying stream dying, Module:handle_open/2/handle_chunk/2 returning a non-normal stop) sends the peer an explicit macula_stream:abort/3 STREAM_ERROR, not just a graceful close — the peer learns the transfer was cancelled/failed instead of mistaking it for an ordinary end-of-stream. A normal stop closes both sides cleanly instead.

Direct-dial

advertise/5,6 registers the handler with the pool's advertise- gossip mechanism only — nothing published lets a caller on another station find this procedure without a route having propagated between the two stations first. advertise_direct/6,7 does that AND publishes a signed procedure_advertisement DHT record naming this pool's currently-connected station as the server — the exact same record type and publish function macula_response:advertise_direct/6,7 uses for plain RPC (a procedure_advertisement does not distinguish RPC from streaming), so a caller using macula_stream_sink:start_link_direct/5,6 can resolve and dial here directly, in one hop, regardless of whether the two stations have a routing edge between them.

Example

   -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_v1">>, log_tailer_provider, self()).
  
   %% elsewhere, once the provider has announced its pid:
   ok = macula_streamer:send(TailerPid, <<"a log line\n">>).

A client_stream-mode provider exports handle_chunk/2 instead, and never calls send/2,3 itself — the consumer is the one pushing:

   -module(batch_upload_provider).
   -behaviour(macula_streamer).
   -export([init/1, handle_open/2, handle_chunk/2]).
  
   init(Parent) -> {ok, {Parent, []}}.
  
   handle_open(_StreamArgs, State) -> {ok, State}.
  
   handle_chunk(Chunk, {Parent, Acc}) ->
       {noreply, {Parent, [Chunk | Acc]}}.

   {ok, _Sup} = macula_streamer:advertise(Pool, Realm,
       <<"bulk.ingest">>, batch_upload_provider, self(),
       #{mode => client_stream}).