Riak postcommit ZeroMQ publisher

riak_zmq.erl

-module(riak_zmq).

-export([postcommit_publish/1, get_zmq_publisher/0, zmq_publisher/0, zmq_publisher/1]).

postcommit_publish(RObj) ->
    Body = riak_object:get_value(RObj),
    {ok, Pid} = get_zmq_publisher(),
    Pid ! {send, Body}.

get_zmq_publisher() ->
    case pg2:get_closest_pid([zmq]) of
        {error, {no_such_group, _}} ->
            pg2:create([zmq]),
            get_zmq_publisher();
        {error, {no_process, _}} ->
            Pid = spawn(?MODULE, zmq_publisher, []),
            pg2:join([zmq], Pid),
            {ok, Pid};
        Pid ->
            {ok, Pid}
    end.

zmq_publisher() ->
    {ok, _Pid} = zmq:start_link(),
    {ok, Socket} = zmq:socket(pub),
    zmq:bind(Socket, "tcp://127.0.0.1:5550"),
    zmq_publisher(Socket).

zmq_publisher(Socket) ->
    receive
        {send, Msg} -> zmq:send(Socket, Msg), zmq_publisher(Socket)
    end.

This is a cool idea (and very little code), but one thought: you should probably start up the zmq publisher as a named process (gen_server, maybe) with a supervisor, rather than using pg2. It feels backwards to have the postcommit ensure that the process is running and publishing on the 0MQ socket.

Is there a way I can start the named process automatically when Riak starts? I'm new to Erlang, so I wasn't sure the best way to do this. I took the pg2 idea from the Riak RabbitMQ postcommit hook (https://github.com/jbrisbin/riak-rabbitmq-commit-hooks). Any guidance here would be appreciated.

@codysoyland Yes, your postcommit hook can make sure the application (and thus its supervisor and worker processes) are started. Rejiggered your code here: https://github.com/seancribbs/riak_zmq

Very cool. thanks. I was looking into how to turn it into an OTP app. Again, totally new to this (I've only dabbled with OTP).

Hi, could you explain please what this is for, for the newbbie I am. I am appealed by riak for availability on one hand, and by 0MQ for ease of use and scallability on the other hand. Is riak_zmq the glue between both ?