Behaviour for supervised, stateful PubSub consumers.

macula:subscribe_callback/5 is the right tool for a stateless one-shot reaction (log a line, forward to another process): it spawns its own receiver process so a slow callback does not back-pressure the pool, but that process is not part of your supervision tree, has no init/1 for setup, and gives you nowhere to thread state across events.

macula_subscriber is the supervised alternative. Implement three callbacks; start_link/6 returns an ordinary gen_server pid you drop straight into your own supervision tree, and every event arrives as a Module:handle_event/4 call against state your module owns and threads itself.

When the subscription ends

The subscriber stops with the pool's reason on macula_event_gone, and with {pool_down, Reason} when the pool process dies without sending it (killed, or taken down by a link). Both are abnormal exits, so a supervisor restarts it against whatever pool is current. A process that calls macula:subscribe/5 itself should monitor the pool the same way.

When events may have been missed

When the last station link carrying the subscription ends, the optional Module:handle_interrupted/2 is called with #{from_ms, reason}, and when a station accepts it again Module:handle_gap/2 with #{from_ms, to_ms, reason}: events may have been missed between the two times. Resync what the subscription feeds there. A module without them has the notice logged and serves on. See macula_pubsub:subscribe/5 (macula-station#129).

Subscribe function

start_link/6 takes subscribe in its options: the function the subscriber subscribes with, called as Subscribe(Pool, Realm, Topic, self(), Opts) with the other options, and macula:subscribe/5 by default. A test gives its own function this way instead of replacing the macula module.

The other options reach macula:subscribe/5 as they are, so they must name sources -- who may deliver to this subscription (macula#147). There is no default, and a start without it fails with {error, {sources, required}}.

Example

   -module(temperature_logger).
   -behaviour(macula_subscriber).
   -export([init/1, handle_event/4, terminate/2]).
  
   init(_Args) -> {ok, #{count => 0}}.
  
   %% The payload arrives in WIRE FORM (macula_frame:to_wire/1):
   %% map keys are {text, K} tuples, atom values are {text, V},
   %% undefined is null. Normalize with macula_record:decode_payload/1
   %% (or read single fields with macula_record:payload_field/2).
   handle_event(_Topic, Payload, _Meta, State) ->
       V = macula_record:decode_payload(Payload),
       Count = maps:get(count, State) + 1,
       io:format("reading ~p: ~p~n", [Count, V]),
       {noreply, State#{count := Count}}.
  
   terminate(_Reason, _State) -> ok.

   {ok, Pid} = macula_subscriber:start_link(temperature_logger, Pool,
       Realm, <<"sensors.temperature_v1">>, [], #{sources => members}).