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

void web_request(void *sweb, req_store_t * req_store, void *spss, void *sgraph){	zmsg_t *msg = zmsg_recv(sweb);	zframe_t *address = zmsg_unwrap(msg);	json_t *req_json;	json_error_t error;	printf("/nbroker:sweb 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));	req_json = json_loadb(data, data_size, 0, &error);	zmsg_destroy(&msg);	int32_t requestId = request_store_add(req_store, address, req_json);	json_t *clientRequest = json_object_get(req_json, "clientRequest");	json_t *request = json_object_get(clientRequest, "request");	const char *type = json_string_value(json_object_get(request, "type"));	if (strcmp(type, "searchRequest") == 0)		web_request_searchRequest(requestId, request, spss);	else if (strcmp(type, "newNode") == 0)		web_request_newNode(requestId, request, sgraph);	else if (strcmp(type, "newPosition") == 0)		web_request_newPosition(requestId, request, spss);	else if (strcmp(type, "newLink") == 0)		web_request_newLink(requestId, request, sgraph);	else if (strcmp(type, "delLink") == 0)		web_request_delLink(requestId, request, sgraph);	else if (strcmp(type, "delNode") == 0)		web_request_delNode(requestId, request, sgraph);	else if (strcmp(type, "newNodeData") == 0)		web_request_newNodeData(requestId, request, sgraph);	else if (strcmp(type, "newLinkData") == 0)		web_request_newLinkData(requestId, request, sgraph);	else {		//TODO process request		//malformed request		printf("/ni received a malformed request : %s", type);		//delete request 		zframe_destroy(&address);		request_store_delete(req_store, requestId);	}}

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

voidzhash_test (bool verbose){    printf (" * zhash: ");    //  @selftest    zhash_t *hash = zhash_new ();    assert (hash);    assert (zhash_size (hash) == 0);    assert (zhash_first (hash) == NULL);    assert (zhash_cursor (hash) == NULL);    //  Insert some items    int rc;    rc = zhash_insert (hash, "DEADBEEF", "dead beef");    char *item = (char *) zhash_first (hash);    assert (streq (zhash_cursor (hash), "DEADBEEF"));    assert (streq (item, "dead beef"));    assert (rc == 0);    rc = zhash_insert (hash, "ABADCAFE", "a bad cafe");    assert (rc == 0);    rc = zhash_insert (hash, "C0DEDBAD", "coded bad");    assert (rc == 0);    rc = zhash_insert (hash, "DEADF00D", "dead food");    assert (rc == 0);    assert (zhash_size (hash) == 4);    //  Look for existing items    item = (char *) zhash_lookup (hash, "DEADBEEF");    assert (streq (item, "dead beef"));    item = (char *) zhash_lookup (hash, "ABADCAFE");    assert (streq (item, "a bad cafe"));    item = (char *) zhash_lookup (hash, "C0DEDBAD");    assert (streq (item, "coded bad"));    item = (char *) zhash_lookup (hash, "DEADF00D");    assert (streq (item, "dead food"));    //  Look for non-existent items    item = (char *) zhash_lookup (hash, "foo");    assert (item == NULL);    //  Try to insert duplicate items    rc = zhash_insert (hash, "DEADBEEF", "foo");    assert (rc == -1);    item = (char *) zhash_lookup (hash, "DEADBEEF");    assert (streq (item, "dead beef"));    //  Some rename tests    //  Valid rename, key is now LIVEBEEF    rc = zhash_rename (hash, "DEADBEEF", "LIVEBEEF");    assert (rc == 0);    item = (char *) zhash_lookup (hash, "LIVEBEEF");    assert (streq (item, "dead beef"));    //  Trying to rename an unknown item to a non-existent key    rc = zhash_rename (hash, "WHATBEEF", "NONESUCH");    assert (rc == -1);    //  Trying to rename an unknown item to an existing key    rc = zhash_rename (hash, "WHATBEEF", "LIVEBEEF");    assert (rc == -1);    item = (char *) zhash_lookup (hash, "LIVEBEEF");    assert (streq (item, "dead beef"));    //  Trying to rename an existing item to another existing item    rc = zhash_rename (hash, "LIVEBEEF", "ABADCAFE");    assert (rc == -1);    item = (char *) zhash_lookup (hash, "LIVEBEEF");    assert (streq (item, "dead beef"));    item = (char *) zhash_lookup (hash, "ABADCAFE");    assert (streq (item, "a bad cafe"));    //  Test keys method    zlist_t *keys = zhash_keys (hash);    assert (zlist_size (keys) == 4);    zlist_destroy (&keys);    //  Test dup method    zhash_t *copy = zhash_dup (hash);    assert (zhash_size (copy) == 4);    item = (char *) zhash_lookup (copy, "LIVEBEEF");    assert (item);    assert (streq (item, "dead beef"));    zhash_destroy (&copy);    //  Test pack/unpack methods    zframe_t *frame = zhash_pack (hash);    copy = zhash_unpack (frame);    zframe_destroy (&frame);    assert (zhash_size (copy) == 4);    item = (char *) zhash_lookup (copy, "LIVEBEEF");    assert (item);    assert (streq (item, "dead beef"));    zhash_destroy (&copy);    //  Test save and load    zhash_comment (hash, "This is a test file");    zhash_comment (hash, "Created by %s", "czmq_selftest");    zhash_save (hash, ".cache");//.........这里部分代码省略.........

voidzsocket_test (bool verbose){    printf (" * zsocket (deprecated): ");    //  @selftest    zctx_t *ctx = zctx_new ();    assert (ctx);    //  Create a detached thread, let it run    char *interf = "127.0.0.1";    char *domain = "localhost";    int service = 5560;    void *writer = zsocket_new (ctx, ZMQ_PUSH);    assert (writer);    void *reader = zsocket_new (ctx, ZMQ_PULL);    assert (reader);    assert (streq (zsocket_type_str (writer), "PUSH"));    assert (streq (zsocket_type_str (reader), "PULL"));    int rc = zsocket_bind (writer, "tcp://%s:%d", interf, service);    assert (rc == service);#if (ZMQ_VERSION >= ZMQ_MAKE_VERSION (3, 2, 0))    //  Check unbind    rc = zsocket_unbind (writer, "tcp://%s:%d", interf, service);    assert (rc == 0);    //  In some cases and especially when running under Valgrind, doing    //  a bind immediately after an unbind causes an EADDRINUSE error.    //  Even a short sleep allows the OS to release the port for reuse.    zclock_sleep (100);    //  Bind again    rc = zsocket_bind (writer, "tcp://%s:%d", interf, service);    assert (rc == service);#endif    rc = zsocket_connect (reader, "tcp://%s:%d", domain, service);    assert (rc == 0);    zstr_send (writer, "HELLO");    char *message = zstr_recv (reader);    assert (message);    assert (streq (message, "HELLO"));    free (message);    //  Test binding to ports    int port = zsocket_bind (writer, "tcp://%s:*", interf);    assert (port >= ZSOCKET_DYNFROM && port <= ZSOCKET_DYNTO);    assert (zsocket_poll (writer, 100) == false);    //  Test error state when connecting to an invalid socket type    //  ('txp://' instead of 'tcp://', typo intentional)    rc = zsocket_connect (reader, "txp://%s:%d", domain, service);    assert (rc == -1);    //  Test sending frames to socket    rc = zsocket_sendmem (writer, "ABC", 3, ZFRAME_MORE);    assert (rc == 0);    rc = zsocket_sendmem (writer, "DEFG", 4, 0);    assert (rc == 0);    zframe_t *frame = zframe_recv (reader);    assert (frame);    assert (zframe_streq (frame, "ABC"));    assert (zframe_more (frame));    zframe_destroy (&frame);    frame = zframe_recv (reader);    assert (frame);    assert (zframe_streq (frame, "DEFG"));    assert (!zframe_more (frame));    zframe_destroy (&frame);    rc = zsocket_signal (writer);    assert (rc == 0);    rc = zsocket_wait (reader);    assert (rc == 0);    zsocket_destroy (ctx, reader);    zsocket_destroy (ctx, writer);    zctx_destroy (&ctx);    //  @end    printf ("OK/n");}

/////  Destroy a framevoid QmlZframeAttached::destruct (QmlZframe *qmlSelf) {    zframe_destroy (&qmlSelf->self);};

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 voids_self_handle_sink (self_t *self){#if defined (ZMQ_EVENT_ALL)#if (ZMQ_VERSION_MAJOR == 4)    //  First frame is event number and value    zframe_t *frame = zframe_recv (self->sink);    int event = *(uint16_t *) (zframe_data (frame));    int value = *(uint32_t *) (zframe_data (frame) + 2);    //  Address is in second message frame    char *address = zstr_recv (self->sink);    zframe_destroy (&frame);#elif (ZMQ_VERSION_MAJOR == 3 && ZMQ_VERSION_MINOR == 2)    //  zmq_event_t is passed as-is in the frame    zframe_t *frame = zframe_recv (self->sink);    zmq_event_t *eptr = (zmq_event_t *) zframe_data (frame);    int event = eptr->event;    int value = eptr->data.listening.fd;    char *address = strdup (eptr->data.listening.addr);    zframe_destroy (&frame);#else    //  We can't plausibly be here with other versions of libzmq    assert (false);#endif    //  Now map event to text equivalent    char *name;    switch (event) {        case ZMQ_EVENT_ACCEPTED:            name = "ACCEPTED";            break;        case ZMQ_EVENT_ACCEPT_FAILED:            name = "ACCEPT_FAILED";            break;        case ZMQ_EVENT_BIND_FAILED:            name = "BIND_FAILED";            break;        case ZMQ_EVENT_CLOSED:            name = "CLOSED";            break;        case ZMQ_EVENT_CLOSE_FAILED:            name = "CLOSE_FAILED";            break;        case ZMQ_EVENT_DISCONNECTED:            name = "DISCONNECTED";            break;        case ZMQ_EVENT_CONNECTED:            name = "CONNECTED";            break;        case ZMQ_EVENT_CONNECT_DELAYED:            name = "CONNECT_DELAYED";            break;        case ZMQ_EVENT_CONNECT_RETRIED:            name = "CONNECT_RETRIED";            break;        case ZMQ_EVENT_LISTENING:            name = "LISTENING";            break;#if (ZMQ_VERSION_MAJOR == 4)        case ZMQ_EVENT_MONITOR_STOPPED:            name = "MONITOR_STOPPED";            break;#endif        default:            zsys_error ("illegal socket monitor event: %d", event);            name = "UNKNOWN";            break;    }    if (self->verbose)        zsys_info ("zmonitor: %s - %s", name, address);    zstr_sendfm (self->pipe, "%s", name);    zstr_sendfm (self->pipe, "%d", value);    zstr_send (self->pipe, address);    free (address);#endif}

intmdp_client_msg_recv (mdp_client_msg_t *self, zsock_t *input){    assert (input);        if (zsock_type (input) == ZMQ_ROUTER) {        zframe_destroy (&self->routing_id);        self->routing_id = zframe_recv (input);        if (!self->routing_id || !zsock_rcvmore (input)) {            zsys_warning ("mdp_client_msg: no routing ID");            return -1;          //  Interrupted or malformed        }    }    zmq_msg_t frame;    zmq_msg_init (&frame);    int size = zmq_msg_recv (&frame, zsock_resolve (input), 0);    if (size == -1) {        zsys_warning ("mdp_client_msg: interrupted");        goto malformed;         //  Interrupted    }    //  Get and check protocol signature    self->needle = (byte *) zmq_msg_data (&frame);    self->ceiling = self->needle + zmq_msg_size (&frame);        uint16_t signature;    GET_NUMBER2 (signature);    if (signature != (0xAAA0 | 4)) {        zsys_warning ("mdp_client_msg: invalid signature");        //  TODO: discard invalid messages and loop, and return        //  -1 only on interrupt        goto malformed;         //  Interrupted    }    //  Get message id and parse per message type    GET_NUMBER1 (self->id);    switch (self->id) {        case MDP_CLIENT_MSG_CLIENT_REQUEST:            {                char version [256];                GET_STRING (version);                if (strneq (version, "MDPC02")) {                    zsys_warning ("mdp_client_msg: version is invalid");                    goto malformed;                }            }            {                byte messageid;                GET_NUMBER1 (messageid);                if (messageid != 1) {                    zsys_warning ("mdp_client_msg: messageid is invalid");                    goto malformed;                }            }            GET_STRING (self->service);            //  Get zero or more remaining frames            zmsg_destroy (&self->body);            if (zsock_rcvmore (input))                self->body = zmsg_recv (input);            else                self->body = zmsg_new ();            break;        case MDP_CLIENT_MSG_CLIENT_PARTIAL:            {                char version [256];                GET_STRING (version);                if (strneq (version, "MDPC02")) {                    zsys_warning ("mdp_client_msg: version is invalid");                    goto malformed;                }            }            {                byte messageid;                GET_NUMBER1 (messageid);                if (messageid != 2) {                    zsys_warning ("mdp_client_msg: messageid is invalid");                    goto malformed;                }            }            GET_STRING (self->service);            //  Get zero or more remaining frames            zmsg_destroy (&self->body);            if (zsock_rcvmore (input))                self->body = zmsg_recv (input);            else                self->body = zmsg_new ();            break;        case MDP_CLIENT_MSG_CLIENT_FINAL:            {                char version [256];                GET_STRING (version);                if (strneq (version, "MDPC02")) {                    zsys_warning ("mdp_client_msg: version is invalid");                    goto malformed;                }            }            {                byte messageid;                GET_NUMBER1 (messageid);//.........这里部分代码省略.........

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, "ipc://frontend.ipc");    zsocket_bind (backend, "ipc://backend.ipc");    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 ();    //  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 (true) {        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            zframe_t *identity = zmsg_unwrap (msg);            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, 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;}

//.........这里部分代码省略.........    assert (zlist_size (peer_groups) == 2);    zlist_destroy (&peer_groups);    char *value = zyre_peer_header_value (node2, zyre_uuid (node1), "X-HELLO");    assert (streq (value, "World"));    zstr_free (&value);    //  One node shouts to GLOBAL    zyre_shouts (node1, "GLOBAL", "Hello, World");    //  Second node should receive ENTER, JOIN, and SHOUT    zmsg_t *msg = zyre_recv (node2);    assert (msg);    char *command = zmsg_popstr (msg);    assert (streq (command, "ENTER"));    zstr_free (&command);    assert (zmsg_size (msg) == 4);    char *peerid = zmsg_popstr (msg);    char *name = zmsg_popstr (msg);    assert (streq (name, "node1"));    zstr_free (&name);    zframe_t *headers_packed = zmsg_pop (msg);    char *address = zmsg_popstr (msg);    char *endpoint = zyre_peer_address (node2, peerid);    assert (streq (address, endpoint));    zstr_free (&peerid);    zstr_free (&endpoint);    zstr_free (&address);    assert (headers_packed);    zhash_t *headers = zhash_unpack (headers_packed);    assert (headers);    zframe_destroy (&headers_packed);    assert (streq ((char *) zhash_lookup (headers, "X-HELLO"), "World"));    zhash_destroy (&headers);    zmsg_destroy (&msg);    msg = zyre_recv (node2);    assert (msg);    command = zmsg_popstr (msg);    assert (streq (command, "JOIN"));    zstr_free (&command);    assert (zmsg_size (msg) == 3);    zmsg_destroy (&msg);    msg = zyre_recv (node2);    assert (msg);    command = zmsg_popstr (msg);    assert (streq (command, "JOIN"));    zstr_free (&command);    assert (zmsg_size (msg) == 3);    zmsg_destroy (&msg);    msg = zyre_recv (node2);    assert (msg);    command = zmsg_popstr (msg);    assert (streq (command, "SHOUT"));    zstr_free (&command);    zmsg_destroy (&msg);    zyre_stop (node2);    msg = zyre_recv (node2);    assert (msg);    command = zmsg_popstr (msg);

static voids_socket_event (agent_t *self){    zframe_t *frame;    zmq_event_t event;    char *description = "Unknown";    char address [1025];    //  Copy event data into event struct    frame = zframe_recv (self->socket);    //  Extract id of the event as bitfield    memcpy (&(event.event), zframe_data (frame), sizeof (event.event));    //  Extract value which is either error code, fd, or reconnect interval    memcpy (&(event.value), zframe_data (frame) + sizeof (event.event),           sizeof (event.value));    zframe_destroy (&frame);    //  Copy address part    frame = zframe_recv (self->socket);    memcpy (address, zframe_data (frame), zframe_size (frame));    address [zframe_size (frame)] = 0;  // Terminate address string    zframe_destroy (&frame);    switch (event.event) {        case ZMQ_EVENT_ACCEPTED:            description = "Accepted";            break;        case ZMQ_EVENT_ACCEPT_FAILED:            description = "Accept failed";            break;        case ZMQ_EVENT_BIND_FAILED:            description = "Bind failed";            break;        case ZMQ_EVENT_CLOSED:            description = "Closed";            break;        case ZMQ_EVENT_CLOSE_FAILED:            description = "Close failed";            break;        case ZMQ_EVENT_DISCONNECTED:            description = "Disconnected";            break;        case ZMQ_EVENT_CONNECTED:            description = "Connected";            break;        case ZMQ_EVENT_CONNECT_DELAYED:            description = "Connect delayed";            break;        case ZMQ_EVENT_CONNECT_RETRIED:            description = "Connect retried";            break;        case ZMQ_EVENT_LISTENING:            description = "Listening";            break;        case ZMQ_EVENT_MONITOR_STOPPED:            description = "Monitor stopped";            break;        default:            if (self->verbose)                printf ("Unknown socket monitor event: %d", event.event);            break;    }    if (self->verbose)        printf ("I: zmonitor: %s - %s/n", description, address);    zmsg_t *msg = zmsg_new();    zmsg_addstrf (msg, "%d", (int) event.event);    zmsg_addstrf (msg, "%d", (int) event.value);    zmsg_addstrf (msg, "%s", address);    zmsg_addstrf (msg, "%s", description);    zmsg_send (&msg, self->pipe);}

voidzbeacon_test (bool verbose){    printf (" * zbeacon: ");    if (verbose)        printf ("/n");    //  @selftest    //  Test 1 - two beacons, one speaking, one listening    //  Create speaker beacon to broadcast our service    zactor_t *speaker = zactor_new (zbeacon, NULL);    assert (speaker);    if (verbose)        zstr_sendx (speaker, "VERBOSE", NULL);    zsock_send (speaker, "si", "CONFIGURE", 9999);    char *hostname = zstr_recv (speaker);    if (!*hostname) {        printf ("OK (skipping test, no UDP broadcasting)/n");        zactor_destroy (&speaker);        free (hostname);        return;    }    free (hostname);    //  Create listener beacon on port 9999 to lookup service    zactor_t *listener = zactor_new (zbeacon, NULL);    assert (listener);    if (verbose)        zstr_sendx (listener, "VERBOSE", NULL);    zsock_send (listener, "si", "CONFIGURE", 9999);    hostname = zstr_recv (listener);    assert (*hostname);    free (hostname);    //  We will broadcast the magic value 0xCAFE    byte announcement [2] = { 0xCA, 0xFE };    zsock_send (speaker, "sbi", "PUBLISH", announcement, 2, 100);    //  We will listen to anything (empty subscription)    zsock_send (listener, "sb", "SUBSCRIBE", "", 0);    //  Wait for at most 1/2 second if there's no broadcasting    zsock_set_rcvtimeo (listener, 500);    char *ipaddress = zstr_recv (listener);    if (ipaddress) {        zframe_t *content = zframe_recv (listener);        assert (zframe_size (content) == 2);        assert (zframe_data (content) [0] == 0xCA);        assert (zframe_data (content) [1] == 0xFE);        zframe_destroy (&content);        zstr_free (&ipaddress);        zstr_sendx (speaker, "SILENCE", NULL);    }    zactor_destroy (&listener);    zactor_destroy (&speaker);    //  Test subscription filter using a 3-node setup    zactor_t *node1 = zactor_new (zbeacon, NULL);    assert (node1);    zsock_send (node1, "si", "CONFIGURE", 5670);    hostname = zstr_recv (node1);    assert (*hostname);    free (hostname);    zactor_t *node2 = zactor_new (zbeacon, NULL);    assert (node2);    zsock_send (node2, "si", "CONFIGURE", 5670);    hostname = zstr_recv (node2);    assert (*hostname);    free (hostname);    zactor_t *node3 = zactor_new (zbeacon, NULL);    assert (node3);    zsock_send (node3, "si", "CONFIGURE", 5670);    hostname = zstr_recv (node3);    assert (*hostname);    free (hostname);    zsock_send (node1, "sbi", "PUBLISH", "NODE/1", 6, 250);    zsock_send (node2, "sbi", "PUBLISH", "NODE/2", 6, 250);    zsock_send (node3, "sbi", "PUBLISH", "RANDOM", 6, 250);    zsock_send (node1, "sb", "SUBSCRIBE", "NODE", 4);    //  Poll on three API sockets at once    zpoller_t *poller = zpoller_new (node1, node2, node3, NULL);    assert (poller);    int64_t stop_at = zclock_mono () + 1000;    while (zclock_mono () < stop_at) {        long timeout = (long) (stop_at - zclock_mono ());        if (timeout < 0)            timeout = 0;        void *which = zpoller_wait (poller, timeout * ZMQ_POLL_MSEC);        if (which) {            assert (which == node1);            char *ipaddress, *received;            zstr_recvx (node1, &ipaddress, &received, NULL);            assert (streq (received, "NODE/2"));            zstr_free (&ipaddress);            zstr_free (&received);        }//.........这里部分代码省略.........

//.........这里部分代码省略.........			} else if (streq (event, "EXIT")) {				peer = zmsg_popstr (incoming);				debugLog ("I: EXIT '%s'", peer);				if(self->callback) {					(*self->callback)(self,							LSD_EVENT_EXIT,							peer,							NULL,							NULL,							0,							self->class_ptr);				}			} else if (streq (event, "WHISPER")) {				peer = zmsg_popstr (incoming);				msg_frame = zmsg_pop (incoming);				debugLog ("I: WHISPER '%s' msglen %d", peer, (int)zframe_size(msg_frame));				if(self->callback) {					(*self->callback)(self,							LSD_EVENT_WHISPER,							peer,							NULL,							(const uint8_t*)zframe_data(msg_frame),							zframe_size(msg_frame),							self->class_ptr);				}			} else 	if (streq (event, "SHOUT")) {				peer = zmsg_popstr (incoming);				group = zmsg_popstr (incoming);				msg_frame = zmsg_pop (incoming); 				debugLog ("I: SHOUT from '%s' group '%s' msglen %d", peer, group, (int)zframe_size(msg_frame));				if(self->callback) {					(*self->callback)(self,							LSD_EVENT_SHOUT,							peer,							group,							zframe_data(msg_frame),							zframe_size(msg_frame),							self->class_ptr);				}			} else if (streq (event, "DELIVER")) {				char *filename = zmsg_popstr (incoming);				char *fullname = zmsg_popstr (incoming);				debugLog ("I: DELIVER file %s", fullname);				if(self->callback) {					(*self->callback)(self,							LSD_EVENT_DELIVER,							NULL,							NULL,							(const uint8_t*)fullname,							strlen(fullname),							self->class_ptr);				}				free (fullname);				free (filename);			}else if (streq (event, "JOIN")) {				peer = zmsg_popstr (incoming);				group = zmsg_popstr (incoming);				debugLog ("I: JOIN '%s - %s'", peer, group);				if(self->callback) {					(*self->callback)(self,							LSD_EVENT_JOIN,							peer,							group,							NULL,							0,							self->class_ptr);				}			} else if (streq (event, "LEAVE")) {				peer = zmsg_popstr (incoming);				group = zmsg_popstr (incoming);				debugLog ("I: LEAVE '%s - %s'", peer, group);				if(self->callback) {					(*self->callback)(self,							LSD_EVENT_LEAVE,							peer,							group,							NULL,							0,							self->class_ptr);				}			}			if(peer) {				free(peer);				peer = NULL;			}			if(group) {				free(group);				group = NULL;			}			if(msg_frame) {				zframe_destroy(&msg_frame);				msg_frame = NULL;			}						free (event);			zmsg_destroy (&incoming);		}	}}

void graph_response(void *sgraph, req_store_t * req_store, void *sweb,		    void *spss){	zmsg_t *msg = zmsg_recv(sgraph);	zframe_t *null = zmsg_unwrap(msg);	zframe_destroy(&null);	json_error_t error;	printf("/nbroker:sgraph 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 *graph_resp_json = json_loadb(data,					     data_size, 0,					     &error);	zmsg_destroy(&msg);	//identify the request	int32_t requestId =	    json_integer_value(json_object_get(graph_resp_json, "requestId"));	req_t *req = request_store_req(req_store, requestId);	json_t *response = json_object_get(graph_resp_json, "response");	const char *resp_type =	    json_string_value(json_object_get(response, "type"));	json_t *request =	    json_object_get(json_object_get(req->request, "clientRequest"),			    "request");	const char *req_type =	    json_string_value(json_object_get(request, "type"));	if ((strcmp(resp_type, "retrieveResponse") == 0)	    && (strcmp(req_type, "searchRequest") == 0))		graph_response_retrieveResponse(req, response,						requestId, sweb, req_store);	else if ((strcmp(resp_type, "newNodeResponse") == 0)		 && (strcmp(req_type, "newNode") == 0))		graph_response_newNodeResponse(request,					       response,					       requestId, spss, req_store);	else if ((strcmp(resp_type, "newLinkResponse") == 0)		 && (strcmp(req_type, "newLink") == 0))		graph_response_newLinkResponse(req,					       request,					       response,					       requestId, sweb, req_store);	else if ((strcmp(resp_type, "delLinkResponse")		  == 0) && (strcmp(req_type, "delLink")			    == 0))		graph_response_delLinkResponse(req,					       request,					       response,					       requestId, sweb, req_store);	else if ((strcmp(resp_type, "delNode") == 0)		 && (strcmp(req_type, "delNode")		     == 0))		graph_response_delNodeResponse(req,					       request,					       response,					       requestId, spss, req_store);	else if ((strcmp(resp_type, "newNodeData")		  == 0) && (strcmp(req_type, "newNodeData")			    == 0))		graph_response_newNodeDataResponse(req,						   request,						   response,						   requestId, sweb, req_store);	else if ((strcmp(resp_type, "newLinkData")		  == 0) && (strcmp(req_type, "newLinkData")			    == 0))		graph_response_newLinkDataResponse(req,						   request,						   response,						   requestId, sweb, req_store);	json_decref(graph_resp_json);}

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_pushstr (request, service);    zmsg_pushstr (request, MDPC_CLIENT);    if (self->verbose) {        zclock_log ("I: send request to '%s' service:", service);        zmsg_dump (request);    }    int retries_left = self->retries;    while (retries_left && !zctx_interrupted) {        zmsg_t *msg = zmsg_dup (request);        zmsg_send (&msg, self->client);        //  Poll socket for a reply, with timeout        zmq_pollitem_t items [] = {            { self->client, 0, ZMQ_POLLIN, 0 } };        int rc = zmq_poll (items, 1, self->timeout * ZMQ_POLL_MSEC);        if (rc == -1)            break;          //  Interrupted        //  If we got a reply, process it        if (items [0].revents & ZMQ_POLLIN) {            zmsg_t *msg = zmsg_recv (self->client);            if (self->verbose) {                zclock_log ("I: received reply:");                zmsg_dump (msg);            }            //  Don't try to handle errors, just assert noisily            assert (zmsg_size (msg) >= 3);            zframe_t *header = zmsg_pop (msg);            assert (zframe_streq (header, MDPC_CLIENT));            zframe_destroy (&header);            zframe_t *reply_service = zmsg_pop (msg);            assert (zframe_streq (reply_service, service));            zframe_destroy (&reply_service);            zmsg_destroy (&request);            return msg;     //  Success        }        else        if (--retries_left) {            if (self->verbose)                zclock_log ("W: no reply, reconnecting...");            //  Reconnect socket            s_mdcli_connect_to_broker (self);        }        else {            if (self->verbose)                zclock_log ("W: permanent error, abandoning");            break;          //  Give up        }    }    if (zctx_interrupted)        printf ("W: interrupt received, killing client.../n");    zmsg_destroy (&request);    return NULL;}

int main (void){    zctx_t *context = zctx_new ();    void *frontend = zsocket_new (context, ZMQ_SUB);    zsocket_bind (frontend, "tcp://*:5557");    void *backend = zsocket_new (context, ZMQ_XPUB);    zsocket_bind (backend, "tcp://*:5558");    //  Subscribe to every single topic from publisher    zsocket_set_subscribe (frontend, "");    //  Store last instance of each topic in a cache    zhash_t *cache = zhash_new ();    //  .split main poll loop    //  We route topic updates from frontend to backend, and    //  we handle subscriptions by sending whatever we cached,    //  if anything:    while (true) {        zmq_pollitem_t items [] = {            { frontend, 0, ZMQ_POLLIN, 0 },            { backend,  0, ZMQ_POLLIN, 0 }        };        if (zmq_poll (items, 2, 1000 * ZMQ_POLL_MSEC) == -1)            break;              //  Interrupted        //  Any new topic data we cache and then forward        if (items [0].revents & ZMQ_POLLIN) {            char *topic = zstr_recv (frontend);            char *current = zstr_recv (frontend);            if (!topic)                break;            char *previous = zhash_lookup (cache, topic);            if (previous) {                zhash_delete (cache, topic);                free (previous);            }            zhash_insert (cache, topic, current);            zstr_sendm (backend, topic);            zstr_send (backend, current);            free (topic);        }        //  .split handle subscriptions        //  When we get a new subscription we pull data from the cache:        if (items [1].revents & ZMQ_POLLIN) {            zframe_t *frame = zframe_recv (backend);            if (!frame)                break;            //  Event is one byte 0=unsub or 1=sub, followed by topic            byte *event = zframe_data (frame);            if (event [0] == 1) {                char *topic = zmalloc (zframe_size (frame));                memcpy (topic, event + 1, zframe_size (frame) - 1);                printf ("Sending cached topic %s/n", topic);                char *previous = zhash_lookup (cache, topic);                if (previous) {                    zstr_sendm (backend, topic);                    zstr_send (backend, previous);                }                free (topic);            }            zframe_destroy (&frame);        }    }    zctx_destroy (&context);    zhash_destroy (&cache);    return 0;}

intzsocket_test (bool verbose){    printf (" * zsocket: ");    //  @selftest    zctx_t *ctx = zctx_new ();    assert (ctx);    //  Create a detached thread, let it run    char *interf = "*";    char *domain = "localhost";    int service = 5560;    void *writer = zsocket_new (ctx, ZMQ_PUSH);    assert (writer);    void *reader = zsocket_new (ctx, ZMQ_PULL);    assert (reader);    assert (streq (zsocket_type_str (writer), "PUSH"));    assert (streq (zsocket_type_str (reader), "PULL"));    int rc = zsocket_bind (writer, "tcp://%s:%d", interf, service);    assert (rc == service);    rc = zsocket_connect (reader, "tcp://%s:%d", domain, service);    assert (rc == 0);    zstr_send (writer, "HELLO");    char *message = zstr_recv (reader);    assert (message);    assert (streq (message, "HELLO"));    free (message);        //  Test binding to ports    int port = zsocket_bind (writer, "tcp://%s:*", interf);    assert (port >= ZSOCKET_DYNFROM && port <= ZSOCKET_DYNTO);    assert (zsocket_poll (writer, 100) == false);    rc = zsocket_connect (reader, "txp://%s:%d", domain, service);    assert (rc == -1);    //  Test sending frames to socket    rc = zsocket_sendmem (writer,"ABC", 3, ZFRAME_MORE);    assert (rc == 0);    rc = zsocket_sendmem (writer, "DEFG", 4, 0);    assert (rc == 0);        zframe_t *frame = zframe_recv (reader);    assert (frame);    assert (zframe_streq (frame, "ABC"));    assert (zframe_more (frame));    zframe_destroy (&frame);        frame = zframe_recv (reader);    assert (frame);    assert (zframe_streq (frame, "DEFG"));    assert (!zframe_more (frame));    zframe_destroy (&frame);    //  Test zframe_sendmem_zero_copy    rc = zsocket_sendmem_zero_copy (writer, strdup ("ABC"), 3,                                    s_test_free_str_cb, NULL, ZFRAME_MORE);    assert (rc == 0);    rc = zsocket_sendmem_zero_copy (writer, strdup ("DEFG"), 4,                                    s_test_free_str_cb, NULL, 0);    assert (rc == 0);        frame = zframe_recv (reader);    assert (frame);    assert (zframe_streq (frame, "ABC"));    assert (zframe_more (frame));    zframe_destroy (&frame);        frame = zframe_recv (reader);    assert (frame);    assert (zframe_streq (frame, "DEFG"));    assert (!zframe_more (frame));    zframe_destroy (&frame);    zsocket_destroy (ctx, writer);    zctx_destroy (&ctx);    //  @end    printf ("OK/n");    return 0;}

voidzyre_test (bool verbose){    printf (" * zyre: ");    //  @selftest    //  We'll use inproc gossip discovery so that this works without networking        int major, minor, patch;    zyre_version (&major, &minor, &patch);    assert (major == ZYRE_VERSION_MAJOR);    assert (minor == ZYRE_VERSION_MINOR);    assert (patch == ZYRE_VERSION_PATCH);        //  Create two nodes    zyre_t *node1 = zyre_new ("node1");    assert (node1);    assert (streq (zyre_name (node1), "node1"));    zyre_set_header (node1, "X-HELLO", "World");    zyre_set_verbose (node1);    //  Set inproc endpoint for this node    zyre_set_endpoint (node1, "inproc://zyre-node1");    //  Set up gossip network for this node    zyre_gossip_bind (node1, "inproc://gossip-hub");    int rc = zyre_start (node1);    assert (rc == 0);    zyre_t *node2 = zyre_new ("node2");    assert (node2);    assert (streq (zyre_name (node2), "node2"));    zyre_set_verbose (node2);    //  Set inproc endpoint for this node    //  First, try to use existing name, it'll fail    zyre_set_endpoint (node2, "inproc://zyre-node1");    assert (streq (zyre_endpoint (node2), ""));    //  Now use available name and confirm that it succeeds    zyre_set_endpoint (node2, "inproc://zyre-node2");    assert (streq (zyre_endpoint (node2), "inproc://zyre-node2"));       //  Set up gossip network for this node    zyre_gossip_connect (node2, "inproc://gossip-hub");    rc = zyre_start (node2);    assert (rc == 0);    assert (strneq (zyre_uuid (node1), zyre_uuid (node2)));        zyre_join (node1, "GLOBAL");    zyre_join (node2, "GLOBAL");    //  Give time for them to interconnect    zclock_sleep (100);    //  One node shouts to GLOBAL    zyre_shouts (node1, "GLOBAL", "Hello, World");    //  Second node should receive ENTER, JOIN, and SHOUT    zmsg_t *msg = zyre_recv (node2);    assert (msg);    char *command = zmsg_popstr (msg);    assert (streq (command, "ENTER"));    zstr_free (&command);    assert (zmsg_size (msg) == 4);    char *peerid = zmsg_popstr (msg);    zstr_free (&peerid);    char *name = zmsg_popstr (msg);    assert (streq (name, "node1"));    zstr_free (&name);    zframe_t *headers_packed = zmsg_pop (msg);    char *peeraddress = zmsg_popstr (msg);    zstr_free (&peeraddress);    assert (headers_packed);    zhash_t *headers = zhash_unpack (headers_packed);    assert (headers);    zframe_destroy (&headers_packed);    assert (streq ((char*)zhash_lookup (headers, "X-HELLO"), "World"));    zhash_destroy (&headers);    zmsg_destroy (&msg);    msg = zyre_recv (node2);    assert (msg);    command = zmsg_popstr (msg);    assert (streq (command, "JOIN"));    zstr_free (&command);    assert (zmsg_size (msg) == 3);    zmsg_destroy (&msg);    msg = zyre_recv (node2);    assert (msg);    command = zmsg_popstr (msg);    assert (streq (command, "SHOUT"));    zstr_free (&command);    zmsg_destroy (&msg);        zyre_stop (node1);    zyre_stop (node2);        zyre_destroy (&node1);    zyre_destroy (&node2);    //  @end    printf ("OK/n");//.........这里部分代码省略.........

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

//.........这里部分代码省略.........    self->needle = zframe_data (frame);    self->ceiling = self->needle + zframe_size (frame);    uint16_t signature;    GET_NUMBER2 (signature);    if (signature != (0xAAA0 | 1))        goto empty;             //  Invalid signature    //  Get message id and parse per message type    GET_NUMBER1 (self->id);    switch (self->id) {        case ZRE_MSG_HELLO:            GET_NUMBER2 (self->sequence);            GET_STRING (self->ipaddress);            GET_NUMBER2 (self->mailbox);            {                size_t list_size;                GET_NUMBER4 (list_size);                self->groups = zlist_new ();                zlist_autofree (self->groups);                while (list_size--) {                    char *string;                    GET_LONGSTR (string);                    zlist_append (self->groups, string);                    free (string);                }            }            GET_NUMBER1 (self->status);            {                size_t hash_size;                GET_NUMBER4 (hash_size);                self->headers = zhash_new ();                zhash_autofree (self->headers);                while (hash_size--) {                    char *key, *value;                    GET_STRING (key);                    GET_LONGSTR (value);                    zhash_insert (self->headers, key, value);                    free (key);                    free (value);                }            }            break;        case ZRE_MSG_WHISPER:            GET_NUMBER2 (self->sequence);            //  Get zero or more remaining frames, leaving current            //  frame untouched            self->content = zmsg_new ();            while (zmsg_size (msg))                zmsg_add (self->content, zmsg_pop (msg));            break;        case ZRE_MSG_SHOUT:            GET_NUMBER2 (self->sequence);            GET_STRING (self->group);            //  Get zero or more remaining frames, leaving current            //  frame untouched            self->content = zmsg_new ();            while (zmsg_size (msg))                zmsg_add (self->content, zmsg_pop (msg));            break;        case ZRE_MSG_JOIN:            GET_NUMBER2 (self->sequence);            GET_STRING (self->group);            GET_NUMBER1 (self->status);            break;        case ZRE_MSG_LEAVE:            GET_NUMBER2 (self->sequence);            GET_STRING (self->group);            GET_NUMBER1 (self->status);            break;        case ZRE_MSG_PING:            GET_NUMBER2 (self->sequence);            break;        case ZRE_MSG_PING_OK:            GET_NUMBER2 (self->sequence);            break;        default:            goto malformed;    }    //  Successful return    zframe_destroy (&frame);    zmsg_destroy (msg_p);    return self;    //  Error returns    malformed:        printf ("E: malformed message '%d'/n", self->id);    empty:        zframe_destroy (&frame);        zmsg_destroy (msg_p);        zre_msg_destroy (&self);        return (NULL);}

intzgossip_msg_recv (zgossip_msg_t *self, zsock_t *input){    assert (input);    if (zsock_type (input) == ZMQ_ROUTER) {        zframe_destroy (&self->routing_id);        self->routing_id = zframe_recv (input);        if (!self->routing_id || !zsock_rcvmore (input)) {            zsys_warning ("zgossip_msg: no routing ID");            return -1;          //  Interrupted or malformed        }    }    zmq_msg_t frame;    zmq_msg_init (&frame);    int size = zmq_msg_recv (&frame, zsock_resolve (input), 0);    if (size == -1) {        zsys_warning ("zgossip_msg: interrupted");        goto malformed;         //  Interrupted    }    //  Get and check protocol signature    self->needle = (byte *) zmq_msg_data (&frame);    self->ceiling = self->needle + zmq_msg_size (&frame);    uint16_t signature;    GET_NUMBER2 (signature);    if (signature != (0xAAA0 | 0)) {        zsys_warning ("zgossip_msg: invalid signature");        //  TODO: discard invalid messages and loop, and return        //  -1 only on interrupt        goto malformed;         //  Interrupted    }    //  Get message id and parse per message type    GET_NUMBER1 (self->id);    switch (self->id) {        case ZGOSSIP_MSG_HELLO:            {                byte version;                GET_NUMBER1 (version);                if (version != 1) {                    zsys_warning ("zgossip_msg: version is invalid");                    goto malformed;                }            }            break;        case ZGOSSIP_MSG_PUBLISH:            {                byte version;                GET_NUMBER1 (version);                if (version != 1) {                    zsys_warning ("zgossip_msg: version is invalid");                    goto malformed;                }            }            GET_STRING (self->key);            GET_LONGSTR (self->value);            GET_NUMBER4 (self->ttl);            break;        case ZGOSSIP_MSG_PING:            {                byte version;                GET_NUMBER1 (version);                if (version != 1) {                    zsys_warning ("zgossip_msg: version is invalid");                    goto malformed;                }            }            break;        case ZGOSSIP_MSG_PONG:            {                byte version;                GET_NUMBER1 (version);                if (version != 1) {                    zsys_warning ("zgossip_msg: version is invalid");                    goto malformed;                }            }            break;        case ZGOSSIP_MSG_INVALID:            {                byte version;                GET_NUMBER1 (version);                if (version != 1) {                    zsys_warning ("zgossip_msg: version is invalid");                    goto malformed;                }            }            break;        default:            zsys_warning ("zgossip_msg: bad message ID");            goto malformed;    }    //  Successful return    zmq_msg_close (&frame);//.........这里部分代码省略.........

/////  Destroy a frameQZframe::~QZframe (){    zframe_destroy (&self);}

static rsRetVal rcvData(){	DEFiRet;		if(!listenerList) {		listenerList = zlist_new();		if(!listenerList) {			errmsg.LogError(0, NO_ERRCODE, "could not allocate list");			ABORT_FINALIZE(RS_RET_ERR);		}	}	zactor_t *authActor;	zcert_t *serverCert;	if(runModConf->authenticator == 1) {		authActor = zactor_new(zauth, NULL);		zstr_sendx(authActor, "CURVE", runModConf->clientCertPath, NULL);		zsock_wait(authActor);	} 	instanceConf_t *inst;	for(inst = runModConf->root; inst != NULL; inst=inst->next) {		CHKiRet(addListener(inst));	}		zpoller_t *poller = zpoller_new(NULL);	if(!poller) {		errmsg.LogError(0, NO_ERRCODE, "could not create poller");			ABORT_FINALIZE(RS_RET_ERR);	}	DBGPRINTF("imczmq: created poller/n");	struct listener_t *pData;	pData = zlist_first(listenerList);	if(!pData) {		errmsg.LogError(0, NO_ERRCODE, "imczmq: no listeners were "						"started, input not activated./n");		ABORT_FINALIZE(RS_RET_NO_RUN);	}	while(pData) {		int rc = zpoller_add(poller, pData->sock);		if(rc != 0) {			errmsg.LogError(0, NO_ERRCODE, "imczmq: could not add "						"socket to poller, input not activated./n");			ABORT_FINALIZE(RS_RET_NO_RUN);		}		pData = zlist_next(listenerList);	}	zframe_t *frame;	zsock_t *which = (zsock_t *)zpoller_wait(poller, -1);	while(which) {		if (zpoller_terminated(poller)) {				break;		}		pData = zlist_first(listenerList);		while(pData->sock != which) {			pData = zlist_next(listenerList);		}			if(which == pData->sock) {			DBGPRINTF("imczmq: found matching socket/n");		}		frame = zframe_recv(which);		char *buf = zframe_strdup(frame);		if(buf == NULL) {			DBGPRINTF("imczmq: null buffer/n");			continue;		}		smsg_t *pMsg;		if(msgConstruct(&pMsg) == RS_RET_OK) {			MsgSetRawMsg(pMsg, buf, strlen(buf));			MsgSetInputName(pMsg, s_namep);			MsgSetHOSTNAME(pMsg, glbl.GetLocalHostName(), ustrlen(glbl.GetLocalHostName()));			MsgSetRcvFrom(pMsg, glbl.GetLocalHostNameProp());			MsgSetRcvFromIP(pMsg, glbl.GetLocalHostIP());			MsgSetMSGoffs(pMsg, 0);			MsgSetFlowControlType(pMsg, eFLOWCTL_NO_DELAY);			MsgSetRuleset(pMsg, pData->ruleset);			pMsg->msgFlags = NEEDS_PARSING | PARSE_HOSTNAME;			submitMsg2(pMsg);		}		free(buf);		which = (zsock_t *)zpoller_wait(poller, -1);	}finalize_it:	zframe_destroy(&frame);	zpoller_destroy(&poller);	pData = zlist_first(listenerList);	while(pData) {		zsock_destroy(&pData->sock);		free(pData->ruleset);		pData = zlist_next(listenerList);	}	zlist_destroy(&listenerList);//.........这里部分代码省略.........

static ints_agent_handle_router (agent_t *self){    zframe_t *address = zframe_recv (self->router);    char *hashkey = zframe_strhex (address);    client_t *client = (client_t *) zhash_lookup (self->clients, hashkey);    if (client == NULL    && self->nbr_pending < self->max_pending) {        client = client_new (self, address);        client_set_pending (client);        curve_codec_set_verbose (client->codec, self->verbose);        zhash_foreach (self->metadata, client_set_metadata, client);        zhash_insert (self->clients, hashkey, client);        zhash_freefn (self->clients, hashkey, client_free);    }    free (hashkey);    zframe_destroy (&address);    //  If we're overloaded, discard client request without any further    //  ado. The client will have to detect this and retry later.    //  TODO: retry in client side to handle overloaded servers.    if (client == NULL)        return 0;    //  If not yet connected, process one command frame    //  We always read one request, and send one reply    if (client->state == pending) {        zframe_t *input = zframe_recv (self->router);        zframe_t *output = curve_codec_execute (client->codec, &input);        if (output) {            zframe_send (&client->address, self->router, ZFRAME_MORE + ZFRAME_REUSE);            zframe_send (&output, self->router, 0);            if (curve_codec_connected (client->codec))                client_set_connected (client);        }        else            client_set_exception (client);    }    else    //  If connected, process one message frame    //  We will queue message frames in the client until we get a    //  whole message ready to deliver up the data socket -- frames    //  from different clients will be randomly intermixed.    if (client->state == connected) {        zframe_t *encrypted = zframe_recv (self->router);        zframe_t *cleartext = curve_codec_decode (client->codec, &encrypted);        if (cleartext) {            if (client->incoming == NULL)                client->incoming = zmsg_new ();            zmsg_add (client->incoming, cleartext);            if (!zframe_more (cleartext)) {                zmsg_pushstr (client->incoming, client->hashkey);                zmsg_send (&client->incoming, self->data);            }        }        else            client_set_exception (client);    }    //  If client is misbehaving, remove it    if (client->state == exception)        zhash_delete (self->clients, client->hashkey);    return 0;}

//.........这里部分代码省略.........    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);    assert (rc == 0);    rc = zmsg_addmem (msg, blank, 253);    assert (rc == 0);

void rb_czmq_free_frame(zframe_t *frame){    if (frame)        if (st_lookup(frames_map, (st_data_t)frame, 0)) zframe_destroy(&frame);}

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

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

intzre_log_msg_send (zre_log_msg_t **self_p, void *output){    assert (output);    assert (self_p);    assert (*self_p);    //  Calculate size of serialized data    zre_log_msg_t *self = *self_p;    size_t frame_size = 2 + 1;          //  Signature and message ID    switch (self->id) {        case ZRE_LOG_MSG_LOG:            //  level is a 1-byte integer            frame_size += 1;            //  event is a 1-byte integer            frame_size += 1;            //  node is a 2-byte integer            frame_size += 2;            //  peer is a 2-byte integer            frame_size += 2;            //  time is a 8-byte integer            frame_size += 8;            //  data is a string with 1-byte length            frame_size++;       //  Size is one octet            if (self->data)                frame_size += strlen (self->data);            break;                    default:            printf ("E: 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);    size_t string_size;    int frame_flags = 0;    PUT_NUMBER2 (0xAAA0 | 2);    PUT_NUMBER1 (self->id);    switch (self->id) {        case ZRE_LOG_MSG_LOG:            PUT_NUMBER1 (self->level);            PUT_NUMBER1 (self->event);            PUT_NUMBER2 (self->node);            PUT_NUMBER2 (self->peer);            PUT_NUMBER8 (self->time);            if (self->data) {                PUT_STRING (self->data);            }            else                PUT_NUMBER1 (0);    //  Empty string            break;                }    //  If we're sending to a ROUTER, we send the address first    if (zsocket_type (output) == ZMQ_ROUTER) {        assert (self->address);        if (zframe_send (&self->address, output, ZFRAME_MORE)) {            zframe_destroy (&frame);            zre_log_msg_destroy (self_p);            return -1;        }    }    //  Now send the data frame    if (zframe_send (&frame, output, frame_flags)) {        zframe_destroy (&frame);        zre_log_msg_destroy (self_p);        return -1;    }    //  Destroy zre_log_msg object    zre_log_msg_destroy (self_p);    return 0;}