自学教程：C++ zctx_new函数代码示例

int main (int argc, char** argv){    int port = 5556;#ifdef FF_USE_LIBGC    GC_INIT();    set_program_name (argv[0]);#endif    if( argc >= 2 )	port = atoi( argv[1] );	    clonesrv_t *self = (clonesrv_t *) zmalloc (sizeof (clonesrv_t));    self->port = port;    self->ctx = zctx_new ();    self->kvmap = zhash_new ();    self->loop = zloop_new ();    zloop_set_verbose (self->loop, false);    //  Set up our clone server sockets    self->snapshot  = zsocket_new (self->ctx, ZMQ_ROUTER);    zsocket_bind (self->snapshot,  "tcp://*:%d", self->port + socket_srv_offset_snapshot );    self->publisher = zsocket_new (self->ctx, ZMQ_PUB);    zsocket_bind (self->publisher, "tcp://*:%d", self->port + socket_srv_offset_publisher );    self->collector = zsocket_new (self->ctx, ZMQ_PULL);    zsocket_bind (self->collector, "tcp://*:%d", self->port + socket_srv_offset_collector );    self->ping = zsocket_new (self->ctx, ZMQ_REP);    zsocket_bind (self->ping,      "tcp://*:%d", self->port + socket_srv_offset_ping );    //  Register our handlers with reactor    zmq_pollitem_t poller = { 0, 0, ZMQ_POLLIN };    poller.socket = self->snapshot;    zloop_poller (self->loop, &poller, s_snapshots, self);    poller.socket = self->collector;    zloop_poller (self->loop, &poller, s_collector, self);    zloop_timer (self->loop, 1000, 0, s_flush_ttl, self);    poller.socket = self->ping;    zloop_poller (self->loop, &poller, s_ping, self);            DEBUG ("I: server up and running on port:%d...", port );    //  Run reactor until process interrupted    zloop_start (self->loop);    zloop_destroy (&self->loop);    zhash_destroy (&self->kvmap);    zctx_destroy (&self->ctx);    free (self);    return 0;}

intzctx_test (Bool verbose){    printf (" * zctx: ");    //  @selftest    //  Create and destroy a context without using it    zctx_t *ctx = zctx_new ();    assert (ctx);    zctx_destroy (&ctx);    assert (ctx == NULL);    //  Create a context with many busy sockets, destroy it    ctx = zctx_new ();    assert (ctx);    zctx_set_iothreads (ctx, 1);    zctx_set_linger (ctx, 5);       //  5 msecs    void *s1 = zctx__socket_new (ctx, ZMQ_PAIR);    void *s2 = zctx__socket_new (ctx, ZMQ_XREQ);    void *s3 = zctx__socket_new (ctx, ZMQ_REQ);    void *s4 = zctx__socket_new (ctx, ZMQ_REP);    void *s5 = zctx__socket_new (ctx, ZMQ_PUB);    void *s6 = zctx__socket_new (ctx, ZMQ_SUB);    zsocket_connect (s1, "tcp://127.0.0.1:5555");    zsocket_connect (s2, "tcp://127.0.0.1:5555");    zsocket_connect (s3, "tcp://127.0.0.1:5555");    zsocket_connect (s4, "tcp://127.0.0.1:5555");    zsocket_connect (s5, "tcp://127.0.0.1:5555");    zsocket_connect (s6, "tcp://127.0.0.1:5555");    assert (zctx_underlying (ctx));    //  Everything should be cleanly closed now    zctx_destroy (&ctx);    //  @end    printf ("OK/n");    return 0;}

int main (void){    //  Start child threads    zctx_t *ctx = zctx_new ();    zthread_fork (ctx, server_thread, NULL);    void *client =    zthread_fork (ctx, client_thread, NULL);    //  Loop until client tells us it's done    char *string = zstr_recv (client);    free (string);    //  Kill server thread    zctx_destroy (&ctx);    return 0;}

static broker_t *s_broker_new(int verbose){	broker_t *self = (broker_t *)zmalloc(sizeof(broker_t));	self->ctx = zctx_new();	self->socket = zsocket_new(self->ctx, ZMQ_ROUTER);	self->verbose = verbose;	self->services = zhash_new();	self->workers = zhash_new();	self->waiting = zlist_new();	self->heartbeat_at = zclock_time() + HEARTBEAT_INTERVAL;	return self;}

mdp_client_t *mdp_client_new (char *broker, int verbose){    assert (broker);    mdp_client_t *self = (mdp_client_t *) zmalloc (sizeof (mdp_client_t));    self->ctx = zctx_new ();    self->broker = strdup (broker);    self->verbose = verbose;    self->timeout = 2500;           //  msecs    s_mdp_client_connect_to_broker (self);    return self;}

zctx_t *new_context(){    zctx_t *ctx;    ctx = zctx_new();    assert(ctx != NULL);    zctx_set_linger(ctx, 0);    //** Disable the CZMQ SIGINT/SIGTERM signale handler    apr_signal(SIGINT, NULL);    apr_signal(SIGTERM, NULL);    return(ctx);}

static void *worker_task (void *args){    zctx_t *ctx = zctx_new ();    void *worker = zsocket_new (ctx, ZMQ_DEALER);    zsocket_connect (worker, "tcp://localhost:5556");        while (true) {        zmsg_t *msg = zmsg_recv (worker);        zmsg_send (&msg, worker);    }    zctx_destroy (&ctx);    return NULL;}

int main(int argc, char *argv[]){    char *bind = strdup("tcp://127.0.0.1:12345");    size_t msg_size = 1024;    int msg_count = 100;    int perf_type = THROUGHPUT;    int opt;    while ((opt = getopt(argc, argv, "b:s:c:t:h")) != -1) {        switch (opt) {        case 'b':            free(bind);            bind = strdup(optarg);            break;        case 's':            msg_size = atoi(optarg);            break;        case 'c':            msg_count = atoi(optarg);            break;        case 't':            perf_type = atoi(optarg);            break;        case 'h':            usage();            exit(EXIT_SUCCESS);        default:            fprintf(stderr, "Try 'nanomsg_req_perf -h' for more information/n");            exit(EXIT_FAILURE);        }    }    thr_info *info = (thr_info *) malloc(sizeof(thr_info));    info->bind_to = bind;    info->msg_size = msg_size + 1;    info->msg_count = msg_count;    info->perf_type = perf_type;    zthread_new(server_task, info);    zctx_t *ctx = zctx_new();    void *client = zthread_fork(ctx, client_task, info);    char *signal = zstr_recv(client);    assert(strcmp(signal, "done") == 0);    free(signal);    zctx_destroy(&ctx);    return 0;}

voidzmonitor_test (bool verbose){    printf (" * zmonitor: ");    if (verbose)        printf ("/n");    //  @selftest    zctx_t *ctx = zctx_new ();    bool result;    void *sink = zsocket_new (ctx, ZMQ_PULL);    zmonitor_t *sinkmon = zmonitor_new (ctx,        sink, ZMQ_EVENT_LISTENING | ZMQ_EVENT_ACCEPTED);    zmonitor_set_verbose (sinkmon, verbose);    //  Check sink is now listening    int port_nbr = zsocket_bind (sink, "tcp://127.0.0.1:*");    assert (port_nbr != -1);    result = s_check_event (sinkmon, ZMQ_EVENT_LISTENING);    assert (result);    void *source = zsocket_new (ctx, ZMQ_PUSH);    zmonitor_t *sourcemon = zmonitor_new (ctx,        source, ZMQ_EVENT_CONNECTED | ZMQ_EVENT_DISCONNECTED);    zmonitor_set_verbose (sourcemon, verbose);    zsocket_connect (source, "tcp://127.0.0.1:%d", port_nbr);    //  Check source connected to sink    result = s_check_event (sourcemon, ZMQ_EVENT_CONNECTED);    assert (result);    //  Check sink accepted connection    result = s_check_event (sinkmon, ZMQ_EVENT_ACCEPTED);    assert (result);    //  Destroy sink to trigger a disconnect event on the source    //  PH: disabled since this causes an access violation in    //  zmonitor_destroy as the socket is no longer valid.//     zsocket_destroy (ctx, sink);//     result = s_check_event (sourcemon, ZMQ_EVENT_DISCONNECTED);//     assert (result);    zmonitor_destroy (&sinkmon);    zmonitor_destroy (&sourcemon);    zctx_destroy (&ctx);    //  @end    printf ("OK/n");}

intzre_log_msg_test (bool verbose){    printf (" * zre_log_msg: ");    //  @selftest    //  Simple create/destroy test    zre_log_msg_t *self = zre_log_msg_new (0);    assert (self);    zre_log_msg_destroy (&self);    //  Create pair of sockets we can send through    zctx_t *ctx = zctx_new ();    assert (ctx);    void *output = zsocket_new (ctx, ZMQ_DEALER);    assert (output);    zsocket_bind (output, "inproc://selftest");    void *input = zsocket_new (ctx, ZMQ_ROUTER);    assert (input);    zsocket_connect (input, "inproc://selftest");        //  Encode/send/decode and verify each message type    self = zre_log_msg_new (ZRE_LOG_MSG_LOG);    zre_log_msg_set_level (self, 123);    zre_log_msg_set_event (self, 123);    zre_log_msg_set_node (self, 123);    zre_log_msg_set_peer (self, 123);    zre_log_msg_set_time (self, 123);    zre_log_msg_set_data (self, "Life is short but Now lasts for ever");    zre_log_msg_send (&self, output);        self = zre_log_msg_recv (input);    assert (self);    assert (zre_log_msg_level (self) == 123);    assert (zre_log_msg_event (self) == 123);    assert (zre_log_msg_node (self) == 123);    assert (zre_log_msg_peer (self) == 123);    assert (zre_log_msg_time (self) == 123);    assert (streq (zre_log_msg_data (self), "Life is short but Now lasts for ever"));    zre_log_msg_destroy (&self);    zctx_destroy (&ctx);    //  @end    printf ("OK/n");    return 0;}

//  @selfteststatic void *s_test_detached (void *args){	void *push;	zctx_t *ctx;    //  Create a socket to check it'll be automatically deleted    ctx = zctx_new ();    assert (ctx);    push = zsocket_new (ctx, ZMQ_PUSH);    assert (push);    zctx_destroy (&ctx);    return NULL;}

int main (void){    //  Create threads    zctx_t *ctx = zctx_new ();    void *client = zthread_fork (ctx, client_task, NULL);    zthread_new (worker_task, NULL);    zthread_new (broker_task, NULL);    //  Wait for signal on client pipe    char *signal = zstr_recv (client);    free (signal);    zctx_destroy (&ctx);    return 0;}

rrsvr_t *rrsvr_new(){    rrsvr_t *svr;    type_malloc_clear(svr, rrsvr_t, 1);    _rrsvr_init(svr);    svr->ctx = zctx_new();    if (!svr->ctx) {        free(svr);        return NULL;    }    return svr;}

mdcli_t *mdcli_new (char *broker, int verbose){    assert (broker);    mdcli_t *self = (mdcli_t *) zmalloc (sizeof (mdcli_t));    self->ctx = zctx_new ();    self->broker = strdup (broker);    self->verbose = verbose;    self->timeout = 2500;           //  msecs    self->retries = 3;              //  Before we abandon    s_mdcli_connect_to_broker (self);    return self;}

int main (int argc, char *argv []){    //  Initialize context for talking to tasks    zctx_t *ctx = zctx_new ();    zctx_set_linger (ctx, 100);        //  Get number of nodes N to simulate    //  We need 3 x N x N + 3N file handles    int max_nodes = 10;    int nbr_nodes = 0;    if (argc > 1)        max_nodes = atoi (argv [1]);    int max_iterations = -1;    int nbr_iterations = 0;    if (argc > 2)        max_iterations = atoi (argv [2]);    //  We address nodes as an array of pipes    void **pipes = zmalloc (sizeof (void *) * max_nodes);    //  We will randomly start and stop node threads    while (!zctx_interrupted) {        uint index = randof (max_nodes);        //  Toggle node thread        if (pipes [index]) {            zstr_send (pipes [index], "STOP");            zsocket_destroy (ctx, pipes [index]);            pipes [index] = NULL;            zclock_log ("I: Stopped node (%d running)", --nbr_nodes);        }        else {            pipes [index] = zthread_fork (ctx, node_task, NULL);            zclock_log ("I: Started node (%d running)", ++nbr_nodes);        }        nbr_iterations++;        if (max_iterations > 0 && nbr_iterations >= max_iterations)            break;        //  Sleep ~750 msecs randomly so we smooth out activity        zclock_sleep (randof (500) + 500);    }    zclock_log ("I: Stopped tester (%d iterations)", nbr_iterations);        //  Does not actually terminate properly... :-/    //  zctx_destroy (&ctx);    free (pipes);    return 0;}

client_state* client_state_init(const char* server_url, const char* login){    srand(time(0));    client_state * state = new client_state();    state->last_message_id = 0;    state->context = zctx_new ();    state->login = login;        // Set random identity to make tracing easier    sprintf (state->identity, "%04X-%04X", randof (0x10000), randof (0x10000));        state->server_url = server_url;        client_state_reset_heartbeat(state);    return state;}

static rsRetVal initZMQ(instanceData* pData) {    DEFiRet;        /* create the context if necessary. */    if (NULL == s_context) {        s_context = zctx_new();        if (s_workerThreads > 0) zctx_set_iothreads(s_context, s_workerThreads);    }        pData->socket = zsocket_new(s_context, pData->type);    /* ALWAYS set the HWM as the zmq3 default is 1000 and we default       to 0 (infinity) */    zsocket_set_rcvhwm(pData->socket, pData->rcvHWM);    zsocket_set_sndhwm(pData->socket, pData->sndHWM);        /* use czmq defaults for these, unless set to non-default values */    if(pData->identity)             zsocket_set_identity(pData->socket, (char*)pData->identity);    if(pData->sndBuf > -1)          zsocket_set_sndbuf(pData->socket, pData->sndBuf);    if(pData->rcvBuf > -1)          zsocket_set_sndbuf(pData->socket, pData->rcvBuf);    if(pData->linger > -1)          zsocket_set_linger(pData->socket, pData->linger);    if(pData->backlog > -1)         zsocket_set_backlog(pData->socket, pData->backlog);    if(pData->sndTimeout > -1)      zsocket_set_sndtimeo(pData->socket, pData->sndTimeout);    if(pData->rcvTimeout > -1)      zsocket_set_rcvtimeo(pData->socket, pData->rcvTimeout);    if(pData->maxMsgSize > -1)      zsocket_set_maxmsgsize(pData->socket, pData->maxMsgSize);    if(pData->rate > -1)            zsocket_set_rate(pData->socket, pData->rate);    if(pData->recoveryIVL > -1)     zsocket_set_recovery_ivl(pData->socket, pData->recoveryIVL);    if(pData->multicastHops > -1)   zsocket_set_multicast_hops(pData->socket, pData->multicastHops);    if(pData->reconnectIVL > -1)    zsocket_set_reconnect_ivl(pData->socket, pData->reconnectIVL);    if(pData->reconnectIVLMax > -1) zsocket_set_reconnect_ivl_max(pData->socket, pData->reconnectIVLMax);    if(pData->ipv4Only > -1)        zsocket_set_ipv4only(pData->socket, pData->ipv4Only);    if(pData->affinity != 1)        zsocket_set_affinity(pData->socket, pData->affinity);                                      /* bind or connect to it */    if (pData->action == ACTION_BIND) {        /* bind asserts, so no need to test return val here           which isn't the greatest api -- oh well */        zsocket_bind(pData->socket, (char*)pData->description);    } else {        if(zsocket_connect(pData->socket, (char*)pData->description) == -1) {            errmsg.LogError(0, RS_RET_SUSPENDED, "omzmq3: connect failed!");            ABORT_FINALIZE(RS_RET_SUSPENDED);        }    } finalize_it:    RETiRet;}

int main (int argc, char *argv []){    zctx_t *ctx = zctx_new ();    //  Use the CZMQ zbeacon class to make sure we listen on the    //  same network interface as our peers    void *collector = zsocket_new (ctx, ZMQ_SUB);    zbeacon_t *beacon = zbeacon_new (ZRE_DISCOVERY_PORT);    char *host = zbeacon_hostname (beacon);    //  Bind to an ephemeral port    int port = zsocket_bind (collector, "tcp://%s:*", host);    //  Announce this to all peers we connect to    zre_node_t *node = zre_node_new ();    zre_node_header_set (node, "X-ZRELOG", "tcp://%s:%d", host, port);    //  Get all log messages (don't filter)    zsocket_set_subscribe (collector, "");    zmq_pollitem_t pollitems [] = {        { collector, 0, ZMQ_POLLIN, 0 },        { zre_node_handle (node), 0, ZMQ_POLLIN, 0 }    };    while (!zctx_interrupted) {        if (zmq_poll (pollitems, 2, 1000 * ZMQ_POLL_MSEC) == -1)            break;              //  Interrupted        //  Handle input on collector        if (pollitems [0].revents & ZMQ_POLLIN)            s_print_log_msg (collector);        //  Handle event from node (ignore it)        if (pollitems [1].revents & ZMQ_POLLIN) {            zmsg_t *msg = zre_node_recv (node);            if (!msg)                break;              //  Interrupted            zmsg_destroy (&msg);        }    }    zre_node_destroy (&node);    zbeacon_destroy (&beacon);    zctx_destroy (&ctx);    return 0;}

zre_interface_t *zre_interface_new (void){    zre_interface_t        *self;    self = (zre_interface_t *) zmalloc (sizeof (zre_interface_t));    //  If caller set a default ctx use that, else create our own    if (zre_global_ctx)        self->ctx = zre_global_ctx;    else {        self->ctx = zctx_new ();        self->ctx_owned = true;    }    self->pipe = zthread_fork (self->ctx, zre_interface_agent, NULL);    return self;}

int main (int argn, char *argv []) {    if (argn < 2) {        puts ("syntax: ./zyrechat myname");        exit (0);    }    zctx_t *ctx = zctx_new ();    void *chat_pipe = zthread_fork (ctx, chat_task, NULL);    while (!zctx_interrupted) {        char message [1024];        if (!fgets (message, 1024, stdin))            break;        zstr_sendf (chat_pipe, "%s:%s", argv [1], message);    }    zctx_destroy (&ctx);    return 0;}

void *server_task (void *args){    zctx_t *ctx = zctx_new ();    void *publisher = zsocket_new (ctx, ZMQ_PUB);    void *puller = zsocket_new (ctx, ZMQ_PULL);    zsocket_bind (publisher, "tcp://*:5563");    zsocket_bind (puller, "tcp://*:5564");    while (true) {        char *message = zstr_recv (puller);        if (!message)            break;        zstr_sendm (publisher, "A");        zstr_send  (publisher, "1");    }    return NULL;}

mdp_worker_t *mdp_worker_new (char *broker,char *service, int verbose){    assert (broker);    assert (service);    mdp_worker_t *self = (mdp_worker_t *) zmalloc (sizeof (mdp_worker_t));    self->ctx = zctx_new ();    self->broker = strdup (broker);    self->service = strdup (service);    self->verbose = verbose;    self->heartbeat = 2500;     //  msecs    self->reconnect = 2500;     //  msecs    s_mdp_worker_connect_to_broker (self);    return self;}

int main (void){    zctx_t *ctx = zctx_new ();    void *subscriber = zsocket_new (ctx, ZMQ_SUB);    zsocket_connect (subscriber, "tcp://localhost:30001");    zsocket_set_subscribe (subscriber, "");        while (true) {        zmsg_t *msg = zmsg_recv (subscriber);        if (!msg)            break;        zmsg_dump (msg);        zmsg_destroy (&msg);    }    zctx_destroy (&ctx);    return 0;}

int main (int argc, char *argv []){    int verbose = (argc > 1 && streq (argv [1], "-v"));    zctx_t *ctx = zctx_new ();    //  Prepare server socket with predictable identity    char *bind_endpoint = "tcp://*:5555";    char *connect_endpoint = "tcp://localhost:5555";    void *server = zsocket_new (ctx, ZMQ_ROUTER);    zmq_setsockopt (server,        ZMQ_IDENTITY, connect_endpoint, strlen (connect_endpoint));    zsocket_bind (server, bind_endpoint);    printf ("I: service is ready at %s/n", bind_endpoint);    while (!zctx_interrupted) {        zmsg_t *request = zmsg_recv (server);        if (verbose && request)            zmsg_dump (request);        if (!request)            break;          //  Interrupted        //  Frame 0: identity of client        //  Frame 1: PING, or client control frame        //  Frame 2: request body        zframe_t *identity = zmsg_pop (request);        zframe_t *control = zmsg_pop (request);        zmsg_t *reply = zmsg_new ();        if (zframe_streq (control, "PING"))            zmsg_addstr (reply, "PONG");        else {            zmsg_add (reply, control);            zmsg_addstr (reply, "OK");        }        zmsg_destroy (&request);        zmsg_push (reply, identity);        if (verbose && reply)            zmsg_dump (reply);        zmsg_send (&reply, server);    }    if (zctx_interrupted)        printf ("W: interrupted/n");    zctx_destroy (&ctx);    return 0;}

intzsocket_test (Bool verbose){    printf (" * zsocket: ");    //  @selftest    zctx_t *ctx = zctx_new ();    assert (ctx);    //  Create a detached thread, let it run    char *interf = "*";    char *domain = "localhost";    int service = 5560;    void *writer = zsocket_new (ctx, ZMQ_PUSH);    assert (writer);    void *reader = zsocket_new (ctx, ZMQ_PULL);    assert (reader);    assert (streq (zsocket_type_str (writer), "PUSH"));    assert (streq (zsocket_type_str (reader), "PULL"));    int rc = zsocket_bind (writer, "tcp://%s:%d", interf, service);    assert (rc == service);    rc = zsocket_connect (reader, "tcp://%s:%d", domain, service);    assert (rc == 0);    zstr_send (writer, "HELLO");    char *message = zstr_recv (reader);    assert (message);    assert (streq (message, "HELLO"));    free (message);    int port = zsocket_bind (writer, "tcp://%s:*", interf);    assert (port >= ZSOCKET_DYNFROM && port <= ZSOCKET_DYNTO);    assert (zsocket_poll (writer, 100) == FALSE);    rc = zsocket_connect (reader, "txp://%s:%d", domain, service);    assert (rc == -1);    zsocket_destroy (ctx, writer);    zctx_destroy (&ctx);    //  @end    printf ("OK/n");    return 0;}

char *  test_line() {  zctx_t *context = zctx_new();  lineconfig_t config;  config.line_id = 2;  config_t base = { "uuid", "inproc:broker", "inproc:portwatcher",                         "inproc:registration", "http://something.com" };  config.base_config = &base;  void * line_in = zsocket_new(context, ZMQ_PUB);  zsocket_bind (line_in, "inproc://line");  // filter(identity, serial_reader, remote, context);    zclock_log("pre-fork");  void * pipe = zthread_fork(context, line_listener, &config);  // TODO a real test  zstr_send(pipe, "ping");  char * result = zstr_recv(pipe);  mu_assert_str("pong response", "pong", result);  free(result);  // do some channel switching  int i,j;  for(i=0;i<100;i++) {    // first channel    for(j=0;j<3; j++) {      val_msg(line_in, "foo", 12);    }    for(j=0;j<3;j++) {      val_msg(line_in, "bar", 99);    }  }  zstr_send(line_in, "DESTROY");  // temp  printf("destroy message sent/n");  result = zstr_recv(pipe);  mu_assert_str("destroy response", "ok", result);  free(result);  zsocket_destroy(context,pipe);  printf("freed pipe/n");  zsocket_destroy(context, line_in);  zctx_destroy(&context);  return NULL;}

mq_socket_context_t *zero_socket_context_new(){    mq_socket_context_t *ctx;    tbx_type_malloc_clear(ctx, mq_socket_context_t, 1);    ctx->arg = zctx_new();    assert(ctx->arg != NULL);    zctx_set_linger(ctx->arg, 0);    ctx->create_socket = zero_create_socket;    ctx->destroy = zero_socket_context_destroy;    //** Disable the CZMQ SIGINT/SIGTERM signale handler    apr_signal(SIGINT, NULL);    apr_signal(SIGTERM, NULL);    return(ctx);}

//  Request-reply client using REQ socket//  To simulate load, clients issue a burst of requests and then//  sleep for a random period.//static void *client_task (void *args){    zctx_t *ctx = zctx_new ();    void *client = zsocket_new (ctx, ZMQ_REQ);    zsocket_connect (client, "ipc://%s-localfe.ipc", self);    void *monitor = zsocket_new (ctx, ZMQ_PUSH);    zsocket_connect (monitor, "ipc://%s-monitor.ipc", self);    while (1) {        sleep (randof (5));        int burst = randof (15);        while (burst--) {            char task_id [5];            sprintf (task_id, "%04X", randof (0x10000));            //  Send request with random hex ID            zstr_send (client, task_id);            //  Wait max ten seconds for a reply, then complain            zmq_pollitem_t pollset [1] = { { client, 0, ZMQ_POLLIN, 0 } };            int rc = zmq_poll (pollset, 1, 10 * 1000 * ZMQ_POLL_MSEC);            if (rc == -1)                break;          //  Interrupted            if (pollset [0].revents & ZMQ_POLLIN) {                char *reply = zstr_recv (client);                if (!reply)                    break;              //  Interrupted                //  Worker is supposed to answer us with our task id                puts (reply);                assert (streq (reply, task_id));                free (reply);            }            else {                zstr_sendf (monitor,                            "E: CLIENT EXIT - lost task %s", task_id);                return NULL;            }        }    }    zctx_destroy (&ctx);    return NULL;}

int main (void){    zctx_t *ctx = zctx_new ();    void *worker = zsocket_new (ctx, ZMQ_REQ);    //  Set random identity to make tracing easier    srandom ((unsigned) time (NULL));    char identity [10];    sprintf (identity, "%04X-%04X", randof (0x10000), randof (0x10000));    zmq_setsockopt (worker, ZMQ_IDENTITY, identity, strlen (identity));    zsocket_connect (worker, "tcp://localhost:5556");    //  Tell broker we're ready for work    printf ("I: (%s) worker ready/n", identity);    zframe_t *frame = zframe_new (LRU_READY, 1);    zframe_send (&frame, worker, 0);    int cycles = 0;    while (true) {        zmsg_t *msg = zmsg_recv (worker);        if (!msg)            break;              //  Interrupted        //  Simulate various problems, after a few cycles        cycles++;        if (cycles > 3 && randof (5) == 0) {            printf ("I: (%s) simulating a crash/n", identity);            zmsg_destroy (&msg);            break;        }        else        if (cycles > 3 && randof (5) == 0) {            printf ("I: (%s) simulating CPU overload/n", identity);            sleep (3);            if (zctx_interrupted)                break;        }        printf ("I: (%s) normal reply/n", identity);        sleep (1);              //  Do some heavy work        zmsg_send (&msg, worker);    }    zctx_destroy (&ctx);    return 0;}

PoolFrontend::PoolFrontend(unsigned port) {		printf("PoolFrontend started on port %d./n", port);		mCtx = zctx_new();		mDealer = zsocket_new(mCtx, ZMQ_DEALER);	mRouter = zsocket_new(mCtx, ZMQ_ROUTER);		zsocket_bind(mDealer, "tcp://*:7777");	unsigned ret = zsocket_bind(mRouter, "tcp://*:%d", port);	if(ret != port){		printf("Frontend: ERROR: zsocket_bind failed./n");		exit(-1);	}		zthread_fork(mCtx, &PoolFrontend::InvokeProxy, this);	}