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

staticint process_meta_information_and_handle_heartbeat(subscriber_state_t *state, zmsg_t* msg){    zframe_t *first = zmsg_first(msg);    char *pub_spec = NULL;    bool is_heartbeat = zframe_streq(first, "heartbeat");    msg_meta_t meta;    int rc = msg_extract_meta_info(msg, &meta);    if (!rc) {        // dump_meta_info(&meta);        if (!state->meta_info_failures++)            fprintf(stderr, "[E] subscriber: received invalid meta info/n");        return is_heartbeat;    }    if (meta.device_number == 0) {        // ignore device number 0        state->messages_dev_zero++;        return is_heartbeat;    }    if (is_heartbeat) {        if (debug)            printf("received heartbeat from device %d/n", meta.device_number);        zmsg_first(msg); // msg_extract_meta_info repositions the pointer, so reset        zframe_t *spec_frame = zmsg_next(msg);        pub_spec = zframe_strdup(spec_frame);    }    state->message_gap_size += device_tracker_calculate_gap(state->tracker, &meta, pub_spec);    return is_heartbeat;}

void pss_response(void *spss, req_store_t * req_store, void *sweb, void *sgraph){	zmsg_t *msg = zmsg_recv(spss);	zframe_t *null = zmsg_unwrap(msg);	zframe_destroy(&null);	json_error_t error;	printf("/nbroker:spss received: %s/n",	       (const char *)zframe_data(zmsg_first(msg)));	const char *data;	size_t data_size = zframe_size(zmsg_first(msg));	data = zframe_data(zmsg_first(msg));	json_t *pss_resp_json = json_loadb(data,					   data_size, 0, &error);	zmsg_destroy(&msg);	//identify the request	int32_t requestId =	    json_integer_value(json_object_get(pss_resp_json, "requestId"));	req_t *req = request_store_req(req_store, requestId);	json_t *response = json_object_get(pss_resp_json, "response");	json_incref(response);	json_decref(pss_resp_json);	const char *resp_type =	    json_string_value(json_object_get(response, "type"));	const char *req_type =	    json_string_value(json_object_get			      (json_object_get			       (json_object_get(req->request, "clientRequest"),				"request"), "type"));	if ((strcmp(resp_type, "searchResponse") == 0)	    && (strcmp(req_type, "searchRequest") == 0))		pss_response_searchResponse(req, response, requestId, sgraph);	else if ((strcmp(resp_type, "newNodeResponse") == 0)		 && (strcmp(req_type, "newNode") == 0))		pss_response_newNodeResponse(req, response, requestId, sweb,					     req_store);	else if ((strcmp(resp_type, "delNode") == 0)		 && (strcmp(req_type, "delNode") == 0))		pss_response_delNode(req, response, requestId, sweb, req_store);	else {	}}

// 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, strlen(WORKER_READY)) == 0) {			zmsg_destroy(&msg);		} else {			zmsg_send(&msg, self->frontend);		}	}	return 0;}

intzsock_wait (void *self){    assert (self);    //  A signal is a message containing one frame with our 8-byte magic     //  value. If we get anything else, we discard it and continue to look    //  for the signal message    while (true) {        zmsg_t *msg = zmsg_recv (self);        if (!msg)            return -1;        if (zmsg_size (msg) == 1        &&  zmsg_content_size (msg) == 8) {            zframe_t *frame = zmsg_first (msg);            int64_t signal_value = *((int64_t *) zframe_data (frame));            if ((signal_value & 0xFFFFFFFFFFFFFF00L) == 0x7766554433221100L) {                zmsg_destroy (&msg);                return signal_value & 255;            }        }        zmsg_destroy (&msg);    }    return -1;}

/* ================ download_data() ================ */int download_data(zsock_t *sock, const char *key){    /* ---------------- Send Message ---------------- */    zmsg_t *download_msg = create_action_message(MSG_ACTION_GET);    message_add_key_data(download_msg, key, "", 0);    zmsg_send(&download_msg, sock);    /* ---------------- Receive Message ---------------- */    zmsg_t *recv_msg = zmsg_recv(sock);    if ( recv_msg == NULL ){        return -2;    }    /*zmsg_print(recv_msg);*/    int rc = -1;    if (message_check_status(recv_msg, MSG_STATUS_WORKER_NOTFOUND) == 0 ){        warning_log("Not Found. key=%s", key);        rc = 0;    } else if ( message_check_status(recv_msg, MSG_STATUS_WORKER_ERROR) == 0 ){        error_log("Return MSG_STATUS_WORKER_ERROR. key=%s", key);        rc = -1;    } else {        /*zmsg_print(recv_msg);*/        zframe_t *frame_msgtype = zmsg_first(recv_msg);        if ( frame_msgtype != NULL ){            int16_t msgtype = *(int16_t*)zframe_data(frame_msgtype);            if ( msgtype == MSGTYPE_DATA ){                zmsg_first(recv_msg);                zframe_t *frame_key = zmsg_next(recv_msg);                UNUSED const char *key = (const char *)zframe_data(frame_key);                zframe_t *frame_data = zmsg_next(recv_msg);                UNUSED const char *data =  (const char *)zframe_data(frame_data);                UNUSED uint32_t data_size = zframe_size(frame_data);                /*notice_log("Receive key:%s data_size:%d", key, data_size);*/                rc = 0;            }        }    }    zmsg_destroy(&recv_msg);    return rc;}

int16_t message_get_msgtype(zmsg_t *msg){    zframe_t *frame_msgtype = zmsg_first(msg);    if ( frame_msgtype != NULL && zframe_size(frame_msgtype) == sizeof(int16_t) ){        return *(int16_t*)zframe_data(frame_msgtype);    }     return MSGTYPE_UNKNOWN;}

int main (void){    zctx_t *ctx = zctx_new ();    void *frontend = zsocket_new (ctx, ZMQ_ROUTER);    void *backend = zsocket_new (ctx, ZMQ_ROUTER);    zsocket_bind (frontend, "tcp://*:5555");    //  For clients    zsocket_bind (backend,  "tcp://*:5556");    //  For workers    //  Queue of available workers    zlist_t *workers = zlist_new ();        //  The body of this example is exactly the same as lruqueue2.    //  .skip    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 address for LRU routing            zmsg_t *msg = zmsg_recv (backend);            if (!msg)                break;          //  Interrupted            zframe_t *address = zmsg_unwrap (msg);            zlist_append (workers, address);            //  Forward message to client if it's not a READY            zframe_t *frame = zmsg_first (msg);            if (memcmp (zframe_data (frame), LRU_READY, 1) == 0)                zmsg_destroy (&msg);            else                zmsg_send (&msg, frontend);        }        if (items [1].revents & ZMQ_POLLIN) {            //  Get client request, route to first available worker            zmsg_t *msg = zmsg_recv (frontend);            if (msg) {                zmsg_wrap (msg, (zframe_t *) zlist_pop (workers));                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;    //  .until}

size_tzmsg_encode (zmsg_t *self, byte **buffer){    assert (self);    assert (zmsg_is (self));    //  Calculate real size of buffer    size_t buffer_size = 0;    zframe_t *frame = zmsg_first (self);    while (frame) {        size_t frame_size = zframe_size (frame);        if (frame_size < 255)            buffer_size += frame_size + 1;        else            buffer_size += frame_size + 1 + 4;        frame = zmsg_next (self);    }    *buffer = (byte *) zmalloc (buffer_size);    if (*buffer) {        //  Encode message now        byte *dest = *buffer;        frame = zmsg_first (self);        while (frame) {            size_t frame_size = zframe_size (frame);            if (frame_size < 255) {                *dest++ = (byte) frame_size;                memcpy (dest, zframe_data (frame), frame_size);                dest += frame_size;            }            else {                *dest++ = 0xFF;                *dest++ = (frame_size >> 24) & 255;                *dest++ = (frame_size >> 16) & 255;                *dest++ = (frame_size >>  8) & 255;                *dest++ =  frame_size        & 255;                memcpy (dest, zframe_data (frame), frame_size);                dest += frame_size;            }            frame = zmsg_next (self);        }        assert ((dest - *buffer) == buffer_size);    }    return buffer_size;}

intzmsg_signal (zmsg_t *self){    if (zmsg_size (self) == 1    &&  zmsg_content_size (self) == 8) {        zframe_t *frame = zmsg_first (self);        int64_t signal_value = *((int64_t *) zframe_data (frame));        if ((signal_value & 0xFFFFFFFFFFFFFF00L) == 0x7766554433221100L)            return signal_value & 255;    }    return -1;}

int message_check_action(zmsg_t *msg, const char *action){    int16_t msgtype = message_get_msgtype(msg);    if ( msgtype == MSGTYPE_ACTION ){        zmsg_first(msg);        zframe_t *frame = zmsg_next(msg);        if ( frame != NULL ){            return memcmp(zframe_data(frame), action, strlen(action));        }    }    return -1;}

int message_check_heartbeat(zmsg_t *msg, const char *heartbeat){    int16_t msgtype = message_get_msgtype(msg);    if ( msgtype == MSGTYPE_HEARTBEAT ){        zmsg_first(msg);        zframe_t *frame = zmsg_next(msg);        if ( frame != NULL ){            return memcmp(zframe_data(frame), heartbeat, strlen(heartbeat));        }    }    return -1;}

int message_check_status(zmsg_t *msg, const char *status){    int16_t msgtype = message_get_msgtype(msg);    if ( msgtype == MSGTYPE_STATUS ){        zmsg_first(msg);        zframe_t *frame = zmsg_next(msg);        if ( frame != NULL ){            return memcmp(zframe_data(frame), status, strlen(status));        }    }    return -1;}

zframe_t *zmsg_unwrap (zmsg_t *self){    assert (self);    zframe_t *frame = zmsg_pop (self);    zframe_t *empty = zmsg_first (self);    if (zframe_size (empty) == 0) {        empty = zmsg_pop (self);        zframe_destroy (&empty);    }    return frame;}

size_tzmsg_encode (zmsg_t *self, byte **buffer){    assert (self);    //  Calculate real size of buffer    size_t buffer_size = 0;    zframe_t *frame = zmsg_first (self);    while (frame) {        size_t frame_size = zframe_size (frame);        if (frame_size < ZMSG_SHORT_LEN)            buffer_size += frame_size + 1;        else        if (frame_size < 0x10000)            buffer_size += frame_size + 3;        else            buffer_size += frame_size + 5;        frame = zmsg_next (self);    }    *buffer = malloc (buffer_size);    //  Encode message now    byte *dest = *buffer;    frame = zmsg_first (self);    while (frame) {        size_t frame_size = zframe_size (frame);        if (frame_size < ZMSG_SHORT_LEN) {            *dest++ = (byte) frame_size;            memcpy (dest, zframe_data (frame), frame_size);            dest += frame_size;        }        else        if (frame_size < 0x10000) {            *dest++ = ZMSG_SHORT_LEN;            *dest++ = (frame_size >> 8) & 255;            *dest++ =  frame_size       & 255;            memcpy (dest, zframe_data (frame), frame_size);            dest += frame_size;        }        else {

voidinterval_minit (interval_t ** interval, zmsg_t * msg){    *interval = malloc (sizeof (interval_t));    zframe_t *frame = zmsg_first (msg);    memcpy (&((*interval)->start), zframe_data (frame), zframe_size (frame));    frame = zmsg_next (msg);    memcpy (&((*interval)->end), zframe_data (frame), zframe_size (frame));}

voidzmsg_print (zmsg_t *self){    assert (self);    assert (zmsg_is (self));    if (!self) {        zsys_debug ("(NULL)");        return;    }    zframe_t *frame = zmsg_first (self);    while (frame) {        zframe_print (frame, NULL);        frame = zmsg_next (self);    }}

static zmsg_t *_thsafe_zmq_client_recv_confirmation (smio_t *self){    DBE_DEBUG (DBG_MSG | DBG_LVL_TRACE, "[smio_thsafe_client:zmq] Calling _thsafe_zmq_client_recv_confirmation/n");    assert (self);    /* Wait for response */    zmsg_t *recv_msg = zmsg_recv (self->pipe);    /* Do not pop the message, just set a cursor to it */    zframe_t *reply_frame = zmsg_first (recv_msg);    DBE_DEBUG (DBG_MSG | DBG_LVL_TRACE, "[smio_thsafe_client:zmq] Receiving message:/n");#ifdef LOCAL_MSG_DBG    zmsg_print (recv_msg);#endif    /* Message is:     * frame 0: Reply code */    if (reply_frame == NULL) {        /* Interrupted or malformed message */        goto err_recv_data;    }    /* Check if the frame has the correct number of bytes */    if (zframe_size (reply_frame) != THSAFE_REPLY_SIZE) {        goto err_recv_data;    }    uint8_t *raw_data = (uint8_t *) zframe_data (reply_frame);    ASSERT_TEST(raw_data != NULL, "Could not receive confirmation code", err_null_raw_data);    uint32_t reply_code = *(uint32_t *) raw_data;    /* Check for confirmation */    if (reply_code != THSAFE_OK) {        DBE_DEBUG (DBG_MSG | DBG_LVL_TRACE, "[smio_thsafe_client:zmq] Received reply code OK/n");        goto err_reply_code_not_ok;    }    /* Caller owns the message and is its responsability to destroy it */    return recv_msg;/* TODO: reduce code repetition */err_reply_code_not_ok:err_null_raw_data:err_recv_data:    zmsg_destroy (&recv_msg);    return NULL;}

//  --------------------------------------------------------------------------//  Save message to an open file, return 0 if OK, else -1.intzmsg_save (zmsg_t *self, FILE *file){    assert (self);    assert (file);    zframe_t *frame = zmsg_first (self);    while (frame) {        size_t frame_size = zframe_size (frame);        if (fwrite (&frame_size, sizeof (frame_size), 1, file) != 1)            return -1;        if (fwrite (zframe_data (frame), frame_size, 1, file) != 1)            return -1;        frame = zmsg_next (self);    }    return 0;}

voidzmsg_fprint (zmsg_t *self, FILE *file){    assert (self);    assert (zmsg_is (self));    fprintf (file, "--------------------------------------/n");    if (!self) {        fprintf (file, "NULL");        return;    }    zframe_t *frame = zmsg_first (self);    int frame_nbr = 0;    while (frame && frame_nbr++ < 10) {        zframe_fprint (frame, NULL, file);        frame = zmsg_next (self);    }}

zmsg_t *zmsg_dup (zmsg_t *self){    assert (self);    assert (zmsg_is (self));    zmsg_t *copy = zmsg_new ();    if (!copy)        return NULL;    zframe_t *frame = zmsg_first (self);    while (frame) {        if (zmsg_addmem (copy, zframe_data (frame), zframe_size (frame))) {            zmsg_destroy (&copy);            return NULL;        }        frame = zmsg_next (self);    }    return copy;}

zmsg_t *zmsg_dup (zmsg_t *self){    if (self) {        assert (zmsg_is (self));        zmsg_t *copy = zmsg_new ();        if (copy) {            zframe_t *frame = zmsg_first (self);            while (frame) {                if (zmsg_addmem (copy, zframe_data (frame), zframe_size (frame))) {                    zmsg_destroy (&copy);                    break;      //  Abandon attempt to copy message                }                frame = zmsg_next (self);            }        }        return copy;    }    else        return NULL;}

static voidhandle_frontend (GPPWorker *self){  GPPWorkerPrivate *priv = GET_PRIV (self);  GPPWorkerClass *klass = GPP_WORKER_GET_CLASS (self);  zmsg_t *msg = zmsg_recv (priv->frontend);  if (!msg)    return;  if (zmsg_size (msg) == 3) {    char *request = zframe_strdup (zmsg_last (msg));    g_info ("I: normal reply/n");    priv->liveness = HEARTBEAT_LIVENESS;    priv->current_task = msg;    if (!klass->handle_request (self, request))      gpp_worker_set_task_done (self, NULL, FALSE);    free (request);  } else {    if (zmsg_size (msg) == 1) {      zframe_t *frame = zmsg_first (msg);      if (memcmp (zframe_data (frame), PPP_HEARTBEAT, 1) == 0) {        priv->liveness = HEARTBEAT_LIVENESS;        g_debug ("got heartbeat from queue !/n");      } else {        g_warning ("E: invalid message/n");        zmsg_dump (msg);      }      zmsg_destroy (&msg);    }    else {      g_warning ("E: invalid message/n");      zmsg_dump (msg);    }  }  priv->interval = INTERVAL_INIT;}

//.........这里部分代码省略.........    //  Queue of available workers    int local_capacity = 0;    int cloud_capacity = 0;    zlist_t *workers = zlist_new ();    while (1) {        zmq_pollitem_t primary [] = {            { localbe, 0, ZMQ_POLLIN, 0 },            { cloudbe, 0, ZMQ_POLLIN, 0 },            { statefe, 0, ZMQ_POLLIN, 0 },            { monitor, 0, ZMQ_POLLIN, 0 }        };        //  If we have no workers anyhow, wait indefinitely        int rc = zmq_poll (primary, 4,                           local_capacity? 1000 * ZMQ_POLL_MSEC: -1);        if (rc == -1)            break;              //  Interrupted        //  Track if capacity changes during this iteration        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) {

int main (int argc, char *argv []){    //  First argument is this broker's name    //  Other arguments are our peers' names    //    if (argc < 2) {        printf ("syntax: peering2 me {you}.../n");        return 0;    }    self = argv [1];    printf ("I: preparing broker at %s.../n", self);    srandom ((unsigned) time (NULL));    zctx_t *ctx = zctx_new ();    //  Bind cloud frontend to endpoint    void *cloudfe = zsocket_new (ctx, ZMQ_ROUTER);    zsocket_set_identity (cloudfe, self);    zsocket_bind (cloudfe, "ipc://%s-cloud.ipc", self);    //  Connect cloud backend to all peers    void *cloudbe = zsocket_new (ctx, ZMQ_ROUTER);    zsocket_set_identity (cloudbe, self);    int argn;    for (argn = 2; argn < argc; argn++) {        char *peer = argv [argn];        printf ("I: connecting to cloud frontend at '%s'/n", peer);        zsocket_connect (cloudbe, "ipc://%s-cloud.ipc", peer);    }    //  Prepare local frontend and backend    void *localfe = zsocket_new (ctx, ZMQ_ROUTER);    zsocket_bind (localfe, "ipc://%s-localfe.ipc", self);    void *localbe = zsocket_new (ctx, ZMQ_ROUTER);    zsocket_bind (localbe, "ipc://%s-localbe.ipc", self);    //  Get user to tell us when we can start...    printf ("Press Enter when all brokers are started: ");    getchar ();    //  Start local workers    int worker_nbr;    for (worker_nbr = 0; worker_nbr < NBR_WORKERS; worker_nbr++)        zthread_new (worker_task, NULL);    //  Start local clients    int client_nbr;    for (client_nbr = 0; client_nbr < NBR_CLIENTS; client_nbr++)        zthread_new (client_task, NULL);    //  .split request-reply handling    //  Here, we handle the request-reply flow. We're using load-balancing    //  to poll workers at all times, and clients only when there are one     //  or more workers available.    //  Least recently used queue of available workers    int capacity = 0;    zlist_t *workers = zlist_new ();    while (true) {        //  First, route any waiting replies from workers        zmq_pollitem_t backends [] = {            { 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])//.........这里部分代码省略.........

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){  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;}

//.........这里部分代码省略.........    assert (file);    fclose (file);    file = fopen ("zmsg.test", "r");    zmsg_t *null_msg = zmsg_load (NULL, file);    assert (null_msg == NULL);    fclose (file);    remove ("zmsg.test");    //  Save to a file, read back    file = fopen ("zmsg.test", "w");    assert (file);    rc = zmsg_save (msg, file);    assert (rc == 0);    fclose (file);    file = fopen ("zmsg.test", "r");    rc = zmsg_save (msg, file);    assert (rc == -1);    fclose (file);    zmsg_destroy (&msg);    file = fopen ("zmsg.test", "r");    msg = zmsg_load (NULL, file);    assert (msg);    fclose (file);    remove ("zmsg.test");    assert (zmsg_size (msg) == 10);    assert (zmsg_content_size (msg) == 60);    //  Remove all frames except first and last    int frame_nbr;    for (frame_nbr = 0; frame_nbr < 8; frame_nbr++) {        zmsg_first (msg);        frame = zmsg_next (msg);        zmsg_remove (msg, frame);        zframe_destroy (&frame);    }    //  Test message frame manipulation    assert (zmsg_size (msg) == 2);    frame = zmsg_last (msg);    assert (zframe_streq (frame, "Frame9"));    assert (zmsg_content_size (msg) == 12);    frame = zframe_new ("Address", 7);    assert (frame);    zmsg_prepend (msg, &frame);    assert (zmsg_size (msg) == 3);    rc = zmsg_addstr (msg, "Body");    assert (rc == 0);    assert (zmsg_size (msg) == 4);    frame = zmsg_pop (msg);    zframe_destroy (&frame);    assert (zmsg_size (msg) == 3);    char *body = zmsg_popstr (msg);    assert (streq (body, "Frame0"));    free (body);    zmsg_destroy (&msg);    //  Test encoding/decoding    msg = zmsg_new ();    assert (msg);    byte *blank = (byte *) zmalloc (100000);    assert (blank);    rc = zmsg_addmem (msg, blank, 0);    assert (rc == 0);    rc = zmsg_addmem (msg, blank, 1);

int main (void){    zctx_t *ctx = zctx_new ();    void *frontend = zsocket_new (ctx, ZMQ_ROUTER);    void *backend = zsocket_new (ctx, ZMQ_ROUTER);    zsocket_bind (frontend, "tcp://*:5555");    //  For clients    zsocket_bind (backend,  "tcp://*:5556");    //  For workers    //  List of available workers    zlist_t *workers = zlist_new ();    //  Send out heartbeats at regular intervals    uint64_t heartbeat_at = zclock_time () + HEARTBEAT_INTERVAL;    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,            HEARTBEAT_INTERVAL * ZMQ_POLL_MSEC);        if (rc == -1)            break;              //  Interrupted        //  Handle worker activity on backend        if (items [0].revents & ZMQ_POLLIN) {            //  Use worker address for LRU routing            zmsg_t *msg = zmsg_recv (backend);            if (!msg)                break;          //  Interrupted            //  Any sign of life from worker means it's ready            zframe_t *address = zmsg_unwrap (msg);            worker_t *worker = s_worker_new (address);            s_worker_ready (worker, workers);            //  Validate control message, or return reply to client            if (zmsg_size (msg) == 1) {                zframe_t *frame = zmsg_first (msg);                if (memcmp (zframe_data (frame), PPP_READY, 1)                &&  memcmp (zframe_data (frame), PPP_HEARTBEAT, 1)) {                    printf ("E: invalid message from worker");                    zmsg_dump (msg);                }                zmsg_destroy (&msg);            }            else                zmsg_send (&msg, frontend);        }        if (items [1].revents & ZMQ_POLLIN) {            //  Now get next client request, route to next worker            zmsg_t *msg = zmsg_recv (frontend);            if (!msg)                break;          //  Interrupted            zmsg_push (msg, s_workers_next (workers));            zmsg_send (&msg, backend);        }        //  .split handle heartbeating        //  We handle heartbeating after any socket activity. First we send        //  heartbeats to any idle workers if it's time. Then we purge any        //  dead workers:                if (zclock_time () >= heartbeat_at) {            worker_t *worker = (worker_t *) zlist_first (workers);            while (worker) {                zframe_send (&worker->address, backend,                             ZFRAME_REUSE + ZFRAME_MORE);                zframe_t *frame = zframe_new (PPP_HEARTBEAT, 1);                zframe_send (&frame, backend, 0);                worker = (worker_t *) zlist_next (workers);            }            heartbeat_at = zclock_time () + HEARTBEAT_INTERVAL;        }        s_workers_purge (workers);    }    //  When we're done, clean up properly    while (zlist_size (workers)) {        worker_t *worker = (worker_t *) zlist_pop (workers);        s_worker_destroy (&worker);    }    zlist_destroy (&workers);    zctx_destroy (&ctx);    return 0;}

int main(int argc, char* argv[]){  int i, client_num, worker_num;  zctx_t* ctx;  void* frontend;  void* backend;  zlist_t* workers;  if (argc < 3) {    fprintf(stderr, "arguments error .../n");    return 1;  }  client_num = atoi(argv[1]);  worker_num = atoi(argv[2]);  ctx = zctx_new();  frontend = zsocket_new(ctx, ZMQ_ROUTER);  backend = zsocket_new(ctx, ZMQ_ROUTER);  zsocket_bind(frontend, "ipc://frontend.ipc");  zsocket_bind(backend, "ipc://backend.ipc");  for (i = 0; i < client_num; ++i)    zthread_new(client_routine, NULL);  for (i = 0; i < worker_num; ++i)    zthread_new(worker_routine, NULL);  workers = zlist_new();  while (1) {    zmq_pollitem_t items[] = {      {backend, 0, ZMQ_POLLIN, 0},       {frontend, 0, ZMQ_POLLIN, 0},     };    int rc = zmq_poll(items, zlist_size(workers) ? 2 : 1, -1);    if (-1 == rc)      break;    if (items[0].revents & ZMQ_POLLIN) {      zmsg_t* msg;      zframe_t* identity;      zframe_t* frame;      msg = zmsg_recv(backend);      if (NULL == msg)        break;      identity = zmsg_unwrap(msg);      zlist_append(workers, identity);      frame = zmsg_first(msg);      if (0 == memcmp(zframe_data(frame), WORKER_READY, 1))        zmsg_destroy(&msg);      else         zmsg_send(&msg, frontend);    }    if (items[1].revents & ZMQ_POLLIN) {      zmsg_t* msg = zmsg_recv(frontend);      if (NULL != msg) {        zmsg_wrap(msg, (zframe_t*)zlist_pop(workers));        zmsg_send(&msg, backend);      }    }  }  while (zlist_size(workers)) {    zframe_t* frame = (zframe_t*)zlist_pop(workers);    zframe_destroy(&frame);  }  zlist_destroy(&workers);  zctx_destroy(&ctx);  return 0;}