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.

The sink's macula_lifetime_announcer publishes these facts, in order, from a process of its own, so a pool that is gone or slow never fails or holds up the sink or its stream; a publish that fails is logged. A sink killed before it could hand over its stream's end has streaming.completed_v1 published for it, with outcome failed and the reason it went down for. An end fact and an abort message name their reason, killed or timeout for example, and carry none of the reason's terms, which go to the local log only.

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

Stream I/O

A sink opens, reads and ends its stream through four functions, call_stream/5, recv/2, close_stream/1 and abort/3, and announces its facts through a fifth, fact_publish/4. They are the macula facade's by default, and a direct-dial sink dials with macula_direct_dial:call_stream/5. start_link/7 and start_link_direct/7 take a stream_io start option, checked by macula_stream:stream_io/2, with the four stream functions and any other macula_stream:stream_io() ones, and a fact_publish start option, to run a sink on something else, such as a test's scripted stream.

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