thread communication with ZMQ

threadzmq.c

// gcc -shared -fPIC -I$JULIAHOME/usr/include -L$JULIAHOME/usr/lib -lzmq -luv -o libthreadzmq.so threadzmq.c

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <assert.h>
#include <math.h>
#include <zmq.h>
#include <uv.h>

static void free_buffer(void *data, void *hint) { free(data); }

void produce_messages(void *sock)
{
    int i;
    for (i = 0;; i++) {
        char *buf;
        zmq_msg_t msg;
        int len = asprintf(&buf, "[message #%d from %p]", i, sock);
        assert(zmq_msg_init_data(&msg, buf, len, free_buffer, NULL) == 0);
        assert(zmq_sendmsg(sock, &msg, 0) != -1);
        usleep(lroundl(1e6*drand48()));
    }
    return;
}

void start_pusher(void *sock) {
    uv_thread_t thread;
    uv_thread_create(&thread, produce_messages, sock);
    return;
}

threadzmq.jl

using ZMQ

ctx = Context(1)
pull = Socket(ctx, PULL)
pushers = [ Socket(ctx, PUSH) for _ = 1:10 ]

ZMQ.bind(pull, "inproc://queue")
for pusher in pushers
    ZMQ.connect(pusher, "inproc://queue")
    ccall((:start_pusher, :libthreadzmq), Void, (Ptr{Void},), pusher.data)
end

while true
    println(bytestring(ZMQ.recv(pull)))
end