mc_eredis

%%
%% Erlang Redis Pub/Sub client
%%
-module(eredis_sub).
-include("eredis.hrl").
-include("eredis_defaults.hrl").

%% ------------------------------------------------------------------
%% API Function Exports
%% ------------------------------------------------------------------

-export([start_link/0, start_link/1, start_link/3, start_link/4, start_link/6, start_link/7,
         stop/1, controlling_process/1, controlling_process/2, controlling_process/3,
         ack_message/1, subscribe/2, unsubscribe/2, channels/1]).

-export([psubscribe/2, punsubscribe/2]).

-ignore_xref(start_link/0).
-ignore_xref(start_link/1).
-ignore_xref(start_link/3).
-ignore_xref(start_link/4).
-ignore_xref(start_link/6).
-ignore_xref(start_link/7).
-ignore_xref(stop/1).
-ignore_xref(controlling_process/1).
-ignore_xref(controlling_process/2).
-ignore_xref(controlling_process/3).
-ignore_xref(ack_message/1).
-ignore_xref(subscribe/2).
-ignore_xref(unsubscribe/2).
-ignore_xref(channels/1).
-ignore_xref(psubscribe/2).
-ignore_xref(punsubscribe/2).

-ifdef(TEST).
-export([receiver/1, sub_example/0, pub_example/0]).

-export([psub_example/0, ppub_example/0]).

-ignore_xref(sub_example/0).
-ignore_xref(pub_example/0).
-ignore_xref(psub_example/0).
-ignore_xref(ppub_example/0).
-endif.

%% ------------------------------------------------------------------
%% Macro Definitions
%% ------------------------------------------------------------------

-define(DEFAULT_MAX_QUEUE_SIZE, infinity).
-define(DEFAULT_QUEUE_BEHAVIOUR, drop).

%% ------------------------------------------------------------------
%% Type Definitions
%% ------------------------------------------------------------------

-type sub_option() ::
        eredis:option() |
        {max_queue_size, non_neg_integer() | infinity} |
        {queue_behaviour, drop | exit}.
-export_type([sub_option/0]).

-type sub_args() :: [sub_option()].
-export_type([sub_args/0]).

-type channel() :: binary().
-export_type([channel/0]).

-type eredis_queue() :: queue:queue().
-export_type([eredis_queue/0]).

%% ------------------------------------------------------------------
%% API Function Definitions
%% ------------------------------------------------------------------

start_link() ->
    start_link([]).

start_link(Host, Port, Password) ->
    start_link([{host, Host},
                {port, Port},
                {password, Password}]).

start_link(Transport, Host, Port, Password) ->
    start_link([{transport, Transport},
                {host, Host},
                {port, Port},
                {password, Password}]).

start_link(Host, Port, Password, ReconnectSleep,
           MaxQueueSize, QueueBehaviour) ->
    start_link([{host, Host},
                {port, Port},
                {password, Password},
                {reconnect_sleep, ReconnectSleep},
                {max_queue_size, MaxQueueSize},
                {queue_behaviour, QueueBehaviour}]).

start_link(Transport, Host, Port, Password, ReconnectSleep,
           MaxQueueSize, QueueBehaviour)
  when is_atom(Transport) andalso
       is_list(Host) andalso
       is_integer(Port) andalso
       is_list(Password) andalso
       (is_integer(ReconnectSleep) orelse ReconnectSleep =:= no_reconnect) andalso
       (is_integer(MaxQueueSize) orelse MaxQueueSize =:= infinity) andalso
       (QueueBehaviour =:= drop orelse QueueBehaviour =:= exit) ->

    eredis_sub_client:start_link(Transport, Host, Port, Password, ReconnectSleep,
                                 MaxQueueSize, QueueBehaviour).

%% @doc Callback for starting from poolboy
-spec start_link(sub_args()) -> {ok, Pid::pid()} | {error, Reason::term()}.
start_link(Args) ->
    Transport      = proplists:get_value(transport, Args, ?DEFAULT_TRANSPORT),
    Host           = proplists:get_value(host, Args, ?DEFAULT_HOST),
    Port           = proplists:get_value(port, Args, ?DEFAULT_PORT(Transport)),
    Password       = proplists:get_value(password, Args, ?DEFAULT_PASSWORD),
    ReconnectSleep = proplists:get_value(reconnect_sleep, Args, ?DEFAULT_RECONNECT_SLEEP),
    MaxQueueSize   = proplists:get_value(max_queue_size, Args, ?DEFAULT_MAX_QUEUE_SIZE),
    QueueBehaviour = proplists:get_value(queue_behaviour, Args, ?DEFAULT_QUEUE_BEHAVIOUR),
    start_link(Transport, Host, Port, Password, ReconnectSleep,
               MaxQueueSize, QueueBehaviour).

