Erlang: simple pubsub for processes — is my approach okay?

Question

Disclaimer: I'm pretty new to Erlang and OTP.

I want a simple pubsub in Erlang/OTP, where processes could subscribe at some "hub" and receive a copy of messages that were sent to that hub.

I know about gen_event, but it processes events in one single event manager process, while I want every subscriber to be a separate, autonomous process. Also, I was unable to grok gen_event's handlers supervision. Unfortunately, Google results were full of XMPP (Ejabberd) and RabbitMQ links, so I didn't find anything relevant to my idea.

My idea is that such pubsub model seamlessly maps to supervision tree. So I thought to extend the supervisor (a gen_server under the hood) to be able to send a cast message to all its children.

I've hacked this in my quick-and-dirty custom "dispatcher" behavior:

-module(dispatcher).
-extends(supervisor).
-export([notify/2, start_link/2, start_link/3, handle_cast/2]).

start_link(Mod, Args) ->
    gen_server:start_link(dispatcher, {self, Mod, Args}, []).

start_link(SupName, Mod, Args) ->
    gen_server:start_link(SupName, dispatcher, {SupName, Mod, Args}, []).

notify(Dispatcher, Message) ->
    gen_server:cast(Dispatcher, {message, Message}).

handle_cast({message, Message}, State) ->
    {reply, Children, State} = supervisor:handle_call(which_children, dummy, State),
    Pids = lists:filter(fun(Pid) -> is_pid(Pid) end,
                 lists:map(fun({_Id, Child, _Type, _Modules}) -> Child end,
                           Children)),
    [gen_server:cast(Pid, Message) || Pid <- Pids],
    {noreply, State}.

However, while everything seem to work fine at the first glance (children receive messages and are seamlessly restarted when they fail), I wonder whenever this was a good idea.

Could someone, please, criticize (or approve) my approach, and/or recommend some alternatives?

Are the messages stored and new processes subscribing get the whole history of them. Or is it that messages are passed on only from the point of time when a process subscribed? — Peer Stritzinger
The latter; like in Redis or 0MQ Pub/Sub. I'll take another look at gen_event, thanks. — drdaeman

puzza007 puzza007 · Accepted Answer · 2011-09-10T18:38:00

I've recently used gproc to implement pubsub. The example from the readme does the trick.

subscribe(EventType) ->
    %% Gproc notation: {p, l, Name} means {(p)roperty, (l)ocal, Name}
    gproc:reg({p, l, {?MODULE, EventType}}).

notify(EventType, Msg) ->
    Key = {?MODULE, EventType},
    gproc:send({p, l, Key}, {self(), Key, Msg}).

Erlang: simple pubsub for processes — is my approach okay?

4 Answers