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

static ints_agent_handle_dealer (agent_t *self){    if (self->state == connecting) {        zframe_t *input = zframe_recv (self->dealer);        zframe_t *output = curve_codec_execute (self->codec, &input);        if (output)            zframe_send (&output, self->dealer, 0);        else        if (curve_codec_connected (self->codec))            self->state = connected;        else        if (curve_codec_exception (self->codec))            self->state = exception;    }    else    if (self->state == connected) {        zframe_t *encrypted = zframe_recv (self->dealer);        zframe_t *cleartext = curve_codec_decode (self->codec, &encrypted);        if (cleartext) {            int flags = zframe_more (cleartext)? ZFRAME_MORE: 0;            zframe_send (&cleartext, self->data, flags);        }        else            self->state = exception;    }    return 0;}

    void unregister(std::string const &name) {	directoryd::ServiceRequest request;	request.set_type(directoryd::UNREGISTER);	auto *r = request.mutable_unregister();	r->set_name(name);	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().register_socket(), 0);	assert(retval == 0);	zframe_t *rf = zframe_recv (DDClient::instance().register_socket());	directoryd::ServiceReply reply;	reply.ParseFromArray(zframe_data(rf),zframe_size(rf));	zframe_destroy(&rf);	if (reply.type() != directoryd::UNREGISTER) {	    throw RegistrationError("Got back incorrect message type when trying to unregister.");	}	if (reply.success() != true) {	    throw RegistrationError(reply.result());	}    }

