minato_listener (minato v0.18.6)

LISTEN on a connection of its own, delivered to subscribing processes.

A notification arrives when the server decides, not when a caller asks, so somebody has to be holding the connection and reading it. That is what this process is. It owns one connection, keeps it under minato_conn:activate/1 so the bytes arrive as messages, and sends what it decodes to whoever subscribed.

ok = minato_listener:listen(events, ~"job_ready"),
receive
    {minato_notification, ~"job_ready", Payload, _Pid} -> Payload
end.

The connection is this process's and nothing else runs on it. LISTEN is session state: a notification is delivered to the session that registered for it, so a pooled connection would deliver to whichever caller happened to hold it next, which is to say nowhere.

Subscribers

Subscribing is per process, and the process is monitored: a subscriber that dies is dropped, and a channel nobody is left listening to is UNLISTENed. Otherwise a supervisor restarting a worker in a loop leaves the server delivering to a session where nothing is waiting.

Notifications get missed

A connection that drops takes its subscriptions with it, and anything the server published while it was gone is gone: NOTIFY is fire and forget, with no replay. This process reconnects with backoff, LISTENs again for every channel that still has a subscriber, and then tells each subscriber {minato_listener, Name, resubscribed}.

That message is the point. A consumer that treats notifications as the only way it learns anything will silently stall; one that is told it was disconnected can go and look. NOTIFY is a hint that something changed, and a poll it makes faster - never the only path.

Summary

Types

info()

name()

How the listener is registered and asked for.

notification()

What a subscriber receives.

opts()

What to connect to.

request()

Functions

channels(Name)

Every channel with a subscriber, and how many each has.

handle_call/3

handle_cast(Message, State)

handle_info/2

init/1

listen(Name, Channel)

Subscribe the calling process to a channel.

start_link(Name, Opts)

Start a listener under a supervisor.

stop(Name)

Stop the listener and close its connection.

terminate/2

unlisten(Name, Channel)

Stop the calling process's subscription.

Types

info()

-type info() ::
          reconnect |
          {'DOWN', reference(), process, pid(), term()} |
          {tcp | ssl, term(), binary()} |
          {tcp_closed | ssl_closed, term()} |
          {tcp_error | ssl_error, term(), term()}.

name()

-type name() :: atom().

How the listener is registered and asked for.

notification()

-type notification() ::
          {minato_notification, Channel :: binary(), Payload :: binary(), From :: integer()}.

What a subscriber receives.

The payload is whatever NOTIFY was given, and the pid is the server side process that sent it, which is how a subscriber tells its own writes from somebody else's.

opts()

-type opts() :: #{connection := minato_conn:opts()}.

What to connect to.

#{connection := minato_conn:opts()}

Passed to minato_conn:connect/1 unchanged. One connection, held for the life of the listener.

request()

-type request() :: {listen, binary(), pid()} | {unlisten, binary(), pid()} | channels.

Functions

channels(Name)

-spec channels(name()) -> #{binary() => non_neg_integer()}.

Every channel with a subscriber, and how many each has.

handle_call/3

-spec handle_call(request(),
                  gen_server:from(),
                  #state{name :: name(),
                         opts :: minato_conn:opts(),
                         conn :: minato_conn:conn() | undefined,
                         channels :: #{binary() => #{pid() => reference()}},
                         backoff :: pos_integer()}) ->
                     {reply,
                      term(),
                      #state{name :: name(),
                             opts :: minato_conn:opts(),
                             conn :: minato_conn:conn() | undefined,
                             channels :: #{binary() => #{pid() => reference()}},
                             backoff :: pos_integer()}}.

handle_cast(Message, State)

-spec handle_cast(term(),
                  #state{name :: name(),
                         opts :: minato_conn:opts(),
                         conn :: minato_conn:conn() | undefined,
                         channels :: #{binary() => #{pid() => reference()}},
                         backoff :: pos_integer()}) ->
                     {noreply,
                      #state{name :: name(),
                             opts :: minato_conn:opts(),
                             conn :: minato_conn:conn() | undefined,
                             channels :: #{binary() => #{pid() => reference()}},
                             backoff :: pos_integer()}}.

handle_info/2

-spec handle_info(info(),
                  #state{name :: name(),
                         opts :: minato_conn:opts(),
                         conn :: minato_conn:conn() | undefined,
                         channels :: #{binary() => #{pid() => reference()}},
                         backoff :: pos_integer()}) ->
                     {noreply,
                      #state{name :: name(),
                             opts :: minato_conn:opts(),
                             conn :: minato_conn:conn() | undefined,
                             channels :: #{binary() => #{pid() => reference()}},
                             backoff :: pos_integer()}}.

init/1

-spec init({name(), opts()}) ->
              {ok,
               #state{name :: name(),
                      opts :: minato_conn:opts(),
                      conn :: minato_conn:conn() | undefined,
                      channels :: #{binary() => #{pid() => reference()}},
                      backoff :: pos_integer()}}.

listen(Name, Channel)

-spec listen(name(), binary()) -> ok | {error, term()}.

Subscribe the calling process to a channel.

Answers ok once the server has acknowledged the LISTEN, so a caller that gets ok is registered rather than about to be. A channel already listened to costs no round trip.

The name is quoted rather than interpolated, so a channel name from outside is a channel name and not a statement.

start_link(Name, Opts)

-spec start_link(name(), opts()) -> gen_server:start_ret().

Start a listener under a supervisor.

stop(Name)

-spec stop(name()) -> ok.

Stop the listener and close its connection.

terminate/2

-spec terminate(term(),
                #state{name :: name(),
                       opts :: minato_conn:opts(),
                       conn :: minato_conn:conn() | undefined,
                       channels :: #{binary() => #{pid() => reference()}},
                       backoff :: pos_integer()}) ->
                   ok.

unlisten(Name, Channel)

-spec unlisten(name(), binary()) -> ok | {error, term()}.

Stop the calling process's subscription.

The channel is UNLISTENed when the last subscriber to it goes.