stop(Pid) ->
    eredis_sub_client:stop(Pid).

-spec controlling_process(Client::pid()) -> ok.
%% @doc Make the calling process the controlling process. The
%% controlling process received pubsub-related messages, of which
%% there are three kinds. In each message, the pid refers to the
%% eredis client process.
%%
%%   {message, Channel::binary(), Message::binary(), pid()}
%%     This is sent for each pubsub message received by the client.
%%
%%   {pmessage, Pattern::binary(), Channel::binary(), Message::binary(), pid()}
%%     This is sent for each pattern pubsub message received by the client.
%%
%%   {dropped, NumMessages::integer(), pid()}
%%     If the queue reaches the max size as specified in start_link
%%     and the behaviour is to drop messages, this message is sent when
%%     the queue is flushed.
%%
%%   {subscribed, Channel::binary(), pid()}
%%     When using eredis_sub:subscribe(pid()), this message will be
%%     sent for each channel Redis aknowledges the subscription. The
%%     opposite, 'unsubscribed' is sent when Redis aknowledges removal
%%     of a subscription.
%%
%%   {eredis_disconnected, pid()}
%%     This is sent when the eredis client is disconnected from redis.
%%
%%   {eredis_connected, pid()}
%%     This is sent when the eredis client reconnects to redis after
%%     an existing connection was disconnected.
%%
%% Any message of the form {message, _, _, _} must be acknowledged
%% before any subsequent message of the same form is sent. This
%% prevents the controlling process from being overrun with redis
%% pubsub messages. See ack_message/1.
controlling_process(Client) ->
    controlling_process(Client, self()).

-spec controlling_process(Client::pid(), Pid::pid()) -> ok.
%% @doc Make the given process (pid) the controlling process.
controlling_process(Client, Pid) ->
    controlling_process(Client, Pid, ?TIMEOUT).

%% @doc Make the given process (pid) the controlling process subscriber
%% with the given Timeout.
controlling_process(Client, Pid, Timeout) ->
    gen_server:call(Client, {controlling_process, Pid}, Timeout).

-spec ack_message(Client::pid()) -> ok.
%% @doc acknowledge the receipt of a pubsub message. each pubsub
%% message must be acknowledged before the next one is received
ack_message(Client) ->
    gen_server:cast(Client, {ack_message, self()}).

%% @doc Subscribe to the given channels. Returns immediately. The
%% result will be delivered to the controlling process as any other
%% message. Delivers {subscribed, Channel::binary(), pid()}
-spec subscribe(pid(), [channel()]) -> ok.
subscribe(Client, Channels) ->
    gen_server:cast(Client, {subscribe, self(), Channels}).

%% @doc Pattern subscribe to the given channels. Returns immediately. The
%% result will be delivered to the controlling process as any other
%% message. Delivers {subscribed, Channel::binary(), pid()}
-spec psubscribe(pid(), [channel()]) -> ok.
psubscribe(Client, Channels) ->
    gen_server:cast(Client, {psubscribe, self(), Channels}).

unsubscribe(Client, Channels) ->
    gen_server:cast(Client, {unsubscribe, self(), Channels}).

punsubscribe(Client, Channels) ->
    gen_server:cast(Client, {punsubscribe, self(), Channels}).

%% @doc Returns the channels the given client is currently
%% subscribing to. Note: this list is based on the channels at startup
%% and any channel added during runtime. It might not immediately
%% reflect the channels Redis thinks the client is subscribed to.
channels(Client) ->
    gen_server:call(Client, get_channels).

%% ------------------------------------------------------------------
%% "Internal" Function Definitions
%% ------------------------------------------------------------------

-ifdef(TEST).
receiver(Sub) ->
    receive
        _Msg ->
            ack_message(Sub),
            ?MODULE:receiver(Sub)
    end.

sub_example() ->
    {ok, Sub} = start_link(),
    Receiver = spawn_link(fun () ->
                                  controlling_process(Sub),
                                  subscribe(Sub, [<<"foo">>]),
                                  receiver(Sub)
                          end),
    {Sub, Receiver}.

psub_example() ->
    {ok, Sub} = start_link(),
    Receiver = spawn_link(fun () ->
                                  controlling_process(Sub),
                                  psubscribe(Sub, [<<"foo*">>]),
                                  receiver(Sub)
                          end),
    {Sub, Receiver}.

pub_example() ->
    {ok, P} = eredis:start_link(),
    {ok, <<_/binary>>} = eredis:q(P, ["PUBLISH", "foo", "bar"]),
    eredis_client:stop(P).

ppub_example() ->
    {ok, P} = eredis:start_link(),
    {ok, <<_/binary>>} = eredis:q(P, ["PUBLISH", "foo123", "bar"]),
    eredis_client:stop(P).
-endif.