macula_subscriber behaviour (macula v9.3.0)

View Source

Behaviour for supervised, stateful PubSub consumers.

macula:subscribe_callback/4 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/5,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.

Example

   -module(temperature_logger).
   -behaviour(macula_subscriber).
   -export([init/1, handle_event/4, terminate/2]).
  
   init(_Args) -> {ok, #{count => 0}}.
  
   handle_event(_Topic, #{value := V}, _Meta, State) ->
       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">>, []).

Summary

Functions

Start a subscriber. Subscribes Module to (Realm, Topic) on Pool; Args is passed to Module:init/1.

Callbacks

handle_event/4

-callback handle_event(Topic :: binary(), Payload :: term(), Meta :: map(), State :: term()) ->
                          {noreply, 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

start_link(Module, Pool, Realm, Topic, Args)

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

Start a subscriber. Subscribes Module to (Realm, Topic) on Pool; Args is passed to Module:init/1.

start_link(Module, Pool, Realm, Topic, Args, Opts)

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

As start_link/5, passing Opts through to macula:subscribe/5 (e.g. delivery).