czmq reactor

#include "czmq.h"
#include <iostream>
#include <string.h>

int check_interrupt(zsock_t *pipe) {
    int interrupt = 0;
    char *msg = zstr_recv(pipe);

    if (zsys_interrupted || (msg && !strcmp(msg, "$TERM")))
        interrupt = -1;

    free(msg);
    return interrupt;
}

static int
timer_handler(zloop_t *loop, int timer_id, void *pipe)
{
    std::cout << "timer_handler: time to do something.\n";
    zstr_send(pipe, "Timer says go!");
    return 0;
}

static void
timer_thread(zsock_t *pipe, void *args) {
    zsock_signal(pipe, 0);
    std::cout << "timer_thread: starting up\n";
    while(1) {
        char *str = zstr_recv(pipe);
        std::cout << "timer_thread received: " << str << std::endl;
        free(str);
        if (check_interrupt(pipe))
            break;
        // request list from socket_thread
        zstr_send(args, "list request");
        // recv list and do stuff
    }
}

static int
socket_handler(zloop_t *loop, zsock_t *handle, void *arg) {
    int interrupt = 0;
    zmsg_t *msg = zmsg_new();
    assert(msg);
    msg = zmsg_recv(handle);
    std::cout << "socket_handler: recvd socket msg\n";
    if (zsys_interrupted || (msg && !strcmp(zmsg_popstr(msg), "$TERM")))
        interrupt = -1;
    zstr_send(arg, "socket-related msg");
    return interrupt;
}

static void
socket_thread(zsock_t *pipe, void *args) {
    zsock_signal(pipe, 0);
    std::cout << "socket_thread: starting up\n";
    while(1) {
        char *str = zstr_recv(pipe);
        std::cout << "socket_thread: received: " << str << std::endl;
        free(str);
        if (check_interrupt(pipe))
            break;
        // Parse the msg, either add msg to list or return list
    }
}



int main(int argc, char *argv[]) {
    zactor_t *socket_zactor = zactor_new(socket_thread, NULL);
    zactor_t *timer_zactor = zactor_new(timer_thread, socket_zactor);

    // Set up our socket
    char *endpoint;
    asprintf(&endpoint, "tcp://%s:%d", "localhost", 5555);
    zsock_t *a_zsock = zsock_new_sub(endpoint, "");
    assert(a_zsock);
    free(endpoint);

    zloop_t *loop = zloop_new();
    assert(loop);
    zloop_set_verbose(loop, true);

    int timer_id = zloop_timer(loop, 10000, 0, timer_handler, timer_zactor);
    assert(timer_id != -1);

    int rc = zloop_reader(loop, a_zsock, socket_handler, socket_zactor);
    assert(rc == 0);

    zloop_reader_set_tolerant(loop, a_zsock);
    while (!zsys_interrupted) {
        if (zloop_start(loop) != 0)
            break;
    }

    zactor_destroy(&socket_zactor);
    zactor_destroy(&timer_zactor);
    zsock_destroy(&a_zsock);
    zloop_destroy(&loop);
    assert (loop == NULL);
}