int handle_event(zloop_t *loop, zsock_t *reader, void *args) {         // initialization        ubx_block_t *b = (ubx_block_t *) args;        struct zmq_receiver_info *inf = (struct zmq_receiver_info*) b->private_data;        printf("zmq_receiver: data available./n");        zframe_t *frame = zframe_recv (reader);        // print out frame data        zframe_print (frame, NULL);        // move to step function?        ubx_type_t* type =  ubx_type_get(b->ni, "unsigned char");        ubx_data_t msg;        msg.data = (void *)zframe_data(frame);        msg.len = zframe_size(frame);        msg.type = type;        //hexdump((unsigned char *)msg.data, msg.len, 16);        __port_write(inf->ports.zmq_in, &msg);        /* Inform potential observers ? */        // clean up temporary frame object        zframe_destroy (&frame);        return 1;}

zmsg_t *zmsg_recv (void *source){    assert (source);    zmsg_t *self = zmsg_new ();    if (!self)        return NULL;    while (true) {        zframe_t *frame = zframe_recv (source);        if (!frame) {            if (errno == EINTR && zlist_head (self->frames))                continue;            else {                zmsg_destroy (&self);                break;              //  Interrupted or terminated            }        }        if (zmsg_append (self, &frame)) {            zmsg_destroy (&self);            break;        }        if (!zsock_rcvmore (source))            break;              //  Last message frame    }    return self;}

int main (void){    zctx_t *ctx = zctx_new ();    zctx_set_linger (ctx, 1000);        void *pub = zsocket_new (ctx, ZMQ_XPUB);    zsocket_set_hwm (pub, 0);    zsocket_connect (pub, "tcp://127.0.0.1:9000");    //  Wait for subscriber to subscribe    zframe_t *frame = zframe_recv (pub);    zframe_destroy (&frame);        //  Send HELLOs for five seconds    size_t total = 0;    int64_t finish_at = zclock_time () + 5000;        while (zclock_time () < finish_at) {        //  Send 100K and then check time again        int count = 0;        for (count = 0; count < 100000; count++)            zstr_send (pub, "HELLO");        total++;    }    printf ("%zd00000 messages sent/n", total);        zstr_send (pub, "WORLD");    zctx_destroy (&ctx);    return 0;}

static ints_snapshots (zloop_t *loop, zmq_pollitem_t *poller, void *args){    clonesrv_t *self = (clonesrv_t *) args;    zframe_t *identity = zframe_recv (poller->socket);    if (identity) {        //  Request is in second frame of message        char *request = zstr_recv (poller->socket);        char *subtree = NULL;        if (streq (request, "ICANHAZ?")) {            free (request);            subtree = zstr_recv (poller->socket);        }        else            printf ("E: bad request, aborting/n");        if (subtree) {            //  Send state socket to client            kvroute_t routing = { poller->socket, identity, subtree };            zhash_foreach (self->kvmap, s_send_single, &routing);            //  Now send END message with sequence number            zclock_log ("I: sending shapshot=%d", (int) self->sequence);            zframe_send (&identity, poller->socket, ZFRAME_MORE);            kvmsg_t *kvmsg = kvmsg_new (self->sequence);            kvmsg_set_key  (kvmsg, "KTHXBAI");            kvmsg_set_body (kvmsg, (byte *) subtree, 0);            kvmsg_send     (kvmsg, poller->socket);            kvmsg_destroy (&kvmsg);            free (subtree);        }    }    return 0;}

zframe_t *zchanneler_recv (zchanneler_t *self){    assert (self);    zframe_t *frame = zframe_recv (self->server);    return frame;}

static voidstate_manager (void *args, zctx_t *ctx, void *pipe){    zhash_t *kvmap = zhash_new ();    zstr_send (pipe, "READY");    void *snapshot = zsocket_new (ctx, ZMQ_ROUTER);    zsocket_bind (snapshot, "tcp://*:5556");    zmq_pollitem_t items [] = {        { pipe, 0, ZMQ_POLLIN, 0 },        { snapshot, 0, ZMQ_POLLIN, 0 }    };    int64_t sequence = 0;       //  Current snapshot version number    while (!zctx_interrupted) {        int rc = zmq_poll (items, 2, -1);        if (rc == -1 && errno == ETERM)            break;              //  Context has been shut down        //  Apply state update from main thread        if (items [0].revents & ZMQ_POLLIN) {            kvmsg_t *kvmsg = kvmsg_recv (pipe);            if (!kvmsg)                break;          //  Interrupted            sequence = kvmsg_sequence (kvmsg);            kvmsg_store (&kvmsg, kvmap);        }        //  Execute state snapshot request        if (items [1].revents & ZMQ_POLLIN) {            zframe_t *identity = zframe_recv (snapshot);            if (!identity)                break;          //  Interrupted            //  Request is in second frame of message            char *request = zstr_recv (snapshot);            if (streq (request, "ICANHAZ?"))                free (request);            else {                printf ("E: bad request, aborting/n");                break;            }            //  Send state snapshot to client            kvroute_t routing = { snapshot, identity };            //  For each entry in kvmap, send kvmsg to client            zhash_foreach (kvmap, s_send_single, &routing);            //  Now send END message with sequence number            printf ("Sending state shapshot=%d/n", (int) sequence);            zframe_send (&identity, snapshot, ZFRAME_MORE);            kvmsg_t *kvmsg = kvmsg_new (sequence);            kvmsg_set_key  (kvmsg, "KTHXBAI");            kvmsg_set_body (kvmsg, (byte *) "", 0);            kvmsg_send     (kvmsg, snapshot);            kvmsg_destroy (&kvmsg);        }    }    zhash_destroy (&kvmap);}

static voids_socket_event (agent_t *self){    //  First frame is event number and value    zframe_t *frame = zframe_recv (self->socket);    int event = *(uint16_t *) (zframe_data (frame));    int value = *(uint32_t *) (zframe_data (frame) + 2);    zframe_destroy (&frame);        //  Second frame is address    char *address = zstr_recv (self->socket);    char *description = "Unknown";    switch (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:            zsys_error ("illegal socket monitor event: %d", event);            break;    }    if (self->verbose)        zsys_info ("zmonitor: %s - %s/n", description, address);    zstr_sendfm (self->pipe, "%d", event);    zstr_sendfm (self->pipe, "%d", value);    zstr_sendm  (self->pipe, address);    zstr_send   (self->pipe, description);    free (address);}

static void listener_thread (void *args, zctx_t *ctx, void *pipe){	while (true) {		zframe_t *frame = zframe_recv (pipe);		if (!frame)			break; //  Interrupted		//zframe_print (frame, NULL);		zframe_destroy (&frame);	}}

static voidlistener_thread (void *args, zctx_t *ctx, void *pipe){    //  Print everything that arrives on pipe    while (true) {        zframe_t *frame = zframe_recv (pipe);        if (!frame)            break;              //  Interrupted        zframe_print (frame, NULL);        zframe_destroy (&frame);    }}

static ints_self_handle_pipe (self_t *self){    //  Get just the command off the pipe    char *command = zstr_recv (self->pipe);    if (!command)        return -1;                  //  Interrupted    if (self->verbose)        zsys_info ("zbeacon: API command=%s", command);    if (streq (command, "VERBOSE"))        self->verbose = true;    else    if (streq (command, "CONFIGURE")) {        int port;        int rc = zsock_recv (self->pipe, "i", &port);        assert (rc == 0);        s_self_configure (self, port);    }    else    if (streq (command, "PUBLISH")) {        zframe_destroy (&self->transmit);        zsock_recv (self->pipe, "fi", &self->transmit, &self->interval);        assert (zframe_size (self->transmit) <= UDP_FRAME_MAX);        if (self->interval == 0)            self->interval = INTERVAL_DFLT;        //  Start broadcasting immediately        self->ping_at = zclock_mono ();    }    else    if (streq (command, "SILENCE"))        zframe_destroy (&self->transmit);    else    if (streq (command, "SUBSCRIBE")) {        zframe_destroy (&self->filter);        self->filter = zframe_recv (self->pipe);        assert (zframe_size (self->filter) <= UDP_FRAME_MAX);    }    else    if (streq (command, "UNSUBSCRIBE"))        zframe_destroy (&self->filter);    else    if (streq (command, "$TERM"))        self->terminated = true;    else {        zsys_error ("zbeacon: - invalid command: %s", command);        assert (false);    }    zstr_free (&command);    return 0;}

int main (void){    zctx_t *ctx = zctx_new ();    void *server = zsocket_new (ctx, ZMQ_XPUB);    zsocket_bind (server, "tcp://*:6001");    void *client = zsocket_new (ctx, ZMQ_XSUB);    zsocket_connect (client, "tcp://localhost:6001");    byte subscribe [] = { 1 };        zmq_send (client, &subscribe, sizeof (subscribe), 0);    zmq_send (client, "MOREMOREMORE", 12, ZMQ_SNDMORE);    zmq_send (client, "LAST", 4, 0);    //  First message is subscription, 1 byte    zframe_t *frame = zframe_recv (server);    assert (zframe_size (frame) == 1);    assert (zsocket_rcvmore (server) == 0);    zframe_destroy (&frame);        //  Second message is 12 bytes with MORE    frame = zframe_recv (server);    assert (zframe_size (frame) == 12);    assert (zsocket_rcvmore (server) == 1);    zframe_destroy (&frame);        //  Third message is 4 bytes with no more    frame = zframe_recv (server);    assert (zframe_size (frame) == 4);    assert (zsocket_rcvmore (server) == 0);    zframe_destroy (&frame);        zctx_destroy (&ctx);    return 0;}

static voidcounter_task (void *args, zctx_t *ctx, void *pipe){    void *counter = zsocket_new (ctx, ZMQ_ROUTER);    zsocket_bind (counter, "tcp://*:6001");    //  Parameters for the census    int census_msec = 250;       //  Msecs to settle down    //  Calling thread tells us the population size    char *population = zstr_recv (pipe);        //  All activity happens on our counter socket    zmq_pollitem_t items [] = { { counter, 0, ZMQ_POLLIN, 0 } };    int headcount = 0;         //  Known target size    int positives = 0;         //  How many said "yes"        int64_t timer_end = zclock_time () + census_msec;    int still_waiting = atoi (population);    while (still_waiting) {        int64_t time_left = timer_end - zclock_time ();        if (time_left <= 0)            break;              //  We're done here        int rc = zmq_poll (items, 1, time_left * ZMQ_POLL_MSEC);        if (rc == -1)            break;              //  Interrupted        if (items [0].revents & ZMQ_POLLIN) {            zframe_t *address = zframe_recv (counter);            char *message = zstr_recv (counter);            if (streq (message, "Hello")) {                headcount++;                zframe_send (&address, counter, ZFRAME_MORE);                zstr_send (counter, "Who wants pizza?");            }            else            if (streq (message, "Yes"))                positives++;                        zframe_destroy (&address);            free (message);        }    }    printf ("Out of %d people, %d want pizza/n", headcount, positives);    zstr_send (pipe, "DONE");}

    void register_service(std::string const &name,			  int port,			  std::map<std::string, std::string> const &txt)    {	directoryd::ServiceRequest request;	request.set_type(directoryd::REGISTER);	auto *r = request.mutable_register_();	auto l = r->add_location();	l->set_port(port);	l->set_type("_hotdec._tcp");	r->set_name(name);	for (auto &t : txt) {	    auto txtfield = r->add_txt();	    txtfield->set_key(t.first);	    txtfield->set_value(t.second);	}	zframe_t *sf = zframe_new(NULL, request.ByteSize());	assert (sf != NULL);	request.SerializeToArray(zframe_data(sf),zframe_size(sf));	string buffer;	if (debug && TextFormat::PrintToString(request, &buffer)) {	    fprintf(stderr, "request: %s/n", buffer.c_str());	}	int retval = zframe_send(&sf, DDClient::instance().register_socket(), 0);	assert(retval == 0);	zframe_t *rf = zframe_recv (DDClient::instance().register_socket());	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);	if (reply.type() != directoryd::REGISTER) {	    throw RegistrationError("Got back incorrect message type when trying to register.");	}	if (reply.success() != true) {	    throw RegistrationError(reply.result());	}	RegistrationManager::instance().add(name);    }

zmsg_t *zmsg_recv (void *socket){    assert (socket);    zmsg_t *self = zmsg_new ();    while (1) {        zframe_t *frame = zframe_recv (socket);        if (!frame) {            zmsg_destroy (&self);            break;              //  Interrupted or terminated        }        zmsg_add (self, frame);        if (!zframe_more (frame))            break;              //  Last message frame    }    return self;}

static intzyre_node_recv_beacon (zyre_node_t *self){    //  Get IP address and beacon of peer    char *ipaddress = zstr_recv (zbeacon_socket (self->beacon));    zframe_t *frame = zframe_recv (zbeacon_socket (self->beacon));    //  Ignore anything that isn't a valid beacon    bool is_valid = true;    beacon_t beacon;    if (zframe_size (frame) == sizeof (beacon_t)) {        memcpy (&beacon, zframe_data (frame), zframe_size (frame));        if (beacon.version != BEACON_VERSION)            is_valid = false;    }    else        is_valid = false;    //  Check that the peer, identified by its UUID, exists    if (is_valid) {        zuuid_t *uuid = zuuid_new ();        zuuid_set (uuid, beacon.uuid);        if (beacon.port) {            char endpoint [30];            sprintf (endpoint, "tcp://%s:%d", ipaddress, ntohs (beacon.port));            zyre_peer_t *peer = zyre_node_require_peer (self, uuid, endpoint);            zyre_peer_refresh (peer);        }        else {            //  Zero port means peer is going away; remove it if            //  we had any knowledge of it already            zyre_peer_t *peer = (zyre_peer_t *) zhash_lookup (                self->peers, zuuid_str (uuid));            if (peer)                zyre_node_remove_peer (self, peer);        }        zuuid_destroy (&uuid);    }    zstr_free (&ipaddress);    zframe_destroy (&frame);    return 0;}

int find_master_input(zloop_t *loop, zmq_pollitem_t *item, void *arg) {    maneater_client *cli = (maneater_client *)arg;    zframe_t *incoming = zframe_recv(cli->local_socket);    unsigned char *data = zframe_data(incoming);    int size = zframe_size(incoming);    msgpack_unpacked msg;    size_t off;    MSG_NEXT(&msg, data, size, &off);    assert(msg.data.type == MSGPACK_OBJECT_POSITIVE_INTEGER);    uint64_t msgid = msg.data.via.u64;    int ret = 0, i = 0;    if (msgid == MID_IS_MASTER) {        MSG_NEXT(&msg, data, size, &off);        assert(msg.data.type == MSGPACK_OBJECT_RAW);        char *host;        COPY_STRING(host, msg.data.via.raw.ptr);        cli->master = host;        cli->master_socket = NULL;        for (i=0; i < cli->num_hosts; i++) {            if (!strcmp(cli->hosts[i], host)) {                cli->master_socket = cli->sockets[i];                break;            }        }        assert(cli->master_socket);        struct timeval tim;        gettimeofday(&tim, NULL);        cli->last_tick = tim.tv_sec;        zclock_log("got master: %s", host);        ret = -1;    }    zframe_destroy(&incoming);    return ret;}

int main (void){  zctx_t *ctx = zctx_new ();  void *dealer = zsocket_new (ctx, ZMQ_DEALER);  zsocket_connect (dealer, "tcp://127.0.0.1:6000");  // We'll allow up to N chunks in transit at once  size_t credit = PIPELINE;  size_t total = 0;     // Total bytes received  size_t chunks = 0;    // Total chunks received  size_t offset = 0;    // Offset of next chunk request  size_t offset_expc = 0;    while (true) {    while (credit) {      // Ask for next chunk      zstr_sendfm (dealer, "fetch");      zstr_sendfm (dealer, "%ld", offset);      zstr_sendf (dealer, "%ld", CHUNK_SIZE);      offset += CHUNK_SIZE;      credit--;    }    zframe_t *chunk = zframe_recv (dealer);    if (!chunk)      break; // Shutting down, quit        chunks++;    credit++;        size_t size = zframe_size (chunk);    zframe_destroy (&chunk);    total += size;    if (size < CHUNK_SIZE)      break; // Last chunk received; exit  }  printf ("%zd chunks received, %zd bytes/n", chunks, total);    zctx_destroy (&ctx);  return total == 102400? 0: -1;}

static voidzyre_node_recv_beacon (zyre_node_t *self){    //  Get IP address and beacon of peer    char *ipaddress = zstr_recv (self->beacon);    zframe_t *frame = zframe_recv (self->beacon);    if (ipaddress == NULL)        return;                 //  Interrupted    //  Ignore anything that isn't a valid beacon    beacon_t beacon;    memset (&beacon, 0, sizeof (beacon_t));    if (zframe_size (frame) == sizeof (beacon_t))        memcpy (&beacon, zframe_data (frame), zframe_size (frame));    zframe_destroy (&frame);    if (beacon.version != BEACON_VERSION)        return;                 //  Garbage beacon, ignore it    zuuid_t *uuid = zuuid_new ();    zuuid_set (uuid, beacon.uuid);    if (beacon.port) {        char endpoint [100];        const char *iface = zsys_interface ();        if (zsys_ipv6 () && iface && !streq (iface, "") && !streq (iface, "*"))            sprintf (endpoint, "tcp://%s%%%s:%d", ipaddress, iface, ntohs (beacon.port));        else            sprintf (endpoint, "tcp://%s:%d", ipaddress, ntohs (beacon.port));        zyre_peer_t *peer = zyre_node_require_peer (self, uuid, endpoint);        zyre_peer_refresh (peer, self->evasive_timeout, self->expired_timeout);    }    else {        //  Zero port means peer is going away; remove it if        //  we had any knowledge of it already        zyre_peer_t *peer = (zyre_peer_t *) zhash_lookup (            self->peers, zuuid_str (uuid));        if (peer)            zyre_node_remove_peer (self, peer);    }    zuuid_destroy (&uuid);    zstr_free (&ipaddress);}

    void heartbeat() {	directoryd::ServiceRequest request;	request.set_type(directoryd::HEARTBEAT);	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().register_socket(), 0);	assert(retval == 0);	zframe_t *rf = zframe_recv (DDClient::instance().register_socket());	directoryd::ServiceReply reply;	reply.ParseFromArray(zframe_data(rf),zframe_size(rf));	zframe_destroy(&rf);	if (reply.type() != directoryd::HEARTBEAT) {	    throw RegistrationError("Got back incorrect message type when trying to send heartbeat.");	}	if (reply.success() != true) {	    throw RegistrationError(reply.result());	}    }

static voidserver_thread (void *args, zctx_t *ctx, void *pipe){    FILE *file = fopen ("testdata", "r");    assert (file);    void *router = zsocket_new (ctx, ZMQ_ROUTER);    //  Default HWM is 1000, which will drop messages here    //  since we send more than 1,000 chunks of test data,    //  so set an infinite HWM as a simple, stupid solution:    zsocket_set_hwm (router, 0);    zsocket_bind (router, "tcp://*:6000");    while (true) {        //  First frame in each message is the sender identity        zframe_t *identity = zframe_recv (router);        if (!identity)            break;              //  Shutting down, quit                    //  Second frame is "fetch" command        char *command = zstr_recv (router);        assert (streq (command, "fetch"));        free (command);        while (true) {            byte *data = malloc (CHUNK_SIZE);            assert (data);            size_t size = fread (data, 1, CHUNK_SIZE, file);            zframe_t *chunk = zframe_new_zero_copy (                data, size, free_chunk, NULL);            zframe_send (&identity, router, ZFRAME_REUSE + ZFRAME_MORE);            zframe_send (&chunk, router, 0);            if (size == 0)                break;          //  Always end with a zero-size frame        }        zframe_destroy (&identity);    }    fclose (file);}

zmsg_t *zmsg_recv (void *source){    assert (source);    zmsg_t *self = zmsg_new ();    if (!self)        return NULL;    while (true) {        zframe_t *frame = zframe_recv (source);        if (!frame) {            zmsg_destroy (&self);            break;              //  Interrupted or terminated        }        if (zmsg_append (self, &frame)) {            zmsg_destroy (&self);            break;        }        if (!zsock_rcvmore (source))            break;              //  Last message frame    }    return self;}

static voidclient_thread (void *args, zctx_t *ctx, void *pipe){    void *dealer = zsocket_new (ctx, ZMQ_DEALER);    zsocket_connect (dealer, "tcp://127.0.0.1:6000");        zstr_send (dealer, "fetch");    size_t total = 0;       //  Total bytes received    size_t chunks = 0;      //  Total chunks received        while (true) {        zframe_t *frame = zframe_recv (dealer);        if (!frame)            break;              //  Shutting down, quit        chunks++;        size_t size = zframe_size (frame);        zframe_destroy (&frame);        total += size;        if (size == 0)            break;              //  Whole file received    }    printf ("%zd chunks received, %zd bytes/n", chunks, total);    zstr_send (pipe, "OK");}

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

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);        }//.........这里部分代码省略.........

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);    assert (address);    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}

intzmailer_msg_recv (zmailer_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 ("zmailer_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 ("zmailer_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 ("zmailer_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 ZMAILER_MSG_MAIL:            {                uint16_t version;                GET_NUMBER2 (version);                if (version != 1) {                    zsys_warning ("zmailer_msg: version is invalid");                    goto malformed;                }            }            GET_STRING (self->from);            GET_LONGSTR (self->to);            GET_LONGSTR (self->subject);            GET_LONGSTR (self->request);            break;        default:            zsys_warning ("zmailer_msg: bad message ID");            goto malformed;    }    //  Successful return    zmq_msg_close (&frame);    return 0;    //  Error returns    malformed:        zsys_warning ("zmailer_msg: zmailer_msg malformed message, fail");        zmq_msg_close (&frame);        return -1;              //  Invalid message}

intzpubsub_filter_recv (zpubsub_filter_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 ("zpubsub_filter: 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 ("zpubsub_filter: 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 | 7)) {        zsys_warning ("zpubsub_filter: 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 ZPUBSUB_FILTER_FILTER:            {                uint16_t magic;                GET_NUMBER2 (magic);                if (magic != ZPUBSUB_FILTER_MAGIC_NUMBER) {                    zsys_warning ("zpubsub_filter: magic is invalid");                    goto malformed;                }            }            {                uint16_t version;                GET_NUMBER2 (version);                if (version != ZPUBSUB_FILTER_VERSION) {                    zsys_warning ("zpubsub_filter: version is invalid");                    goto malformed;                }            }            GET_STRING (self->partition);            GET_STRING (self->topic);            break;        default:            zsys_warning ("zpubsub_filter: bad message ID");            goto malformed;    }    //  Successful return    zmq_msg_close (&frame);    return 0;    //  Error returns    malformed:        zsys_warning ("zpubsub_filter: zpubsub_filter malformed message, fail");        zmq_msg_close (&frame);        return -1;              //  Invalid message}

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