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.

Cancel

Stopping this gen_server for any non-normal reason (a recv error, the reader crashing, Module:handle_chunk/2 returning a non-normal stop) sends the provider an explicit macula:abort/3 STREAM_ERROR instead of an ordinary close — the provider learns the consumer cancelled or failed rather than mistaking it for a clean end-of-stream. A normal stop closes both sides cleanly instead, same as before.

Direct-dial

start_link/5,6 opens through the pool's existing links — the same gossip-propagated routing call_stream/5 always used. start_link_direct/5,6 is the direct-dial counterpart: it resolves the procedure's procedure_advertisement from the DHT (published by macula_streamer:advertise_direct/6,7 on the provider side) and opens the stream there directly, in one hop, instead of depending on advertise-gossip having propagated a route between arbitrary stations. Requires the provider to have advertised via advertise_direct/6,7, not plain advertise/5,6. See macula_direct_dial's module doc, "Trust model".

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">>, []).