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

void s_connect_to_broker (mdwrk_t *self){    if (self->worker)        zmq_close (self->worker);    self->worker = zmq_socket (self->context, ZMQ_XREQ);    int linger = 0;    zmq_setsockopt (self->worker, ZMQ_LINGER, &linger, sizeof (linger));    zmq_connect (self->worker, self->broker);    //  Register service with broker    zmsg_t *msg = zmsg_new ();    zmsg_append (msg, MDPS_HEADER);    zmsg_append (msg, MDPS_READY);    zmsg_append (msg, self->service);    zmsg_send (&msg, self->worker);    //  If liveness hits zero, queue is considered disconnected    self->liveness = HEARTBEAT_LIVENESS;    self->heartbeat_at = s_clock () + HEARTBEAT_INTERVAL;}

int main (void){    srandom ((unsigned) time (NULL));    void *context = zmq_init (1);    void *worker = zmq_socket (context, ZMQ_REQ);    //  Set random identity to make tracing easier    char identity [10];    sprintf (identity, "%04X-%04X", randof (0x10000), randof (0x10000));    zmq_setsockopt (worker, ZMQ_IDENTITY, identity, strlen (identity));    zmq_connect (worker, "tcp://localhost:5556");    //  Tell queue we're ready for work    printf ("I: (%s) worker ready/n", identity);    s_send (worker, "READY");    int cycles = 0;    while (1) {        zmsg_t *zmsg = zmsg_recv (worker);        //  Simulate various problems, after a few cycles        cycles++;        if (cycles > 3 && randof (5) == 0) {            printf ("I: (%s) simulating a crash/n", identity);            zmsg_destroy (&zmsg);            break;        }        else        if (cycles > 3 && randof (5) == 0) {            printf ("I: (%s) simulating CPU overload/n", identity);            sleep (5);        }        printf ("I: (%s) normal reply - %s/n", identity, zmsg_body (zmsg));        sleep (1);              //  Do some heavy work        zmsg_send (&zmsg, worker);    }    zmq_close (worker);    zmq_term (context);    return 0;}

static voids_worker_send (worker_t *self, char *command, char *option, zmsg_t *msg){    msg = msg? zmsg_dup (msg): zmsg_new ();    //  Stack protocol envelope to start of message    if (option)        zmsg_pushstr (msg, option);    zmsg_pushstr (msg, command);    zmsg_pushstr (msg, MDPW_WORKER);    //  Stack routing envelope to start of message    zmsg_wrap (msg, zframe_dup (self->address));    if (self->broker->verbose) {        zclock_log ("I: sending %s to worker",            mdpw_commands [(int) *command]);        zmsg_dump (msg);    }    zmsg_send (&msg, self->broker->socket);}

int _tmain(int argc, _TCHAR* argv[]){	zclock_sleep(20000);	char* broker_loc = "tcp://localhost:5555";	zctx_t* ctx = zctx_new();	void* scket = zsocket_new(ctx,ZMQ_REQ);	zsocket_connect(scket,broker_loc);	zmsg_t* msg = zmsg_new();	zmsg_addstr(msg,TWRK_CLI_VER);	zmsg_addstr(msg,"Echo");	zmsg_addstr(msg,TMSG_TYPE_REQUEST);		int64_t sleep = 10*1000;	zclock_sleep(sleep);	zmsg_add(msg,zframe_new(&sleep,sizeof(sleep)));	zmsg_send(&msg,scket);	zmsg_t* reply =  zmsg_recv(scket);	zmsg_dump(reply);	return 0;}

intmdcli_send (mdcli_t *self, char *service, zmsg_t **request_p){    assert (self);    assert (request_p);    zmsg_t *request = *request_p;    //  Prefix request with protocol frames    //  Frame 0: empty (REQ emulation)    //  Frame 1: "MDPCxy" (six bytes, MDP/Client x.y)    //  Frame 2: Service name (printable string)    zmsg_push (request, service);    zmsg_push (request, MDPC_CLIENT);    zmsg_push (request, "");    if (self->verbose) {        s_console ("I: send request to '%s' service:", service);        zmsg_dump (request);    }    zmsg_send (&request, self->client);    return 0;}

void graph_response_newLinkDataResponse(req_t * req, json_t * request,					json_t * response, int32_t requestId,					void *sweb, req_store_t * req_store){	if (strcmp	    (json_string_value(json_object_get(response, "ack")), "ok") == 0) {		json_t *link = json_object_get(request, "link");		json_t *web_resp = json_object();		json_object_set_new(web_resp, "type",				    json_string("newLinkData"));		//TODO at the moment only the original node gets the update, which is good enough for me		json_t *sessionIds = json_array();		json_array_append		    (sessionIds, json_object_get(req->request, "sessionId"));		json_object_set_new(web_resp, "sessionIds", sessionIds);		json_t *newData = json_object();		json_t *newLinkData = json_array();		json_array_append(newLinkData, link);		json_object_set_new(newData, "newLinkData", newLinkData);		json_object_set_new(web_resp, "newData", newData);		zmsg_t *res = zmsg_new();		char *web_res_str = json_dumps(web_resp,					       JSON_COMPACT);		printf("/nbroker:sweb sent: %s/n", web_res_str);		zmsg_addstr(res, web_res_str);		free(web_res_str);		zmsg_wrap(res, req->address);		zmsg_send(&res, sweb);		json_decref(web_resp);	} else {	}	request_store_delete(req_store, requestId);}

static voids_service_internal (broker_t *self, char *service_name, zmsg_t *msg){    if (streq (service_name, "mmi.service")) {        service_t *service =             (service_t *) zhash_lookup (self->services, zmsg_body (msg));        if (service && service->workers)            zmsg_body_set (msg, "200");        else            zmsg_body_set (msg, "404");    }    else        zmsg_body_set (msg, "501");    //  Remove & save client return envelope and insert the    //  protocol header and service name, then rewrap envelope.    char *client = zmsg_unwrap (msg);    zmsg_wrap (msg, MDPC_CLIENT, service_name);    zmsg_wrap (msg, client, "");    free (client);    zmsg_send (&msg, self->socket);}

//  Worker using REQ socket to do load-balancing//static void *worker_task (void *args){    zctx_t *ctx = zctx_new ();    void *worker = zsocket_new (ctx, ZMQ_REQ);    zsocket_connect (worker, "ipc://backend.ipc");    //  Tell broker we're ready for work    zframe_t *frame = zframe_new (WORKER_READY, 1);    zframe_send (&frame, worker, 0);    //  Process messages as they arrive    while (true) {        zmsg_t *msg = zmsg_recv (worker);        if (!msg)            break;              //  Interrupted        zframe_reset (zmsg_last (msg), "OK", 2);        zmsg_send (&msg, worker);    }    zctx_destroy (&ctx);    return NULL;}

static int sleep_loop(zloop_t *loop, int item, void *arg){		int *c = (int*)arg;	int i;	for (i = 0; i < *c; i++) 	{		if (interrupt)			return -1;				char *actionid = getActionid();		zmsg_t *msg = create_call(actionid, "worker", "sleep", "echo");		free(actionid);		if (msg) {			zmsg_send(&msg, dealer);			counter ++;		}	}	return 0;}

void prob_get_label_group(prob_client_t pc, ProBState s, int group, int *res) {    zmsg_t *request = zmsg_new();    zmsg_addstr(request, "get-state-label-group");    zmsg_addstrf(request, "%d", pc->id_count);    zmsg_addstrf(request, "DA%d", group);    prob_put_state(request, s);    zmsg_send(&request, pc->zocket);    zmsg_destroy(&request);    zmsg_t *response = zmsg_recv(pc->zocket);    drop_frame(response);    drop_frame(response);    char *result_s;    for (int i = 0; (result_s = zmsg_popstr(response)) != NULL; i++) {        int r;        sscanf(result_s, "%d", &r);        res[i] = r;        RTfree(result_s);    }    zmsg_destroy(&response);}

intprob_get_state_label(prob_client_t pc, ProBState s, char *label){    zmsg_t *request = zmsg_new();    zmsg_addstr(request, "get-state-label");    zmsg_addstrf(request, "%d", pc->id_count);    zmsg_addstrf(request, "DA%s", label);    prob_put_state(request, s);    zmsg_send(&request, pc->zocket);    zmsg_destroy(&request);    zmsg_t *response = zmsg_recv(pc->zocket);    drop_frame(response);    drop_frame(response);    char *result_s = zmsg_popstr(response);    int res;    sscanf(result_s, "%d", &res);    RTfree(result_s);    zmsg_destroy(&response);    return res;}

static voids_worker_send (    broker_t *self, worker_t *worker,    char *command, char *option, zmsg_t *msg){    msg = msg? zmsg_dup (msg): zmsg_new (NULL);    //  Stack protocol envelope to start of message    if (option)                 //  Optional frame after command        zmsg_push (msg, option);    zmsg_push (msg, command);    zmsg_push (msg, MDPW_WORKER);    //  Stack routing envelope to start of message    zmsg_wrap (msg, worker->identity, "");    if (self->verbose) {        s_console ("I: sending %s to worker",            mdps_commands [(int) *command]);        zmsg_dump (msg);    }    zmsg_send (&msg, self->socket);}

zmsg_t *flclient_request (flclient_t *self, zmsg_t **request_p){    assert (self);    assert (*request_p);    zmsg_t *request = *request_p;        //  Prefix request with sequence number and empty envelope    char sequence_text [10];    sprintf (sequence_text, "%u", ++self->sequence);    zmsg_push (request, sequence_text);    zmsg_push (request, "");        //  Blast the request to all connected servers    int server;    for (server = 0; server < self->servers; server++) {        zmsg_t *msg = zmsg_dup (request);        zmsg_send (&msg, self->socket);    }    //  Wait for a matching reply to arrive from anywhere    //  Since we can poll several times, calculate each one    zmsg_t *reply = NULL;    uint64_t endtime = s_clock () + GLOBAL_TIMEOUT;    while (s_clock () < endtime) {        zmq_pollitem_t items [] = { { self->socket, 0, ZMQ_POLLIN, 0 } };        zmq_poll (items, 1, (endtime - s_clock ()) * 1000);        if (items [0].revents & ZMQ_POLLIN) {            reply = zmsg_recv (self->socket);            assert (zmsg_parts (reply) == 3);            free (zmsg_pop (reply));            if (atoi (zmsg_address (reply)) == self->sequence)                break;            zmsg_destroy (&reply);        }    }    zmsg_destroy (request_p);    return reply;}

int main(void) {    zmsg_t *msg;    zframe_t *frame;    char *str;    int i, rc;    // create push/pull sockets    zsock_t *push = zsock_new_push("inproc://example");    zsock_t *pull = zsock_new_pull("inproc://example");    // send multi-frame message    msg = zmsg_new();    zmsg_addmem(msg, "apple", 5);    zmsg_addmem(msg, "banana", 6);    zmsg_addmem(msg, "cherry", 6);    assert(zmsg_size(msg) == 3);    assert(zmsg_content_size(msg) == 5+6+6);    rc = zmsg_send(&msg, push);    assert(msg == NULL);    assert(rc == 0);    // receive multi-frame message    msg = zmsg_recv(pull);    assert(msg);    assert(zmsg_size(msg) == 3);    assert(zmsg_content_size(msg) == 5+6+6);    for (i = 0; i < 3; i++) {        str = zmsg_popstr(msg);        puts(str);    }    zmsg_destroy(&msg);    // disconnect    zsock_destroy(&push);    zsock_destroy(&pull);    return 0;}

static ProBState *prob_next_x(prob_client_t pc, ProBState s, char *transitiongroup, int *size, char *header) {    zmsg_t *request = zmsg_new();    zmsg_addstr(request, header);    zmsg_addstrf(request, "%d", pc->id_count);    zmsg_addstr(request, transitiongroup);    prob_put_state(request, s);    Debugf("requesting next-state, contents:/n");#ifdef LTSMIN_DEBUG    if (log_active(debug)) zmsg_print(request);#endif    zmsg_send(&request, pc->zocket);    zmsg_destroy(&request);    zmsg_t *response = zmsg_recv(pc->zocket);    Debugf("response for next-state, contents:/n");#ifdef LTSMIN_DEBUG    if (log_active(debug)) zmsg_print(response);#endif    drop_frame(response);    drop_frame(response);    char *nr_of_states_s = zmsg_popstr(response);    sscanf(nr_of_states_s, "%d", size);    RTfree(nr_of_states_s);    ProBState *successors = RTmalloc(sizeof(ProBState) * (*size));    int i;    for (i = 0; i < (*size); i++) {        successors[i] = prob_get_state(response);    }    zmsg_destroy(&response);    return successors;}

//  Handle input from worker, on backendint s_handle_backend (zloop_t *loop, zmq_pollitem_t *poller, void *arg){    //  Use worker identity for load-balancing    lbbroker_t *self = (lbbroker_t *) arg;    zmsg_t *msg = zmsg_recv (self->backend);    if (msg) {        zframe_t *identity = zmsg_unwrap (msg);        zlist_append (self->workers, identity);        //  Enable reader on frontend if we went from 0 to 1 workers        if (zlist_size (self->workers) == 1) {            zmq_pollitem_t poller = { self->frontend, 0, ZMQ_POLLIN };            zloop_poller (loop, &poller, s_handle_frontend, self);        }        //  Forward message to client if it's not a READY        zframe_t *frame = zmsg_first (msg);        if (memcmp (zframe_data (frame), WORKER_READY, 1) == 0)            zmsg_destroy (&msg);        else            zmsg_send (&msg, self->frontend);    }    return 0;}

void graph_response_delNodeResponse(req_t * req, json_t * request,				    json_t * response, int32_t requestId,				    void *spss, req_store_t * req_store){	const	    char	*ack = json_string_value(json_object_get(response,						 "ack"));	if (strcmp(ack, "ok") == 0) {		json_t *wrequest = json_object();		json_object_set_new		    (wrequest, "requestId", json_integer(requestId));		json_object_set(wrequest, "request", request);		zmsg_t *req = zmsg_new();		char *req_str = json_dumps(wrequest,					   JSON_COMPACT);		printf("/nbroker:spss sent: %s/n", req_str);		zmsg_addstr(req, req_str);		free(req_str);		zmsg_send(&req, spss);		json_decref(wrequest);	}	else {		if (strcmp(ack, "fail") == 0) {			// This can fail because of the existence of links			// clean things up		}		request_store_delete(req_store, requestId);	}}

static voids_self_handle_udp (self_t *self){    assert (self);    char peername [INET_ADDRSTRLEN];    zframe_t *frame = zsys_udp_recv (self->udpsock, peername, INET_ADDRSTRLEN);    //  If filter is set, check that beacon matches it    bool is_valid = false;    if (self->filter) {        byte  *filter_data = zframe_data (self->filter);        size_t filter_size = zframe_size (self->filter);        if (zframe_size (frame) >= filter_size        && memcmp (zframe_data (frame), filter_data, filter_size) == 0)            is_valid = true;    }    //  If valid, discard our own broadcasts, which UDP echoes to us    if (is_valid && self->transmit) {        byte  *transmit_data = zframe_data (self->transmit);        size_t transmit_size = zframe_size (self->transmit);        if (zframe_size (frame) == transmit_size        && memcmp (zframe_data (frame), transmit_data, transmit_size) == 0)            is_valid = false;    }    //  If still a valid beacon, send on to the API    if (is_valid) {        zmsg_t *msg = zmsg_new ();        assert (msg);        zmsg_addstr (msg, peername);        zmsg_append (msg, &frame);        zmsg_send (&msg, self->pipe);    }    else        zframe_destroy (&frame);}

static void* worker_routine(void* arg){  zframe_t* frame;  zctx_t* ctx = zctx_new();  void* worker = zsocket_new(ctx, ZMQ_REQ);  zsocket_connect(worker, "ipc://backend.ipc");  frame = zframe_new(WORKER_READY, 1);  zframe_send(&frame, worker, 0);  while (1) {    zmsg_t* msg = zmsg_recv(worker);    if (NULL == msg)      break;    zframe_reset(zmsg_last(msg), "OK", 2);    zmsg_send(&msg, worker);  }  zctx_destroy(&ctx);  return NULL;}

static void *worker_task (void *args){    zctx_t *ctx = zctx_new ();    void *worker = zsocket_new (ctx, ZMQ_REQ);    zsocket_connect (worker, "ipc://%s-localbe.ipc", self);    //  Tell broker we're ready for work    zframe_t *frame = zframe_new (WORKER_READY, 1);    zframe_send (&frame, worker, 0);    //  Process messages as they arrive    while (true) {        zmsg_t *msg = zmsg_recv (worker);        if (!msg)            break;              //  Interrupted        //  Workers are busy for 0/1 seconds        sleep (randof (2));        zmsg_send (&msg, worker);    }    zctx_destroy (&ctx);    return NULL;}

//  Accept a request and reply with the same text a random number of//  times, with random delays between replies.//static void *server_worker (void *context) {    void *worker = zmq_socket (context, ZMQ_XREQ);    zmq_connect (worker, "inproc://backend");    while (1) {        //  The XREQ socket gives us the address envelope and message        zmsg_t *msg = zmsg_recv (worker);        assert (zmsg_parts (msg) == 2);                //  Send 0..4 replies back        int reply, replies = randof (5);        for (reply = 0; reply < replies; reply++) {            //  Sleep for some fraction of a second            struct timespec t = { 0, randof (100000000) + 1 };            nanosleep (&t, NULL);            zmsg_t *dup = zmsg_dup (msg);            zmsg_send (&dup, worker);        }        zmsg_destroy (&msg);    }    zmq_close (worker);    return (NULL);}

static voids_service_internal (broker_t *self, zframe_t *service_frame, zmsg_t *msg){    char *return_code;    if (zframe_streq (service_frame, "mmi.service")) {        char *name = zframe_strdup (zmsg_last (msg));        service_t *service =            (service_t *) zhash_lookup (self->services, name);        return_code = service && service->workers? "200": "404";        free (name);    }    else        return_code = "501";    zframe_reset (zmsg_last (msg), return_code, strlen (return_code));    //  Remove & save client return envelope and insert the    //  protocol header and service name, then rewrap envelope.    zframe_t *client = zmsg_unwrap (msg);    zmsg_push (msg, zframe_dup (service_frame));    zmsg_pushstr (msg, MDPC_CLIENT);    zmsg_wrap (msg, client);    zmsg_send (&msg, self->socket);}

//.........这里部分代码省略.........        int previous = local_capacity;        //  Handle reply from local worker        zmsg_t *msg = NULL;        if (primary [0].revents & ZMQ_POLLIN) {            msg = zmsg_recv (localbe);            if (!msg)                break;          //  Interrupted            zframe_t *address = zmsg_unwrap (msg);            zlist_append (workers, address);            local_capacity++;            //  If it's READY, don't route the message any further            zframe_t *frame = zmsg_first (msg);            if (memcmp (zframe_data (frame), LRU_READY, 1) == 0)                zmsg_destroy (&msg);        }        //  Or handle reply from peer broker        else if (primary [1].revents & ZMQ_POLLIN) {            msg = zmsg_recv (cloudbe);            if (!msg)                break;          //  Interrupted            //  We don't use peer broker address for anything            zframe_t *address = zmsg_unwrap (msg);            zframe_destroy (&address);        }        //  Route reply to cloud if it's addressed to a broker        for (argn = 2; msg && argn < argc; argn++) {            char *data = (char *) zframe_data (zmsg_first (msg));            size_t size = zframe_size (zmsg_first (msg));            if (size == strlen (argv [argn])                    &&  memcmp (data, argv [argn], size) == 0)                zmsg_send (&msg, cloudfe);        }        //  Route reply to client if we still need to        if (msg)            zmsg_send (&msg, localfe);        //  Handle capacity updates        if (primary [2].revents & ZMQ_POLLIN) {            char *status = zstr_recv (statefe);            cloud_capacity = atoi (status);            free (status);        }        //  Handle monitor message        if (primary [3].revents & ZMQ_POLLIN) {            char *status = zstr_recv (monitor);            printf ("%s/n", status);            free (status);        }        //  Now route as many clients requests as we can handle        //  - If we have local capacity we poll both localfe and cloudfe        //  - If we have cloud capacity only, we poll just localfe        //  - Route any request locally if we can, else to cloud        //        while (local_capacity + cloud_capacity) {            zmq_pollitem_t secondary [] = {                { localfe, 0, ZMQ_POLLIN, 0 },                { cloudfe, 0, ZMQ_POLLIN, 0 }            };            if (local_capacity)                rc = zmq_poll (secondary, 2, 0);            else                rc = zmq_poll (secondary, 1, 0);

//.........这里部分代码省略.........            { localbe, 0, ZMQ_POLLIN, 0 },            { cloudbe, 0, ZMQ_POLLIN, 0 }        };        //  If we have no workers, wait indefinitely        int rc = zmq_poll (backends, 2,            capacity? 1000 * ZMQ_POLL_MSEC: -1);        if (rc == -1)            break;              //  Interrupted        //  Handle reply from local worker        zmsg_t *msg = NULL;        if (backends [0].revents & ZMQ_POLLIN) {            msg = zmsg_recv (localbe);            if (!msg)                break;          //  Interrupted            zframe_t *identity = zmsg_unwrap (msg);            zlist_append (workers, identity);            capacity++;            //  If it's READY, don't route the message any further            zframe_t *frame = zmsg_first (msg);            if (memcmp (zframe_data (frame), WORKER_READY, 1) == 0)                zmsg_destroy (&msg);        }        //  Or handle reply from peer broker        else        if (backends [1].revents & ZMQ_POLLIN) {            msg = zmsg_recv (cloudbe);            if (!msg)                break;          //  Interrupted            //  We don't use peer broker identity for anything            zframe_t *identity = zmsg_unwrap (msg);            zframe_destroy (&identity);        }        //  Route reply to cloud if it's addressed to a broker        for (argn = 2; msg && argn < argc; argn++) {            char *data = (char *) zframe_data (zmsg_first (msg));            size_t size = zframe_size (zmsg_first (msg));            if (size == strlen (argv [argn])            &&  memcmp (data, argv [argn], size) == 0)                zmsg_send (&msg, cloudfe);        }        //  Route reply to client if we still need to        if (msg)            zmsg_send (&msg, localfe);        //  .split route client requests        //  Now we route as many client requests as we have worker capacity        //  for. We may reroute requests from our local frontend, but not from         //  the cloud frontend. We reroute randomly now, just to test things        //  out. In the next version, we'll do this properly by calculating        //  cloud capacity:        while (capacity) {            zmq_pollitem_t frontends [] = {                { localfe, 0, ZMQ_POLLIN, 0 },                { cloudfe, 0, ZMQ_POLLIN, 0 }            };            rc = zmq_poll (frontends, 2, 0);            assert (rc >= 0);            int reroutable = 0;            //  We'll do peer brokers first, to prevent starvation            if (frontends [1].revents & ZMQ_POLLIN) {                msg = zmsg_recv (cloudfe);                reroutable = 0;            }            else            if (frontends [0].revents & ZMQ_POLLIN) {                msg = zmsg_recv (localfe);                reroutable = 1;            }            else                break;      //  No work, go back to backends            //  If reroutable, send to cloud 20% of the time            //  Here we'd normally use cloud status information            //            if (reroutable && argc > 2 && randof (5) == 0) {                //  Route to random broker peer                int peer = randof (argc - 2) + 2;                zmsg_pushmem (msg, argv [peer], strlen (argv [peer]));                zmsg_send (&msg, cloudbe);            }            else {                zframe_t *frame = (zframe_t *) zlist_pop (workers);                zmsg_wrap (msg, frame);                zmsg_send (&msg, localbe);                capacity--;            }        }    }    //  When we're done, clean up properly    while (zlist_size (workers)) {        zframe_t *frame = (zframe_t *) zlist_pop (workers);        zframe_destroy (&frame);    }    zlist_destroy (&workers);    zctx_destroy (&ctx);    return EXIT_SUCCESS;}

int main(void){	zctx_t *ctx = zctx_new();	void *frontend = zsocket_new(ctx, ZMQ_ROUTER);	void *backend = zsocket_new(ctx, ZMQ_ROUTER);	// IPC doesn't yet work on MS Windows.#if (defined (WIN32))	zsocket_bind(frontend, "tcp://*:5672");	zsocket_bind(backend, "tcp://*:5673");#else	zsocket_bind(frontend, "ipc://frontend.ipc");	zsocket_bind(backend, "ipc://backend.ipc");#endif	int client_nbr;	for (client_nbr = 0; client_nbr < NBR_CLIENTS; client_nbr++)		zthread_new(client_task, NULL);	int worker_nbr;	for (worker_nbr = 0; worker_nbr < NBR_WORKERS; worker_nbr++)		zthread_new(worker_task, NULL);	//  Queue of available workers	zlist_t *workers = zlist_new();	//  .split main load-balancer loop	//  Here is the main loop for the load balancer. It works the same way	//  as the previous example, but is a lot shorter because CZMQ gives	//  us an API that does more with fewer calls:	while (1) {		zmq_pollitem_t items[] = {				{ backend, 0, ZMQ_POLLIN, 0 },				{ frontend, 0, ZMQ_POLLIN, 0 }		};		//  Poll frontend only if we have available workers		int rc = zmq_poll(items, zlist_size(workers) ? 2 : 1, -1);		if (rc == -1)			break;              //  Interrupted		//  Handle worker activity on backend		if (items[0].revents & ZMQ_POLLIN) {			//  Use worker identity for load-balancing			zmsg_t *msg = zmsg_recv(backend);			if (!msg)				break;          //  Interrupted#if 0			// zmsg_unwrap is DEPRECATED as over-engineered, poor style			zframe_t *identity = zmsg_unwrap(msg);#else			zframe_t *identity = zmsg_pop(msg);			zframe_t *delimiter = zmsg_pop(msg);			zframe_destroy(&delimiter); #endif			zlist_append(workers, identity);			//  Forward message to client if it's not a READY			zframe_t *frame = zmsg_first(msg);			if (memcmp(zframe_data(frame), WORKER_READY, strlen(WORKER_READY)) == 0) {				zmsg_destroy(&msg);			} else {				zmsg_send(&msg, frontend);				if (--client_nbr == 0)					break; // Exit after N messages			}		}		if (items[1].revents & ZMQ_POLLIN) {			//  Get client request, route to first available worker			zmsg_t *msg = zmsg_recv(frontend);			if (msg) {#if 0				// zmsg_wrap is DEPRECATED as unsafe				zmsg_wrap(msg, (zframe_t *)zlist_pop(workers));#else				zmsg_pushmem(msg, NULL, 0); // delimiter				zmsg_push(msg, (zframe_t *)zlist_pop(workers));#endif				zmsg_send(&msg, backend);			}		}	}	//  When we're done, clean up properly	while (zlist_size(workers)) {		zframe_t *frame = (zframe_t *)zlist_pop(workers);		zframe_destroy(&frame);	}	zlist_destroy(&workers);	zctx_destroy(&ctx);	return 0;}

int main (void){    s_version_assert (2, 1);    srandom ((unsigned) time (NULL));    void *context = zmq_init (1);    void *worker = s_worker_socket (context);    //  If liveness hits zero, queue is considered disconnected    size_t liveness = HEARTBEAT_LIVENESS;    size_t interval = INTERVAL_INIT;    //  Send out heartbeats at regular intervals    uint64_t heartbeat_at = s_clock () + HEARTBEAT_INTERVAL;    int cycles = 0;    while (1) {        zmq_pollitem_t items [] = { { worker,  0, ZMQ_POLLIN, 0 } };        zmq_poll (items, 1, HEARTBEAT_INTERVAL * 1000);        if (items [0].revents & ZMQ_POLLIN) {            //  Get message            //  - 3-part envelope + content -> request            //  - 1-part "HEARTBEAT" -> heartbeat            zmsg_t *msg = zmsg_recv (worker);            if (msg_parts (msg) == 3) {                //  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 (5);                }                printf ("I: (%s) normal reply - %s/n",                    identity, zmsg_body (msg));                zmsg_send (&msg, worker);                liveness = HEARTBEAT_LIVENESS;                sleep (1);              //  Do some heavy work            }            else            if (msg_parts (msg) == 1            && strcmp (msg_body (msg), "HEARTBEAT") == 0)                liveness = HEARTBEAT_LIVENESS;            else {                printf ("E: (%s) invalid message/n", identity);                zmsg_dump (msg);            }            interval = INTERVAL_INIT;        }        else        if (--liveness == 0) {            printf ("W: (%s) heartbeat failure, can't reach queue/n",                identity);            printf ("W: (%s) reconnecting in %zd msec.../n",                identity, interval);            s_sleep (interval);            if (interval < INTERVAL_MAX)                interval *= 2;            zmq_close (worker);            worker = s_worker_socket (context);            liveness = HEARTBEAT_LIVENESS;        }        //  Send heartbeat to queue if it's time        if (s_clock () > heartbeat_at) {            heartbeat_at = s_clock () + HEARTBEAT_INTERVAL;            printf ("I: (%s) worker heartbeat/n", identity);            s_send (worker, "HEARTBEAT");        }    }    zmq_close (worker);    zmq_term (context);    return 0;}

int main(void){  zctx_t *ctx=zctx_new();  void *worker=_worker_socket(ctx);  size_t liveness=HEARTBEAT_LIVENESS;  size_t interval=INTERVAL_INIT;  uint64_t heartbeat_at=zclock_time()+HEARTBEAT_INTERVAL;  srandom((unsigned)time(NULL));  int cycles=0;  while (true){    zmq_pollitem_t items[]={{worker, 0, ZMQ_POLLIN, 0}};    int rc=zmq_poll(items, 1, HEARTBEAT_INTERVAL*ZMQ_POLL_MSEC);    if (rc==-1){      break;    }    if (items[0].revents & ZMQ_POLLIN){      zmsg_t *msg=zmsg_recv(worker);      if (!msg){        break;      }      if (zmsg_size(msg)==3){        ++cycles;        if (cycles>3 && randof(5)==0){          debug_log("I: simulating a crash/n");          zmsg_destroy(&msg);          break;        } else if (cycles>3 && randof(5)==0){          debug_log("I: simulating CPU overload/n");          sleep(3);          if (zctx_interrupted){            break;          }        } else{          debug_log("I: normal reply/n");          zmsg_send(&msg, worker);          sleep(1);          if (zctx_interrupted){            break;          }        }      } else if (zmsg_size(msg)==1){        zframe_t *frame=zmsg_first(msg);        if (memcmp(zframe_data(frame), PPP_HEARTBEAT, 1)==0){          liveness=HEARTBEAT_LIVENESS;        } else{          debug_log(ERROR_COLOR"E: inval message/n"NORMAL_COLOR);          zmsg_dump(msg);        }        zmsg_destroy(&msg);      } else{        debug_log(ERROR_COLOR"E: invalid message/n"NORMAL_COLOR);        zmsg_dump(msg);      }      interval=INTERVAL_INIT;    } else if (--liveness==0){      debug_log(WARN_COLOR"W: heartbeat failure, can't reach queue/n"                NORMAL_COLOR);      debug_log(WARN_COLOR"W: reconnecting in %zd msec"STR_ELLIPSIS"/n"                NORMAL_COLOR, interval);      zclock_sleep(interval);      if (interval<INTERVAL_MAX){        interval*=2;      }      zsocket_destroy(ctx, worker);      worker=_worker_socket(ctx);      liveness=HEARTBEAT_LIVENESS;    }    if (zclock_time()>heartbeat_at){      heartbeat_at=zclock_time()+HEARTBEAT_INTERVAL;      debug_log("I: worker heartbeat/n");      zframe_t *frame=zframe_new(PPP_HEARTBEAT, 1);      zframe_send(&frame, worker, 0);    }  }    zctx_destroy(&ctx);  return 0;}

void thread_test_inv(void* args, zctx_t *ctx, void *pipe) {		int param = (uint64)args;		void* server = zsocket_new(ctx, ZMQ_DEALER);		printf("thread_test_inv: port = %d/n", param);	int err = zsocket_connect(server, "tcp://localhost:%d", param);	assert(!err);		static const unsigned maxsize = 1024;	unsigned char buffer[maxsize+10];	memset(buffer, 0, sizeof(buffer));		uint64 invcount = 0;	proto::Request req;	while(true){				zmsg_t* msg = zmsg_new();		size_t fsize = rand() % maxsize;				for(size_t i = 0; i < fsize; ++i)			buffer[i] = rand();				zframe_t* frame = zframe_new(buffer, fsize);				zmsg_append(msg, &frame);		zmsg_send(&msg, server);				/*if(invcount % 2 == 0){						req.set_type(proto::Request::SHARE);			req.set_reqid(1);			GetNewReqNonce(req);			req.set_version(10);			req.set_height(1337);						proto::Share* share = req.mutable_share();			setRandStr(share->mutable_addr(), rand()%32);			setRandStr(share->mutable_name(), rand()%32);			setRandStr(share->mutable_hash(), rand()%32);			setRandStr(share->mutable_merkle(), rand()%32);			setRandStr(share->mutable_multi(), rand()%50);			setRandStr(share->mutable_blockhash(), rand()%40);			share->set_clientid(1313);			share->set_bits(3444);			share->set_length(10);			share->set_chaintype(1);			share->set_height(2334);			share->set_isblock(true);			share->set_nonce(333434);					}else{						req.set_type(proto::Request::STATS);			req.set_reqid(1);			GetNewReqNonce(req);			req.set_version(10);			req.set_height(1337);						proto::ClientStats* stats = req.mutable_stats();			setRandStr(stats->mutable_addr(), rand()%32);			setRandStr(stats->mutable_name(), rand()%32);			stats->set_clientid(345345);			stats->set_instanceid(34344555);			stats->set_version(10);			stats->set_latency(445);			stats->set_ngpus(4);			stats->set_height(433435);					}				Send(req, server);*/				if(zctx_interrupted)			break;				invcount++;		if(invcount % 102400 == 0)			printf("thread_test_inv: invcount = %d/%d/n", (unsigned)(invcount >> 32), (unsigned)invcount);			}		zsocket_destroy(ctx, server);	}

int PrimeWorker::HandleInput(zmq_pollitem_t *item) {		zmsg_t* msg = zmsg_recv(item->socket);	zframe_t* frame = zmsg_next(msg);	size_t fsize = zframe_size(frame);	const byte* fbytes = zframe_data(frame);		proto::Signal& sig = mSignal;	sig.ParseFromArray(fbytes+1, fsize-1);		if(sig.type() == proto::Signal::NEWBLOCK){				mCurrBlock = sig.block();		mCurrHeight = mCurrBlock.height();		//printf("HandleInput(): proto::Signal::NEWBLOCK %d/n", mCurrHeight);				zmsg_send(&msg, mSignals);				while(true){						while(vNodes.empty())				MilliSleep(1000);						mIndexPrev = pindexBest;						if(!mIndexPrev)				MilliSleep(1000);			else				break;					}				mWorkerCount = mNonceMap.size();				mNonceMap.clear();		mReqNonces.clear();		mShares.clear();				if(mBlockTemplate)			delete mBlockTemplate;				mBlockTemplate = CreateNewBlock(mReserveKey);		if(!mBlockTemplate){			printf("ERROR: CreateNewBlock() failed./n");			return -1;		}			}else if(sig.type() == proto::Signal::SHUTDOWN){				printf("HandleInput(): proto::Signal::SHUTDOWN/n");				zmsg_send(&msg, mSignals);				FlushStats();				return -1;			}		zmsg_destroy(&msg);	return 0;	}

int main (int argc, char *argv []){    //  Arguments can be either of:    //      -p  primary server, at tcp://localhost:5001    //      -b  backup server, at tcp://localhost:5002    zctx_t *ctx = zctx_new ();    void *statepub = zsocket_new (ctx, ZMQ_PUB);    void *statesub = zsocket_new (ctx, ZMQ_SUB);    zsockopt_set_subscribe (statesub, "");    void *frontend = zsocket_new (ctx, ZMQ_ROUTER);    bstar_t fsm = { 0 };    if (argc == 2 && streq (argv [1], "-p")) {        printf ("I: Primary master, waiting for backup (slave)/n");        zsocket_bind (frontend, "tcp://*:5001");        zsocket_bind (statepub, "tcp://*:5003");        zsocket_connect (statesub, "tcp://localhost:5004");        fsm.state = STATE_PRIMARY;    }    else    if (argc == 2 && streq (argv [1], "-b")) {        printf ("I: Backup slave, waiting for primary (master)/n");        zsocket_bind (frontend, "tcp://*:5002");        zsocket_bind (statepub, "tcp://*:5004");        zsocket_connect (statesub, "tcp://localhost:5003");        fsm.state = STATE_BACKUP;    }    else {        printf ("Usage: bstarsrv { -p | -b }/n");        zctx_destroy (&ctx);        exit (0);    }    //  Set timer for next outgoing state message    int64_t send_state_at = zclock_time () + HEARTBEAT;    while (!zctx_interrupted) {        zmq_pollitem_t items [] = {            { frontend, 0, ZMQ_POLLIN, 0 },            { statesub, 0, ZMQ_POLLIN, 0 }        };        int time_left = (int) ((send_state_at - zclock_time ()));        if (time_left < 0)            time_left = 0;        int rc = zmq_poll (items, 2, time_left * ZMQ_POLL_MSEC);        if (rc == -1)            break;              //  Context has been shut down        if (items [0].revents & ZMQ_POLLIN) {            //  Have a client request            zmsg_t *msg = zmsg_recv (frontend);            fsm.event = CLIENT_REQUEST;            if (s_state_machine (&fsm) == FALSE)                //  Answer client by echoing request back                zmsg_send (&msg, frontend);            else                zmsg_destroy (&msg);        }        if (items [1].revents & ZMQ_POLLIN) {            //  Have state from our peer, execute as event            char *message = zstr_recv (statesub);            fsm.event = atoi (message);            free (message);            if (s_state_machine (&fsm))                break;          //  Error, so exit            fsm.peer_expiry = zclock_time () + 2 * HEARTBEAT;        }        //  If we timed-out, send state to peer        if (zclock_time () >= send_state_at) {            char message [2];            sprintf (message, "%d", fsm.state);            zstr_send (statepub, message);            send_state_at = zclock_time () + HEARTBEAT;        }    }    if (zctx_interrupted)        printf ("W: interrupted/n");    //  Shutdown sockets and context    zctx_destroy (&ctx);    return 0;}