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

static int job_request_cb (flux_t h, int typemask, zmsg_t **zmsg, void *arg){    const char *json_str;    json_object *o = NULL;    const char *topic;    if (flux_msg_get_topic (*zmsg, &topic) < 0)        goto out;    if (flux_msg_get_payload_json (*zmsg, &json_str) < 0)        goto out;    if (json_str && !(o = json_tokener_parse (json_str)))        goto out;    if (strcmp (topic, "job.shutdown") == 0) {        flux_reactor_stop (h);    }    if (strcmp (topic, "job.next-id") == 0) {        if (flux_rank (h) == 0) {            unsigned long id = lwj_next_id (h);            json_object *ox = json_id (id);            flux_json_respond (h, ox, zmsg);            json_object_put (o);        }        else {            fprintf (stderr, "%s: forwarding request/n", topic);            flux_json_request (h, FLUX_NODEID_ANY,                                  FLUX_MATCHTAG_NONE, topic, o);        }    }    if (strcmp (topic, "job.create") == 0) {        json_object *jobinfo = NULL;        unsigned long id = lwj_next_id (h);        bool should_workaround = false;        //"Fix" for Race Condition        if (util_json_object_get_boolean (o, "race_workaround",                                           &should_workaround) < 0) {            should_workaround = false;        } else if (should_workaround) {            if (wait_for_lwj_watch_init (h, id) < 0) {              flux_err_respond (h, errno, zmsg);              goto out;            }        }        int rc = kvs_job_new (h, id);        if (rc < 0) {            flux_err_respond (h, errno, zmsg);            goto out;        }        add_jobinfo (h, id, o);        kvs_commit (h);        /* Generate reply with new jobid */        jobinfo = util_json_object_new_object ();        util_json_object_add_int64 (jobinfo, "jobid", id);        flux_json_respond (h, jobinfo, zmsg);        json_object_put (jobinfo);    }out:    if (o)        json_object_put (o);    zmsg_destroy (zmsg);    return 0;}

static ints_agent_handle_pipe (agent_t *self){    //  Get the whole message off the pipe in one go    zmsg_t *request = zmsg_recv (self->pipe);    char *command = zmsg_popstr (request);    if (!command)        return -1;                  //  Interrupted    if (streq (command, "ALLOW")) {        char *address = zmsg_popstr (request);        zhash_insert (self->whitelist, address, "OK");        zstr_free (&address);        zstr_send (self->pipe, "OK");    }    else    if (streq (command, "DENY")) {        char *address = zmsg_popstr (request);        zhash_insert (self->blacklist, address, "OK");        zstr_free (&address);        zstr_send (self->pipe, "OK");    }    else    if (streq (command, "PLAIN")) {        //  For now we don't do anything with domains        char *domain = zmsg_popstr (request);        zstr_free (&domain);        //  Get password file and load into zhash table        //  If the file doesn't exist we'll get an empty table        char *filename = zmsg_popstr (request);        zhash_destroy (&self->passwords);        self->passwords = zhash_new ();        zhash_load (self->passwords, filename);        zstr_free (&filename);        zstr_send (self->pipe, "OK");    }    else    if (streq (command, "CURVE")) {        //  For now we don't do anything with domains        char *domain = zmsg_popstr (request);        zstr_free (&domain);        //  If location is CURVE_ALLOW_ANY, allow all clients. Otherwise         //  treat location as a directory that holds the certificates.        char *location = zmsg_popstr (request);        if (streq (location, CURVE_ALLOW_ANY))            self->allow_any = true;        else {            zcertstore_destroy (&self->certstore);            self->certstore = zcertstore_new (location);            self->allow_any = false;        }        zstr_free (&location);        zstr_send (self->pipe, "OK");    }    else    if (streq (command, "VERBOSE")) {        char *verbose = zmsg_popstr (request);        self->verbose = *verbose == '1';        zstr_free (&verbose);        zstr_send (self->pipe, "OK");    }    else    if (streq (command, "TERMINATE")) {        self->terminated = true;        zstr_send (self->pipe, "OK");    }    else {        printf ("E: invalid command from API: %s/n", command);        assert (false);    }    zstr_free (&command);    zmsg_destroy (&request);    return 0;}

int main (int argc, char *argv[]){    //  Prepare our context and sockets    void *context = zmq_init (1);    void *frontend = zmq_socket (context, ZMQ_XREP);    void *backend  = zmq_socket (context, ZMQ_XREP);    zmq_bind (frontend, "ipc://frontend.ipc");    zmq_bind (backend,  "ipc://backend.ipc");    int client_nbr;    for (client_nbr = 0; client_nbr < NBR_CLIENTS; client_nbr++) {        pthread_t client;        pthread_create (&client, NULL, client_thread, context);    }    int worker_nbr;    for (worker_nbr = 0; worker_nbr < NBR_WORKERS; worker_nbr++) {        pthread_t worker;        pthread_create (&worker, NULL, worker_thread, context);    }    //  Logic of LRU loop    //  - Poll backend always, frontend only if 1+ worker ready    //  - If worker replies, queue worker as ready and forward reply    //    to client if necessary    //  - If client requests, pop next worker and send request to it    //  Queue of available workers    int available_workers = 0;    char *worker_queue [NBR_WORKERS];    while (1) {        //  Initialize poll set        zmq_pollitem_t items [] = {            //  Always poll for worker activity on backend            { backend,  0, ZMQ_POLLIN, 0 },            //  Poll front-end only if we have available workers            { frontend, 0, ZMQ_POLLIN, 0 }        };        if (available_workers)            zmq_poll (items, 2, -1);        else            zmq_poll (items, 1, -1);        //  Handle worker activity on backend        if (items [0].revents & ZMQ_POLLIN) {            zmsg_t *zmsg = zmsg_recv (backend);            //  Use worker address for LRU routing            assert (available_workers < NBR_WORKERS);            worker_queue [available_workers++] = zmsg_unwrap (zmsg);            //  Forward message to client if it's not a READY            if (strcmp (zmsg_address (zmsg), "READY") == 0)                zmsg_destroy (&zmsg);            else {                zmsg_send (&zmsg, frontend);                if (--client_nbr == 0)                    break;      //  Exit after N messages            }        }        if (items [1].revents & ZMQ_POLLIN) {            //  Now get next client request, route to next worker            zmsg_t *zmsg = zmsg_recv (frontend);            zmsg_wrap (zmsg, worker_queue [0], "");            zmsg_send (&zmsg, backend);            //  Dequeue and drop the next worker address            free (worker_queue [0]);            DEQUEUE (worker_queue);            available_workers--;        }    }    sleep (1);    zmq_term (context);    return 0;}

zyre_event_t *zyre_event_new (zyre_t *node){    zmsg_t *msg = zyre_recv (node);    if (!msg)        return NULL;            //  Interrupted    zyre_event_t *self = (zyre_event_t *) zmalloc (sizeof (zyre_event_t));    assert (self);    char *type = zmsg_popstr (msg);    self->sender = zmsg_popstr (msg);    self->name = zmsg_popstr (msg);    if (streq (type, "ENTER")) {        self->type = ZYRE_EVENT_ENTER;        zframe_t *headers = zmsg_pop (msg);        if (headers) {            self->headers = zhash_unpack (headers);            zframe_destroy (&headers);        }        self->address = zmsg_popstr (msg);    }    else    if (streq (type, "EXIT"))        self->type = ZYRE_EVENT_EXIT;    else    if (streq (type, "JOIN")) {        self->type = ZYRE_EVENT_JOIN;        self->group = zmsg_popstr (msg);    }    else    if (streq (type, "LEAVE")) {        self->type = ZYRE_EVENT_LEAVE;        self->group = zmsg_popstr (msg);    }    else    if (streq (type, "WHISPER")) {        self->type = ZYRE_EVENT_WHISPER;        self->msg = msg;        msg = NULL;    }    else    if (streq (type, "SHOUT")) {        self->type = ZYRE_EVENT_SHOUT;        self->group = zmsg_popstr (msg);        self->msg = msg;        msg = NULL;    }    else    if (streq (type, "STOP")) {        self->type = ZYRE_EVENT_STOP;    }    else    if (streq (type, "EVASIVE")) {        self->type = ZYRE_EVENT_EVASIVE;    }    else        zsys_warning ("bad message received from node: %s/n", type);    free (type);    zmsg_destroy (&msg);    return self;}

void watch_port(void *cvoid,                 zctx_t * context,                 void * pipe ) {  zclock_log("watch_port started!");  monitorconfig_t * config = (monitorconfig_t*) cvoid;  dump_monitorconfig(config);  void * linein = zsocket_new(context, ZMQ_SUB);  char * listen_socket = to_linesocket(config->line_id);  char line_id[16];  snprintf(line_id, 15, "%d", config->line_id);  zsocket_connect(linein, listen_socket);  zsockopt_set_unsubscribe(linein, "");  zsockopt_set_subscribe(linein, "CLEAR_MONITORS");  zsockopt_set_subscribe(linein, "VALUE");  // have set up subscription, can signal parent that we're ok.  child_handshake(pipe);  zsocket_destroy(context, pipe); // no longer require pipe  void * lineout = zsocket_new(context, ZMQ_PUB);  zsocket_connect(lineout, config->out_socket);  time_t until = time(NULL) + 60;  while(time(NULL)<until) {    zmsg_t * msg = zmsg_recv(linein);    if(!msg) {      zclock_log("monitor quitting!");      return;    }    zframe_t * cmd = zmsg_pop(msg);    if(zframe_streq(cmd, "CLEAR_MONITORS")) {      zclock_log("ephemeral monitor quitting");      zmsg_destroy(&msg);      zframe_destroy(&cmd);      break;    } else if (zframe_streq(cmd, "VALUE")) {      // TODO perhaps some rate limiting necessary      assert(zmsg_size(msg) == 2);            zframe_t * value = zmsg_pop(msg);      int res = *(int*)zframe_data(value);      char * new_channel = zmsg_popstr(msg);      if(strcmp(new_channel, config->channel)!=0) {        zclock_log("monitor on %d: listening for %s, channel changed to %s quitting",                   config->line_id, config->channel, new_channel);        zmsg_destroy(&msg);        zframe_destroy(&cmd);        break;      }      zmsg_t * to_send = zmsg_new();      char buf[1024];      snprintf(buf,1023, "%d", res);      zmsg_pushstr(to_send, buf);      zmsg_pushstr(to_send, line_id);      zmsg_pushstr(to_send, config->source_worker);      zclock_log("%s sending line %s -> %s", config->source_worker, line_id, buf);      zmsg_send(&to_send, lineout);      // don't destroy value frame, now owned by zmsg    }    // else ignore    zmsg_destroy(&msg);    zframe_destroy(&cmd);  }  zclock_log("monitor on %d: listening for %s, expiring naturally",             config->line_id, config->channel);  //cleanup  zsocket_destroy(context, linein);  zsocket_destroy(context, lineout);}

static intzyre_node_recv_api (zyre_node_t *self){    //  Get the whole message off the pipe in one go    zmsg_t *request = zmsg_recv (self->pipe);    char *command = zmsg_popstr (request);    if (!command)        return -1;                  //  Interrupted    if (streq (command, "SET")) {        char *name = zmsg_popstr (request);        char *value = zmsg_popstr (request);        zhash_update (self->headers, name, value);        zstr_free (&name);        zstr_free (&value);    }    else    if (streq (command, "START")) {        zyre_node_start (self);        zstr_send (self->pipe, "OK");    }    else    if (streq (command, "STOP")) {        zyre_node_stop (self);        zstr_send (self->pipe, "OK");    }    else    if (streq (command, "WHISPER")) {        //  Get peer to send message to        char *identity = zmsg_popstr (request);        zyre_peer_t *peer = (zyre_peer_t *) zhash_lookup (self->peers, identity);                //  Send frame on out to peer's mailbox, drop message        //  if peer doesn't exist (may have been destroyed)        if (peer) {            zre_msg_t *msg = zre_msg_new (ZRE_MSG_WHISPER);            zre_msg_set_content (msg, request);            zyre_peer_send (peer, &msg);            request = NULL;        }        zstr_free (&identity);    }    else    if (streq (command, "SHOUT")) {        //  Get group to send message to        char *name = zmsg_popstr (request);        zyre_group_t *group = (zyre_group_t *) zhash_lookup (self->peer_groups, name);        if (group) {            zre_msg_t *msg = zre_msg_new (ZRE_MSG_SHOUT);            zre_msg_set_group (msg, name);            zre_msg_set_content (msg, request);            zyre_group_send (group, &msg);            request = NULL;        }        zstr_free (&name);    }    else    if (streq (command, "JOIN")) {        char *name = zmsg_popstr (request);        zyre_group_t *group = (zyre_group_t *) zhash_lookup (self->own_groups, name);        if (!group) {            //  Only send if we're not already in group            group = zyre_group_new (name, self->own_groups);            zre_msg_t *msg = zre_msg_new (ZRE_MSG_JOIN);            zre_msg_set_group (msg, name);            //  Update status before sending command            zre_msg_set_status (msg, ++(self->status));            zhash_foreach (self->peers, zyre_node_send_peer, msg);            zre_msg_destroy (&msg);            zyre_log_info (self->log, ZRE_LOG_MSG_EVENT_JOIN, NULL, name);        }        zstr_free (&name);    }    else    if (streq (command, "LEAVE")) {        char *name = zmsg_popstr (request);        zyre_group_t *group = (zyre_group_t *) zhash_lookup (self->own_groups, name);        if (group) {            //  Only send if we are actually in group            zre_msg_t *msg = zre_msg_new (ZRE_MSG_LEAVE);            zre_msg_set_group (msg, name);            //  Update status before sending command            zre_msg_set_status (msg, ++(self->status));            zhash_foreach (self->peers, zyre_node_send_peer, msg);            zre_msg_destroy (&msg);            zhash_delete (self->own_groups, name);            zyre_log_info (self->log, ZRE_LOG_MSG_EVENT_LEAVE, NULL, name);        }        zstr_free (&name);    }    else    if (streq (command, "TERMINATE")) {        self->terminated = true;        zstr_send (self->pipe, "OK");    }    zstr_free (&command);    zmsg_destroy (&request);    return 0;}

static ints_agent_handle_control (agent_t *self){    //  Get the whole message off the control socket in one go    zmsg_t *request = zmsg_recv (self->control);    char *command = zmsg_popstr (request);    if (!command)        return -1;                  //  Interrupted    if (streq (command, "SET")) {        char *name = zmsg_popstr (request);        char *value = zmsg_popstr (request);        zhash_insert (self->metadata, name, value);        free (name);        free (value);    }    else    if (streq (command, "VERBOSE")) {        char *verbose = zmsg_popstr (request);        self->verbose = *verbose == '1';        free (verbose);    }    else    if (streq (command, "MAX CLIENTS")) {        char *limit = zmsg_popstr (request);        self->max_clients = atoi (limit);        free (limit);    }    else    if (streq (command, "MAX PENDING")) {        char *limit = zmsg_popstr (request);        self->max_pending = atoi (limit);        free (limit);    }    else    if (streq (command, "CLIENT TTL")) {        char *limit = zmsg_popstr (request);        self->client_ttl = atoi (limit);        free (limit);    }    else    if (streq (command, "PENDING TTL")) {        char *limit = zmsg_popstr (request);        self->pending_ttl = atoi (limit);        free (limit);    }    else    if (streq (command, "BIND")) {        char *endpoint = zmsg_popstr (request);puts (endpoint);        int rc = zsocket_bind (self->router, "%s", endpoint);        assert (rc != -1);        free (endpoint);    }    else    if (streq (command, "UNBIND")) {        char *endpoint = zmsg_popstr (request);        int rc = zsocket_unbind (self->router, "%s", endpoint);        assert (rc != -1);        free (endpoint);    }    else    if (streq (command, "TERMINATE")) {        self->terminated = true;        zstr_send (self->control, "OK");    }    free (command);    zmsg_destroy (&request);    return 0;}

/////  Destroy a message object and all frames it containsQZmsg::~QZmsg (){    zmsg_destroy (&self);}

int main(int argc, char *argv[]){  char *broker;  broker = strdup("tcp://localhost:5555");  char *msg_str;  msg_str = strdup("empty");  while ((argc > 1) && (argv[1][0] == '-')) {    switch (argv[1][1]) {          case 'b':      free(broker);      broker = strdup(argv[2]);      ++argv;      --argc;      break;    case 'm':      free(msg_str);      msg_str = strdup(argv[2]);      ++argv;      --argc;      break;    default:	printf("Wrong Argument: %s/n", argv[1]);	usage();	break;    }    ++argv;    --argc;  }  printf("Connecting to broker %s .../n",broker);  zctx_t *ctx = zctx_new();  void *b_sock = zsocket_new(ctx, ZMQ_PUSH);  assert(b_sock);  int rc;  /* connecting to master */  rc = zsocket_connect(b_sock, broker);  assert(!rc);  zmsg_t *msg = zmsg_new();;  /* send packet */  rc = zmsg_addstr(msg, msg_str);  assert(rc == 0);    printf ("Sending msg='%s' .../n",msg_str);  /* zmsg_dump(msg); */      /* sending message */  rc = zmsg_send(&msg, b_sock);  assert(rc == 0);  /* wait for message to be sent */  zclock_sleep(100);  zmsg_destroy(&msg);    free(broker);  free(msg_str);    /* everything should be cleanly closed now */  zctx_destroy(&ctx);    return 0;}

static voidinterface_task (void *args, zctx_t *ctx, void *pipe){    zre_interface_t *interface = zre_interface_new ();    int64_t counter = 0;    char *to_peer = NULL;        //  Either of these set,    char *to_group = NULL;       //    and we set a message    char *cookie = NULL;         //  received message    char *sending_cookie = NULL; //  sending message        zmq_pollitem_t pollitems [] = {        { pipe,                             0, ZMQ_POLLIN, 0 },        { zre_interface_handle (interface), 0, ZMQ_POLLIN, 0 }    };    // all interface joins GLOBAL    zre_interface_join (interface, "GLOBAL");    while (!zctx_interrupted) {        if (zmq_poll (pollitems, 2, randof (1000) * ZMQ_POLL_MSEC) == -1)            break;              //  Interrupted        if (pollitems [0].revents & ZMQ_POLLIN)            break;              //  Any command from parent means EXIT        //  Process an event from interface        if (pollitems [1].revents & ZMQ_POLLIN) {            zmsg_t *incoming = zre_interface_recv (interface);            if (!incoming)                break;              //  Interrupted            char *event = zmsg_popstr (incoming);            if (streq (event, "ENTER")) {                //  Always say hello to new peer                to_peer = zmsg_popstr (incoming);                sending_cookie = "R:HELLO";            }            else            if (streq (event, "EXIT")) {                //  Do nothing            }            else            if (streq (event, "WHISPER")) {                to_peer = zmsg_popstr (incoming);                cookie = zmsg_popstr (incoming);                // if a message comes from zre_perf_local, send back a special response                if (streq (cookie, "S:WHISPER")) {                    sending_cookie = "R:WHISPER";                }                else {                    free (to_peer);                    free (cookie);                    to_peer = NULL;                    cookie = NULL;                }            }            else            if (streq (event, "SHOUT")) {                to_peer = zmsg_popstr (incoming);                to_group = zmsg_popstr (incoming);                cookie = zmsg_popstr (incoming);                // if a message comes from zre_perf_local, send back a special response                if (streq (cookie, "S:SHOUT")) {                    free (to_peer);                    to_peer = NULL;                    sending_cookie = "R:SHOUT";                }                else {                    free (to_peer);                    free (to_group);                    to_peer = NULL;                    to_group = NULL;                }            }            free (event);            zmsg_destroy (&incoming);            //  Send outgoing messages if needed            if (to_peer) {                zmsg_t *outgoing = zmsg_new ();                zmsg_addstr (outgoing, to_peer);                zmsg_addstr (outgoing, sending_cookie);                zre_interface_whisper (interface, &outgoing);                free (to_peer);                to_peer = NULL;            }            if (to_group) {                zmsg_t *outgoing = zmsg_new ();                zmsg_addstr (outgoing, to_group);                zmsg_addstr (outgoing, sending_cookie);                zre_interface_shout (interface, &outgoing);                free (to_group);                to_group = NULL;            }            if (cookie) {                free (cookie);                cookie = NULL;            }//.........这里部分代码省略.........

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 active, waiting for backup (passive)/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 passive, waiting for primary (active)/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);    }    //  .split handling socket input    //  We now process events on our two input sockets, and process these    //  events one at a time via our finite-state machine. Our "work" for    //  a client request is simply to echo it back:    //  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;}

voidzyre_event_test (bool verbose){    printf (" * zyre_event: ");    //  @selftest    //  Create two nodes    zyre_t *node1 = zyre_new ("node1");    assert (node1);    zyre_set_header (node1, "X-HELLO", "World");    int rc = zyre_set_endpoint (node1, "inproc://zyre-node1");    assert (rc == 0);    // use gossiping instead of beaconing, suits Travis better    zyre_gossip_bind (node1, "inproc://gossip-hub");    if (verbose)        zyre_set_verbose (node1);    if (zyre_start (node1)) {        zyre_destroy (&node1);        printf ("OK (skipping test, no UDP discovery)/n");        return;    }    zyre_join (node1, "GLOBAL");    zyre_t *node2 = zyre_new ("node2");    assert (node2);    if (verbose)        zyre_set_verbose (node2);    rc = zyre_set_endpoint (node2, "inproc://zyre-node2");    assert (rc == 0);    // use gossiping instead of beaconing, suits Travis better    zyre_gossip_connect (node2, "inproc://gossip-hub");    rc = zyre_start (node2);    assert (rc == 0);    zyre_join (node2, "GLOBAL");    //  Give time for them to interconnect    zclock_sleep (250);    //  One node shouts to GLOBAL    zmsg_t *msg = zmsg_new ();    zmsg_addstr (msg, "Hello, World");    zyre_shout (node1, "GLOBAL", &msg);    zclock_sleep (100);    //  Parse ENTER    zyre_event_t *event = zyre_event_new (node2);    assert (streq (zyre_event_type (event), "ENTER"));    const char *sender = zyre_event_peer_uuid (event);    assert (sender);    const char *name = zyre_event_peer_name (event);    assert (name);    assert (streq (name, "node1"));    const char *address = zyre_event_peer_addr (event);    assert (address);    const char *header = zyre_event_header (event, "X-HELLO");    assert (header);    zyre_event_destroy (&event);    //  Parse JOIN    //  We tolerate other events, which we can get if there are instances    //  of Zyre running somewhere on the network.    event = zyre_event_new (node2);    if (streq (zyre_event_type (event), "JOIN")) {        //  Parse SHOUT        zyre_event_destroy (&event);        event = zyre_event_new (node2);        if (streq (zyre_event_type (event), "SHOUT")) {            assert (streq (zyre_event_group (event), "GLOBAL"));            zmsg_t *msg = zyre_event_get_msg (event);            char *string = zmsg_popstr (msg);            zmsg_destroy (&msg);            assert (streq (string, "Hello, World"));            free (string);        }        zyre_event_destroy (&event);    }    zyre_destroy (&node1);    zyre_destroy (&node2);    //  @end    printf ("OK/n");}

zmsg_t *mdp_worker_recv (mdp_worker_t *self, zframe_t **reply_to_p){    while (TRUE) {        zmq_pollitem_t items [] = {            { self->worker,  0, ZMQ_POLLIN, 0 } };        int rc = zmq_poll (items, 1, self->heartbeat * ZMQ_POLL_MSEC);        if (rc == -1)            break;              //  Interrupted        if (items [0].revents & ZMQ_POLLIN) {            zmsg_t *msg = zmsg_recv (self->worker);            if (!msg)                break;          //  Interrupted            if (self->verbose) {                zclock_log ("I: received message from broker:");                zmsg_dump (msg);            }            self->liveness = HEARTBEAT_LIVENESS;            //  Don't try to handle errors, just assert noisily            assert (zmsg_size (msg) >= 3);            zframe_t *empty = zmsg_pop (msg);            assert (zframe_streq (empty, ""));            zframe_destroy (&empty);            zframe_t *header = zmsg_pop (msg);            assert (zframe_streq (header, MDPW_WORKER));            zframe_destroy (&header);            zframe_t *command = zmsg_pop (msg);            if (zframe_streq (command, MDPW_REQUEST)) {                //  We should pop and save as many addresses as there are                //  up to a null part, but for now, just save one...                zframe_t *reply_to = zmsg_unwrap (msg);                if (reply_to_p)                    *reply_to_p = reply_to;                else                    zframe_destroy (&reply_to);                zframe_destroy (&command);                //  Here is where we actually have a message to process; we                //  return it to the caller application                return msg;     //  We have a request to process            }            else            if (zframe_streq (command, MDPW_HEARTBEAT))                ;               //  Do nothing for heartbeats            else            if (zframe_streq (command, MDPW_DISCONNECT))                s_mdp_worker_connect_to_broker (self);            else {                zclock_log ("E: invalid input message");                zmsg_dump (msg);            }            zframe_destroy (&command);            zmsg_destroy (&msg);        }        else        if (--self->liveness == 0) {            if (self->verbose)                zclock_log ("W: disconnected from broker - retrying...");            zclock_sleep (self->reconnect);            s_mdp_worker_connect_to_broker (self);        }        //  Send HEARTBEAT if it's time        if (zclock_time () > self->heartbeat_at) {            s_mdp_worker_send_to_broker (self, MDPW_HEARTBEAT, NULL, NULL);            self->heartbeat_at = zclock_time () + self->heartbeat;        }    }    if (zctx_interrupted)        printf ("W: interrupt received, killing worker.../n");    return NULL;}

voidzmsg_test (bool verbose){    printf (" * zmsg: ");    int rc = 0;    //  @selftest    //  Create two PAIR sockets and connect over inproc    zsock_t *output = zsock_new_pair ("@inproc://zmsg.test");    assert (output);    zsock_t *input = zsock_new_pair (">inproc://zmsg.test");    assert (input);    //  Test send and receive of single-frame message    zmsg_t *msg = zmsg_new ();    assert (msg);    zframe_t *frame = zframe_new ("Hello", 5);    assert (frame);    zmsg_prepend (msg, &frame);    assert (zmsg_size (msg) == 1);    assert (zmsg_content_size (msg) == 5);    rc = zmsg_send (&msg, output);    assert (msg == NULL);    assert (rc == 0);    msg = zmsg_recv (input);    assert (msg);    assert (zmsg_size (msg) == 1);    assert (zmsg_content_size (msg) == 5);    zmsg_destroy (&msg);    //  Test send and receive of multi-frame message    msg = zmsg_new ();    assert (msg);    rc = zmsg_addmem (msg, "Frame0", 6);    assert (rc == 0);    rc = zmsg_addmem (msg, "Frame1", 6);    assert (rc == 0);    rc = zmsg_addmem (msg, "Frame2", 6);    assert (rc == 0);    rc = zmsg_addmem (msg, "Frame3", 6);    assert (rc == 0);    rc = zmsg_addmem (msg, "Frame4", 6);    assert (rc == 0);    rc = zmsg_addmem (msg, "Frame5", 6);    assert (rc == 0);    rc = zmsg_addmem (msg, "Frame6", 6);    assert (rc == 0);    rc = zmsg_addmem (msg, "Frame7", 6);    assert (rc == 0);    rc = zmsg_addmem (msg, "Frame8", 6);    assert (rc == 0);    rc = zmsg_addmem (msg, "Frame9", 6);    assert (rc == 0);    zmsg_t *copy = zmsg_dup (msg);    assert (copy);    rc = zmsg_send (&copy, output);    assert (rc == 0);    rc = zmsg_send (&msg, output);    assert (rc == 0);    copy = zmsg_recv (input);    assert (copy);    assert (zmsg_size (copy) == 10);    assert (zmsg_content_size (copy) == 60);    zmsg_destroy (&copy);    msg = zmsg_recv (input);    assert (msg);    assert (zmsg_size (msg) == 10);    assert (zmsg_content_size (msg) == 60);    // create empty file for null test    FILE *file = fopen ("zmsg.test", "w");    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);//.........这里部分代码省略.........

int main (int argc, char *argv []){    bool verbose;    if (argc == 2 && streq (argv [1], "-v")) {        argc--;        verbose = true;    }    else        verbose = false;    //  Do normal checks if run without arguments    if (argc < 2) {        printf ("Running self tests.../n");        fmq_patch_test (verbose);        fmq_dir_test (verbose);        fmq_msg_test (verbose);        fmq_sasl_test (verbose);        fmq_hash_test (verbose);        fmq_server_test (verbose);        fmq_client_test (verbose);        printf ("Tests passed OK/n");        return 0;    }    //  Else run as FILEMQ server or client    if (streq (argv [1], "-s")) {        fmq_server_t *server = fmq_server_new ();        fmq_server_configure (server, "server_test.cfg");        fmq_server_publish (server, "./fmqroot/send", "/");        fmq_server_publish (server, "./fmqroot/logs", "/logs");        //  We do this last        fmq_server_bind (server, "tcp://*:5670");        while (!zctx_interrupted)            zclock_sleep (1000);        fmq_server_destroy (&server);    }    else    if (streq (argv [1], "-c")) {        fmq_client_t *client = fmq_client_new ();        fmq_client_configure (client, "client_test.cfg");        fmq_client_setoption (client, "client/inbox", "./fmqroot/recv");        fmq_client_set_resync (client, true);        fmq_client_subscribe (client, "/photos");        fmq_client_subscribe (client, "/logs");        fmq_client_connect (client, "tcp://localhost:5670");        while (true) {            //  Get message from fmq_client API            zmsg_t *msg = fmq_client_recv (client);            if (!msg)                break;              //  Interrupted            char *command = zmsg_popstr (msg);            if (streq (command, "DELIVER")) {                char *filename = zmsg_popstr (msg);                char *fullname = zmsg_popstr (msg);                printf ("I: received %s (%s)/n", filename, fullname);                free (filename);                free (fullname);            }            free (command);            zmsg_destroy (&msg);        }        fmq_client_destroy (&client);    }    return 0;}

/////  Destroy a message object and all frames it containsvoid QmlZmsgAttached::destruct (QmlZmsg *qmlSelf) {    zmsg_destroy (&qmlSelf->self);};

static ints_self_handle_pipe (self_t *self){    //  Get the whole message off the pipe in one go    zmsg_t *request = zmsg_recv (self->pipe);    if (!request)        return -1;                  //  Interrupted    char *command = zmsg_popstr (request);    assert (command);    if (self->verbose)        zsys_info ("zproxy: API command=%s", command);    if (streq (command, "FRONTEND")) {        s_self_configure (self, &self->frontend, request, "frontend");        zsock_signal (self->pipe, 0);    }    else    if (streq (command, "BACKEND")) {        s_self_configure (self, &self->backend, request, "backend");        zsock_signal (self->pipe, 0);    }    else    if (streq (command, "CAPTURE")) {        self->capture = zsock_new (ZMQ_PUSH);        assert (self->capture);        char *endpoint = zmsg_popstr (request);        assert (endpoint);        int rc = zsock_connect (self->capture, "%s", endpoint);        assert (rc == 0);        zstr_free (&endpoint);        zsock_signal (self->pipe, 0);    }    else    if (streq (command, "PAUSE")) {        zpoller_destroy (&self->poller);        self->poller = zpoller_new (self->pipe, NULL);        assert (self->poller);        zsock_signal (self->pipe, 0);    }    else    if (streq (command, "RESUME")) {        zpoller_destroy (&self->poller);        self->poller = zpoller_new (self->pipe, self->frontend, self->backend, NULL);        assert (self->poller);        zsock_signal (self->pipe, 0);    }    else    if (streq (command, "VERBOSE")) {        self->verbose = true;        zsock_signal (self->pipe, 0);    }    else    if (streq (command, "$TERM"))        self->terminated = true;    else {        zsys_error ("zproxy: - invalid command: %s", command);        assert (false);    }    zstr_free (&command);    zmsg_destroy (&request);    return 0;}

//.........这里部分代码省略.........    if (streq (command, "JOIN")) {        char *name = zmsg_popstr (request);        if (!zlist_exists (self->own_groups, name)) {            void *item;            //  Only send if we're not already in group            zlist_append (self->own_groups, name);            zre_msg_t *msg = zre_msg_new (ZRE_MSG_JOIN);            zre_msg_set_group (msg, name);            //  Update status before sending command            zre_msg_set_status (msg, ++(self->status));            for (item = zhash_first (self->peers); item != NULL;                    item = zhash_next (self->peers))                zyre_node_send_peer (zhash_cursor (self->peers), item, msg);            zre_msg_destroy (&msg);            if (self->verbose)                zsys_info ("(%s) JOIN group=%s", self->name, name);        }        zstr_free (&name);    }    else    if (streq (command, "LEAVE")) {        char *name = zmsg_popstr (request);        if (zlist_exists (self->own_groups, name)) {            void *item;            //  Only send if we are actually in group            zre_msg_t *msg = zre_msg_new (ZRE_MSG_LEAVE);            zre_msg_set_group (msg, name);            //  Update status before sending command            zre_msg_set_status (msg, ++(self->status));            for (item = zhash_first (self->peers); item != NULL;                    item = zhash_next (self->peers))                zyre_node_send_peer (zhash_cursor (self->peers), item, msg);            zre_msg_destroy (&msg);            zlist_remove (self->own_groups, name);            if (self->verbose)                zsys_info ("(%s) LEAVE group=%s", self->name, name);        }        zstr_free (&name);    }    else    if (streq (command, "PEERS"))        zsock_send (self->pipe, "p", zhash_keys (self->peers));    else    if (streq (command, "GROUP PEERS")) {        char *name = zmsg_popstr (request);        zyre_group_t *group = (zyre_group_t *) zhash_lookup (self->peer_groups, name);        if (group)            zsock_send (self->pipe, "p", zyre_group_peers (group));        else            zsock_send (self->pipe, "p", NULL);        zstr_free (&name);    }    else    if (streq (command, "PEER ENDPOINT")) {        char *uuid = zmsg_popstr (request);        zyre_peer_t *peer = (zyre_peer_t *) zhash_lookup (self->peers, uuid);        assert (peer);        zsock_send (self->pipe, "s", zyre_peer_endpoint (peer));        zstr_free (&uuid);    }    else    if (streq (command, "PEER NAME")) {        char *uuid = zmsg_popstr (request);        zyre_peer_t *peer = (zyre_peer_t *) zhash_lookup (self->peers, uuid);        assert (peer);        zsock_send (self->pipe, "s", zyre_peer_name (peer));        zstr_free (&uuid);    }    else    if (streq (command, "PEER HEADER")) {        char *uuid = zmsg_popstr (request);        char *key = zmsg_popstr (request);        zyre_peer_t *peer = (zyre_peer_t *) zhash_lookup (self->peers, uuid);        if (!peer)            zstr_send (self->pipe, "");        else            zstr_send (self->pipe, zyre_peer_header (peer, key, NULL));        zstr_free (&uuid);        zstr_free (&key);    }    else    if (streq (command, "PEER GROUPS"))        zsock_send (self->pipe, "p", zhash_keys (self->peer_groups));    else    if (streq (command, "OWN GROUPS"))        zsock_send (self->pipe, "p", zlist_dup (self->own_groups));    else    if (streq (command, "DUMP"))        zyre_node_dump (self);    else    if (streq (command, "$TERM"))        self->terminated = true;    else {        zsys_error ("invalid command '%s'", command);        assert (false);    }    zstr_free (&command);    zmsg_destroy (&request);}

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 (zmsg_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 (zmsg_parts (msg) == 1            && strcmp (zmsg_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;}

zgossip_msg_t *zgossip_msg_decode (zmsg_t **msg_p){    assert (msg_p);    zmsg_t *msg = *msg_p;    if (msg == NULL)        return NULL;            zgossip_msg_t *self = zgossip_msg_new (0);    //  Read and parse command in frame    zframe_t *frame = zmsg_pop (msg);    if (!frame)         goto empty;             //  Malformed or empty    //  Get and check protocol signature    self->needle = zframe_data (frame);    self->ceiling = self->needle + zframe_size (frame);    uint16_t signature;    GET_NUMBER2 (signature);    if (signature != (0xAAA0 | 0))        goto empty;             //  Invalid signature    //  Get message id and parse per message type    GET_NUMBER1 (self->id);    switch (self->id) {        case ZGOSSIP_MSG_HELLO:            GET_NUMBER1 (self->version);            if (self->version != 1)                goto malformed;            break;        case ZGOSSIP_MSG_PUBLISH:            GET_NUMBER1 (self->version);            if (self->version != 1)                goto malformed;            GET_STRING (self->key);            GET_LONGSTR (self->value);            GET_NUMBER4 (self->ttl);            break;        case ZGOSSIP_MSG_PING:            GET_NUMBER1 (self->version);            if (self->version != 1)                goto malformed;            break;        case ZGOSSIP_MSG_PONG:            GET_NUMBER1 (self->version);            if (self->version != 1)                goto malformed;            break;        case ZGOSSIP_MSG_INVALID:            GET_NUMBER1 (self->version);            if (self->version != 1)                goto malformed;            break;        default:            goto malformed;    }    //  Successful return    zframe_destroy (&frame);    zmsg_destroy (msg_p);    return self;    //  Error returns    malformed:        zsys_error ("malformed message '%d'/n", self->id);    empty:        zframe_destroy (&frame);        zmsg_destroy (msg_p);        zgossip_msg_destroy (&self);        return (NULL);}

static void *client_task (void *args){    bool verbose = *((bool *) args);        char filename [256];    snprintf (filename, 255, TESTDIR "/client-%07d.cert", randof (10000000));    zcert_t *client_cert = zcert_new ();    zcert_save_public (client_cert, filename);    curve_client_t *client = curve_client_new (&client_cert);    curve_client_set_verbose (client, verbose);    zcert_t *server_cert = zcert_load (TESTDIR "/server.cert");    assert (server_cert);    curve_client_connect (client, "tcp://127.0.0.1:9006", zcert_public_key (server_cert));    zcert_destroy (&server_cert);    curve_client_sendstr (client, "Hello, World");    char *reply = curve_client_recvstr (client);    assert (streq (reply, "Hello, World"));    free (reply);    //  Try a multipart message    zmsg_t *msg = zmsg_new ();    zmsg_addstr (msg, "Hello, World");    zmsg_addstr (msg, "Second frame");    curve_client_send (client, &msg);    msg = curve_client_recv (client);    assert (zmsg_size (msg) == 2);    zmsg_destroy (&msg);    //  Now send messages of increasing size, check they work    int count;    int size = 0;    for (count = 0; count < 18; count++) {        zframe_t *data = zframe_new (NULL, size);        int byte_nbr;        //  Set data to sequence 0...255 repeated        for (byte_nbr = 0; byte_nbr < size; byte_nbr++)            zframe_data (data)[byte_nbr] = (byte) byte_nbr;        msg = zmsg_new ();        zmsg_prepend (msg, &data);        curve_client_send (client, &msg);        msg = curve_client_recv (client);        data = zmsg_pop (msg);        assert (data);        assert (zframe_size (data) == size);        for (byte_nbr = 0; byte_nbr < size; byte_nbr++) {            assert (zframe_data (data)[byte_nbr] == (byte) byte_nbr);        }        zframe_destroy (&data);        zmsg_destroy (&msg);        size = size * 2 + 1;    }    //  Signal end of test    curve_client_sendstr (client, "END");    reply = curve_client_recvstr (client);    free (reply);    curve_client_destroy (&client);    return NULL;}

zmsg_t *zgossip_msg_encode (zgossip_msg_t **self_p){    assert (self_p);    assert (*self_p);        zgossip_msg_t *self = *self_p;    zmsg_t *msg = zmsg_new ();    size_t frame_size = 2 + 1;          //  Signature and message ID    switch (self->id) {        case ZGOSSIP_MSG_HELLO:            //  version is a 1-byte integer            frame_size += 1;            break;                    case ZGOSSIP_MSG_PUBLISH:            //  version is a 1-byte integer            frame_size += 1;            //  key is a string with 1-byte length            frame_size++;       //  Size is one octet            if (self->key)                frame_size += strlen (self->key);            //  value is a string with 4-byte length            frame_size += 4;            if (self->value)                frame_size += strlen (self->value);            //  ttl is a 4-byte integer            frame_size += 4;            break;                    case ZGOSSIP_MSG_PING:            //  version is a 1-byte integer            frame_size += 1;            break;                    case ZGOSSIP_MSG_PONG:            //  version is a 1-byte integer            frame_size += 1;            break;                    case ZGOSSIP_MSG_INVALID:            //  version is a 1-byte integer            frame_size += 1;            break;                    default:            zsys_error ("bad message type '%d', not sent/n", self->id);            //  No recovery, this is a fatal application error            assert (false);    }    //  Now serialize message into the frame    zframe_t *frame = zframe_new (NULL, frame_size);    self->needle = zframe_data (frame);    PUT_NUMBER2 (0xAAA0 | 0);    PUT_NUMBER1 (self->id);    switch (self->id) {        case ZGOSSIP_MSG_HELLO:            PUT_NUMBER1 (1);            break;        case ZGOSSIP_MSG_PUBLISH:            PUT_NUMBER1 (1);            if (self->key) {                PUT_STRING (self->key);            }            else                PUT_NUMBER1 (0);    //  Empty string            if (self->value) {                PUT_LONGSTR (self->value);            }            else                PUT_NUMBER4 (0);    //  Empty string            PUT_NUMBER4 (self->ttl);            break;        case ZGOSSIP_MSG_PING:            PUT_NUMBER1 (1);            break;        case ZGOSSIP_MSG_PONG:            PUT_NUMBER1 (1);            break;        case ZGOSSIP_MSG_INVALID:            PUT_NUMBER1 (1);            break;    }    //  Now send the data frame    if (zmsg_append (msg, &frame)) {        zmsg_destroy (&msg);        zgossip_msg_destroy (self_p);        return NULL;    }    //  Destroy zgossip_msg object    zgossip_msg_destroy (self_p);    return msg;}

static bpm_client_err_e _bpm_get_data_block (bpm_client_t *self, char *service,        acq_trans_t *acq_trans){    assert (self);    assert (service);    assert (acq_trans);    assert (acq_trans->block.data);    bpm_client_err_e err = BPM_CLIENT_SUCCESS;    ACQ_OPCODE_TYPE operation = ACQ_OPCODE_GET_DATA_BLOCK;    /* Message is:     * frame 0: operation code     * frame 1: channel     * frame 2: block required          */    zmsg_t *request = zmsg_new ();    zmsg_addmem (request, &operation, sizeof (operation));    zmsg_addmem (request, &acq_trans->req.chan, sizeof (acq_trans->req.chan));    zmsg_addmem (request, &acq_trans->block.idx, sizeof (acq_trans->block.idx));    mdp_client_send (self->mdp_client, service, &request);    /* Receive report */    zmsg_t *report = mdp_client_recv (self->mdp_client, NULL, NULL);    ASSERT_TEST(report != NULL, "Report received is NULL", err_null_report);    assert (zmsg_size (report) == 3);    /* Message is:     * frame 0: error code     * frame 1: data size     * frame 2: data block              */    zframe_t *err_code = zmsg_pop (report);    ASSERT_TEST(err_code != NULL, "Could not receive error code", err_null_code);    zframe_t *data_size_frm = zmsg_pop (report);    ASSERT_TEST(data_size_frm != NULL, "Could not receive data size", err_null_data_size);    uint32_t data_size = *(uint32_t *) zframe_data(data_size_frm);    zframe_t *data = zmsg_pop (report);    ASSERT_TEST(data != NULL, "Could not receive data", err_null_data);    if ( *(ACQ_REPLY_TYPE *) zframe_data (err_code) != ACQ_OK) {        DBE_DEBUG (DBG_LIB_CLIENT | DBG_LVL_TRACE, "[libclient] bpm_get_data_block: "            "Data block was not acquired/n");        err = BPM_CLIENT_ERR_SERVER;        goto err_get_data_block;    }    /* Data size effectively returned */    uint32_t read_size = (acq_trans->block.data_size < data_size) ?        acq_trans->block.data_size : data_size;    memcpy (acq_trans->block.data, (uint32_t *) zframe_data (data), read_size);	/* Print some debug messages */    DBE_DEBUG (DBG_LIB_CLIENT | DBG_LVL_TRACE, "[libclient] bpm_get_data_block: "            "read_size: %u/n", read_size);    DBE_DEBUG (DBG_LIB_CLIENT | DBG_LVL_TRACE, "[libclient] bpm_get_data_block: "            "acq_trans->block.data: %p/n", acq_trans->block.data);    acq_trans->block.bytes_read = read_size;err_get_data_block:    zframe_destroy (&data);err_null_data:    zframe_destroy (&data_size_frm);err_null_data_size:    zframe_destroy (&err_code);err_null_code:    zmsg_destroy (&report);err_null_report:    return err;}

voidcurve_client_test (bool verbose){    printf (" * curve_client: ");    //  @selftest    //  Create temporary directory for test files    zsys_dir_create (TESTDIR);        //  We'll create two new certificates and save the client public     //  certificate on disk; in a real case we'd transfer this securely    //  from the client machine to the server machine.    zcert_t *server_cert = zcert_new ();    zcert_save (server_cert, TESTDIR "/server.cert");    //  We'll run the server as a background task, and the    //  client in this foreground thread.    zthread_new (server_task, &verbose);    zcert_t *client_cert = zcert_new ();    zcert_save_public (client_cert, TESTDIR "/client.cert");    curve_client_t *client = curve_client_new (&client_cert);    curve_client_set_metadata (client, "Client", "CURVEZMQ/curve_client");    curve_client_set_metadata (client, "Identity", "E475DA11");    curve_client_set_verbose (client, verbose);    curve_client_connect (client, "tcp://127.0.0.1:9005", (byte *)zcert_public_key (server_cert));    curve_client_sendstr (client, "Hello, World");    char *reply = curve_client_recvstr (client);    assert (streq (reply, "Hello, World"));    free (reply);    //  Try a multipart message    zmsg_t *msg = zmsg_new ();    zmsg_addstr (msg, "Hello, World");    zmsg_addstr (msg, "Second frame");    curve_client_send (client, &msg);    msg = curve_client_recv (client);    assert (zmsg_size (msg) == 2);    zmsg_destroy (&msg);    //  Now send messages of increasing size, check they work    int count;    int size = 0;    for (count = 0; count < 18; count++) {        if (verbose)            printf ("Testing message of size=%d.../n", size);        zframe_t *data = zframe_new (NULL, size);        int byte_nbr;        //  Set data to sequence 0...255 repeated        for (byte_nbr = 0; byte_nbr < size; byte_nbr++)            zframe_data (data)[byte_nbr] = (byte) byte_nbr;        msg = zmsg_new ();        zmsg_prepend (msg, &data);        curve_client_send (client, &msg);        msg = curve_client_recv (client);        data = zmsg_pop (msg);        assert (data);        assert (zframe_size (data) == size);        for (byte_nbr = 0; byte_nbr < size; byte_nbr++) {            assert (zframe_data (data)[byte_nbr] == (byte) byte_nbr);        }        zframe_destroy (&data);        zmsg_destroy (&msg);        size = size * 2 + 1;    }    //  Signal end of test    curve_client_sendstr (client, "END");    reply = curve_client_recvstr (client);    free (reply);    zcert_destroy (&server_cert);    zcert_destroy (&client_cert);    curve_client_destroy (&client);        //  Delete all test files    zdir_t *dir = zdir_new (TESTDIR, NULL);    zdir_remove (dir, true);    zdir_destroy (&dir);    //  @end    //  Ensure server thread has exited before we do    zclock_sleep (100);    printf ("OK/n");}

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

zmsg_t *mdcli_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 1: "MDPCxy" (six bytes, MDP/Client x.y)    //  Frame 2: Service name (printable string)    zmsg_push (request, service);    zmsg_push (request, MDPC_CLIENT);    if (self->verbose) {        s_console ("I: send request to '%s' service:", service);        zmsg_dump (request);    }    int retries_left = self->retries;    while (retries_left && !s_interrupted) {        zmsg_t *msg = zmsg_dup (request);        zmsg_send (&msg, self->client);        while (!s_interrupted) {            //  Poll socket for a reply, with timeout            zmq_pollitem_t items [] = {                 { self->client, 0, ZMQ_POLLIN, 0 } };            zmq_poll (items, 1, self->timeout * 1000);            //  If we got a reply, process it            if (items [0].revents & ZMQ_POLLIN) {                zmsg_t *msg = zmsg_recv (self->client);                if (self->verbose) {                    s_console ("I: received reply:");                    zmsg_dump (msg);                }                //  Don't try to handle errors, just assert noisily                assert (zmsg_parts (msg) >= 3);                char *header = zmsg_pop (msg);                assert (streq (header, MDPC_CLIENT));                free (header);                char *reply_service = zmsg_pop (msg);                assert (streq (reply_service, service));                free (reply_service);                zmsg_destroy (&request);                return msg;     //  Success            }            else            if (--retries_left) {                if (self->verbose)                    s_console ("W: no reply, reconnecting...");                //  Reconnect, and resend message                s_mdcli_connect_to_broker (self);                zmsg_t *msg = zmsg_dup (request);                zmsg_send (&msg, self->client);            }            else {                if (self->verbose)                    s_console ("W: permanent error, abandoning request");                break;          //  Give up            }        }    }    if (s_interrupted)        printf ("W: interrupt received, killing client.../n");    zmsg_destroy (&request);    return NULL;}

static voidnode_task (void *args, zctx_t *ctx, void *pipe){    zyre_t *node = zyre_new (ctx);    if (!node)        return;                 //  Could not create new node    zyre_set_verbose (node);    zyre_start (node);    int64_t counter = 0;    char *to_peer = NULL;        //  Either of these set,    char *to_group = NULL;       //    and we set a message    char *cookie = NULL;    zpoller_t *poller = zpoller_new (pipe, zyre_socket (node), NULL);    int64_t trigger = zclock_time () + 1000;    while (!zctx_interrupted) {        void *which = zpoller_wait (poller, randof (1000));        //  Any command from parent means EXIT        if (which == pipe)            break;        //  Process an event from node        if (which == zyre_socket (node)) {            zmsg_t *incoming = zyre_recv (node);            if (!incoming)                break;              //  Interrupted            char *event = zmsg_popstr (incoming);            if (streq (event, "ENTER")) {                //  Always say hello to new peer                to_peer = zmsg_popstr (incoming);            }            else            if (streq (event, "EXIT")) {                //  Always try talk to departed peer                to_peer = zmsg_popstr (incoming);            }            else            if (streq (event, "WHISPER")) {                //  Send back response 1/2 the time                if (randof (2) == 0) {                    to_peer = zmsg_popstr (incoming);                    cookie = zmsg_popstr (incoming);                }            }            else            if (streq (event, "SHOUT")) {                to_peer = zmsg_popstr (incoming);                to_group = zmsg_popstr (incoming);                cookie = zmsg_popstr (incoming);                //  Send peer response 1/3rd the time                if (randof (3) > 0) {                    free (to_peer);                    to_peer = NULL;                }                //  Send group response 1/3rd the time                if (randof (3) > 0) {                    free (to_group);                    to_group = NULL;                }            }            else            if (streq (event, "JOIN")) {                char *from_peer = zmsg_popstr (incoming);                char *group = zmsg_popstr (incoming);                printf ("I: %s joined %s/n", from_peer, group);                free (from_peer);                free (group);            }            else            if (streq (event, "LEAVE")) {                char *from_peer = zmsg_popstr (incoming);                char *group = zmsg_popstr (incoming);                printf ("I: %s left %s/n", from_peer, group);                free (from_peer);                free (group);            }            else            if (streq (event, "DELIVER")) {                char *filename = zmsg_popstr (incoming);                char *fullname = zmsg_popstr (incoming);                printf ("I: received file %s/n", fullname);                free (fullname);                free (filename);            }            free (event);            zmsg_destroy (&incoming);            //  Send outgoing messages if needed            if (to_peer) {                zyre_whispers (node, to_peer, "%lu", counter++);                free (to_peer);                to_peer = NULL;            }            if (to_group) {                zyre_shouts (node, to_group, "%lu", counter++);                free (to_group);                to_group = NULL;//.........这里部分代码省略.........

    std::vector<Service> find(std::map<std::string, std::string> const &txt,			      std::string const &name)    {	directoryd::ServiceRequest request;	request.set_type(directoryd::FIND);	directoryd::ServiceRequest::Find *f = request.mutable_find();	f->set_type("_hotdec._tcp");	if (name.empty() == false) {	    f->set_name(name);	}	for (auto &t : txt) {	    directoryd::TxtField *txtfield = f->add_txt();	    txtfield->set_key(t.first);	    txtfield->set_value(t.second);	}	string buffer;	if (debug && TextFormat::PrintToString(request, &buffer)) {	    fprintf(stderr, "request: %s/n", buffer.c_str());	}	zframe_t *sf = zframe_new(NULL, request.ByteSize());	assert (sf != NULL);	request.SerializeToArray(zframe_data(sf),zframe_size(sf));	int retval = zframe_send(&sf, DDClient::instance().query_socket(), 0);	assert(retval == 0);	zmsg_t *repmsg = zmsg_recv(DDClient::instance().query_socket());	if (debug)	    zmsg_fprint(repmsg, stderr);	zframe_t *rf = zmsg_pop(repmsg);	directoryd::ServiceReply reply;	reply.ParseFromArray(zframe_data(rf),zframe_size(rf));	if (debug && TextFormat::PrintToString(reply, &buffer)) {	    fprintf(stderr, "reply: %s/n", buffer.c_str());	}	zframe_destroy(&rf);	zmsg_destroy(&repmsg);	std::vector<Service> services;	if (reply.type() != directoryd::FIND) {	    throw QueryError("Got back incorrect message type when trying to query.");	}	if (reply.success() != true) {	    throw QueryError(reply.result());	}	for (int i = 0; i < reply.findresult_size(); ++i) {	    Service s;	    auto location = reply.findresult(i).location();	    s.hostname = location.host();	    s.port = location.port();	    for (int j = 0; j < reply.findresult(i).txt_size(); ++j) {		auto t = reply.findresult(i).txt(j);		s.txt[t.key()] = t.value();	    }	    services.push_back(s);	}	return services;    }

//.........这里部分代码省略.........    PUT_NUMBER2 (0xAAA0 | 1);    PUT_NUMBER1 (self->id);    switch (self->id) {        case ZRE_MSG_HELLO:            PUT_NUMBER2 (self->sequence);            if (self->ipaddress) {                PUT_STRING (self->ipaddress);            }            else                PUT_NUMBER1 (0);    //  Empty string            PUT_NUMBER2 (self->mailbox);            if (self->groups) {                PUT_NUMBER4 (zlist_size (self->groups));                char *groups = (char *) zlist_first (self->groups);                while (groups) {                    PUT_LONGSTR (groups);                    groups = (char *) zlist_next (self->groups);                }            }            else                PUT_NUMBER4 (0);    //  Empty string array            PUT_NUMBER1 (self->status);            if (self->headers) {                PUT_NUMBER4 (zhash_size (self->headers));                zhash_foreach (self->headers, s_headers_write, self);            }            else                PUT_NUMBER4 (0);    //  Empty dictionary            break;        case ZRE_MSG_WHISPER:            PUT_NUMBER2 (self->sequence);            break;        case ZRE_MSG_SHOUT:            PUT_NUMBER2 (self->sequence);            if (self->group) {                PUT_STRING (self->group);            }            else                PUT_NUMBER1 (0);    //  Empty string            break;        case ZRE_MSG_JOIN:            PUT_NUMBER2 (self->sequence);            if (self->group) {                PUT_STRING (self->group);            }            else                PUT_NUMBER1 (0);    //  Empty string            PUT_NUMBER1 (self->status);            break;        case ZRE_MSG_LEAVE:            PUT_NUMBER2 (self->sequence);            if (self->group) {                PUT_STRING (self->group);            }            else                PUT_NUMBER1 (0);    //  Empty string            PUT_NUMBER1 (self->status);            break;        case ZRE_MSG_PING:            PUT_NUMBER2 (self->sequence);            break;        case ZRE_MSG_PING_OK:            PUT_NUMBER2 (self->sequence);            break;    }    //  Now send the data frame    if (zmsg_append (msg, &frame)) {        zmsg_destroy (&msg);        zre_msg_destroy (&self);        return NULL;    }    //  Now send the content field if set    if (self->id == ZRE_MSG_WHISPER) {        zframe_t *content_part = zmsg_pop (self->content);        while (content_part) {            zmsg_append (msg, &content_part);            content_part = zmsg_pop (self->content);        }    }    //  Now send the content field if set    if (self->id == ZRE_MSG_SHOUT) {        zframe_t *content_part = zmsg_pop (self->content);        while (content_part) {            zmsg_append (msg, &content_part);            content_part = zmsg_pop (self->content);        }    }    //  Destroy zre_msg object    zre_msg_destroy (&self);    return msg;}