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

static agent_t *s_agent_new (zctx_t *ctx, void *pipe){    agent_t *self = (agent_t *) zmalloc (sizeof (agent_t));    if (!self)        return NULL;    self->ctx = ctx;    self->pipe = pipe;    self->whitelist = zhash_new ();    if (self->whitelist)        self->blacklist = zhash_new ();    //  Create ZAP handler and get ready for requests    if (self->blacklist)        self->handler = zsocket_new (self->ctx, ZMQ_REP);    if (self->handler) {        if (zsocket_bind (self->handler, ZAP_ENDPOINT) == 0)            zstr_send (self->pipe, "OK");        else            zstr_send (self->pipe, "ERROR");    }    else        s_agent_destroy (&self);    return self;}

int main (void){    puts ("Starting the zchat server");    zsock_t *responder = zsock_new_rep ("tcp://*:5555");    assert (responder);    zsock_t *publisher = zsock_new_pub ("tcp://*:5556");    assert (publisher);    while (!zsys_interrupted) {        //  Receive messages from client.        char *client_msg = zstr_recv (responder);        if (!client_msg)            break;                //  Let the zchat client know we got it.        zstr_send (responder, "OK");        //  Publish message to all zchat client subscribers        zstr_send (publisher, client_msg);        zstr_free (&client_msg);    }    puts ("Stopping the zchat server");    zsock_destroy (&publisher);    zsock_destroy (&responder);    return 0;}

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

int main (int argc, char *argv []){    zctx_t *context = zctx_new ();    void *publisher = zsocket_new (context, ZMQ_PUB);    if (argc == 2)        zsocket_connect (publisher, argv [1]);    else        zsocket_bind (publisher, "tcp://*:5556");    //  Ensure subscriber connection has time to complete    sleep (1);    //  Send out all 1,000 topic messages    int topic_nbr;    for (topic_nbr = 0; topic_nbr < 1000; topic_nbr++) {        zstr_sendm (publisher, "%03d", topic_nbr, ZMQ_SNDMORE);        zstr_send (publisher, "Save Roger");    }    //  Send one random update per second    srandom ((unsigned) time (NULL));    while (!zctx_interrupted) {        sleep (1);        zstr_sendm (publisher, "%03d", randof (1000), ZMQ_SNDMORE);        zstr_send (publisher, "Off with his head!");    }    zctx_destroy (&context);    return 0;}

voidzproxy_test (bool verbose){    printf (" * zproxy: ");    //  @selftest    zctx_t *ctx = zctx_new ();    void *frontend = zsocket_new (ctx, ZMQ_PULL);    int rc = zsocket_bind (frontend, "inproc://frontend");    assert (rc == 0);    void *backend = zsocket_new (ctx, ZMQ_PUSH);    rc = zsocket_bind (backend, "inproc://backend");    assert (rc == 0);    zproxy_t *proxy = zproxy_new (ctx, frontend, backend);    //  Connect application sockets to proxy    void *faucet = zsocket_new (ctx, ZMQ_PUSH);    rc = zsocket_connect (faucet, "inproc://frontend");    assert (rc == 0);    void *sink = zsocket_new (ctx, ZMQ_PULL);    rc = zsocket_connect (sink, "inproc://backend");    assert (rc == 0);    //  Send some messages and check they arrived    zstr_send (faucet, "Hello");    char *string = zstr_recv (sink);    assert (streq (string, "Hello"));    zstr_free (&string);        //  Check pause/resume functionality    zproxy_pause (proxy);    zstr_send (faucet, "World");    zproxy_resume (proxy);    string = zstr_recv (sink);    assert (streq (string, "World"));    zstr_free (&string);        //  Create capture socket, must be a PULL socket    void *capture = zsocket_new (ctx, ZMQ_PULL);    rc = zsocket_bind (capture, "inproc://capture");    assert (rc == 0);    //  Switch on capturing, check that it works    zproxy_capture (proxy, "inproc://capture");    zstr_send (faucet, "Hello");        string = zstr_recv (sink);    assert (streq (string, "Hello"));    zstr_free (&string);        string = zstr_recv (capture);    assert (streq (string, "Hello"));    zstr_free (&string);    zproxy_destroy (&proxy);    zctx_destroy (&ctx);    //  @end    printf ("OK/n");}

int main (void){    zvudp_t *zvudp = zvudp_new ();    void *client = zvudp_socket (zvudp);    zvudp_connect (zvudp, "127.0.0.1", 31000);    //  Send test set to server    puts ("Sending test set...");    zstr_send (client, "START");    int count;    for (count = 0; count < 1000000; count++) {        if (zctx_interrupted)            break;        zstr_send (client, "This is a test");    }    zstr_send (client, "END");    //  Wait for server to confirm    puts ("Waiting for server...");    char *input = zstr_recv (client);    if (input) {        puts (input);        free (input);    }    zvudp_destroy (&zvudp);    return 0;}

static voidtest_tcp_pair_cli (void *args, zctx_t *ctx, void *pipe){    vtx_t *vtx = vtx_new (ctx);    int rc = vtx_tcp_load (vtx, FALSE);    assert (rc == 0);    char *port = zstr_recv (pipe);    void *pair = vtx_socket (vtx, ZMQ_PAIR);    assert (pair);    rc = vtx_connect (vtx, pair, "tcp://localhost:%s", port);    assert (rc == 0);    int sent = 0;    int recd = 0;    while (!zctx_interrupted) {        zstr_send (pair, "ICANHAZ?");        sent++;        char *reply = zstr_recv_nowait (pair);        if (reply) {            recd++;            free (reply);        }        char *end = zstr_recv_nowait (pipe);        if (end) {            free (end);            zstr_send (pipe, "OK");            break;        }    }    zclock_log ("I: PAIR CLI: sent=%d recd=%d", sent, recd);    free (port);    vtx_destroy (&vtx);}

static voidclient_task (void *args, zctx_t *ctx, void *pipe){    void *client = zsocket_new (ctx, ZMQ_DEALER);    zsocket_connect (client, "tcp://localhost:5555");    printf ("Setting up test.../n");    zclock_sleep (100);    int requests;    int64_t start;    printf ("Synchronous round-trip test.../n");    start = zclock_time ();    for (requests = 0; requests < 10000; requests++) {        zstr_send (client, "hello");        char *reply = zstr_recv (client);        free (reply);    }    printf (" %d calls/second/n",        (1000 * 10000) / (int) (zclock_time () - start));    printf ("Asynchronous round-trip test.../n");    start = zclock_time ();    for (requests = 0; requests < 100000; requests++)        zstr_send (client, "hello");    for (requests = 0; requests < 100000; requests++) {        char *reply = zstr_recv (client);        free (reply);    }    printf (" %d calls/second/n",        (1000 * 100000) / (int) (zclock_time () - start));    zstr_send (pipe, "done");}

int main(int argc, char** argv) {    zsock_t * sc = zsock_new(ZMQ_PUB);    zsock_connect(sc, "tcp://%s:5560", argv[1]);    while(!zsys_interrupted) {        if(random()%1) {            zstr_send(sc, "ON");        } else {            zstr_send(sc, "OFF");        }        sleep(1);    }    zsock_destroy(&sc);}

static voidtarget_task (void *args, zctx_t *ctx, void *pipe){    void *subscriber = zsocket_new (ctx, ZMQ_DEALER);    zsocket_connect (subscriber, "tcp://localhost:6001");    zstr_send (subscriber, "Hello");    char *question = zstr_recv (subscriber);    char *answer = randof (2) == 0? "Yes": "No";    printf ("%s %s/n", question, answer);    free (question);    zstr_send (subscriber, answer);}

staticint send_tick_commands(zloop_t *loop, int timer_id, void *arg){    controller_state_t *state = arg;    // send tick commands to actors to let them print out their stats    zstr_send(state->subscriber, "tick");    zstr_send(state->writer, "tick");    int rc = zloop_timer(loop, 1000, 1, send_tick_commands, state);    assert(rc != -1);    return 0;}

void zsock_tx_cb (struct ev_loop *loop, ev_zmq *w, int revents){    static int count = 50; /* send two per invocation */    if ((revents & EV_WRITE)) {        if (zstr_send (w->zsock, "PING") < 0)            fprintf (stderr, "zstr_send: %s", strerror (errno));        if (zstr_send (w->zsock, "PING") < 0)            fprintf (stderr, "zstr_send: %s", strerror (errno));        if (--count == 0)            ev_zmq_stop (loop, w);    }    if ((revents & EV_ERROR))        ev_break (loop, EVBREAK_ALL);}

static intagent_ping_peer (const char *key, void *item, void *argument){    agent_t *self = (agent_t *) argument;    zre_peer_t *peer = (zre_peer_t *) item;    char *identity = zre_peer_identity (peer);    if (zclock_time () >= zre_peer_expired_at (peer)) {        zre_log_info (self->log, ZRE_LOG_MSG_EVENT_EXIT,                      zre_peer_endpoint (peer),                      zre_peer_endpoint (peer));        //  If peer has really vanished, expire it        zstr_sendm (self->pipe, "EXIT");        zstr_send (self->pipe, identity);        zhash_foreach (self->peer_groups, agent_peer_delete, peer);        zhash_delete (self->peers, identity);    }    else    if (zclock_time () >= zre_peer_evasive_at (peer)) {        //  If peer is being evasive, force a TCP ping.        //  TODO: do this only once for a peer in this state;        //  it would be nicer to use a proper state machine        //  for peer management.        zre_msg_t *msg = zre_msg_new (ZRE_MSG_PING);        zre_peer_send (peer, &msg);    }    return 0;}

voidzproxy_resume (zproxy_t *self){    assert (self);    zstr_send (self->pipe, "RESUME");    zsocket_wait (self->pipe);}

voidzproxy_pause (zproxy_t *self){    assert (self);    zstr_send (self->pipe, "PAUSE");    zsocket_wait (self->pipe);}

static void *client_task (void *args){    zctx_t *ctx = zctx_new ();    void *client = zsocket_new (ctx, ZMQ_DEALER);    //  Set random identity to make tracing easier    char identity [10];    sprintf (identity, "%04X-%04X", randof (0x10000), randof (0x10000));    zsocket_set_identity (client, identity);    zsocket_connect (client, "tcp://localhost:5570");    zmq_pollitem_t items [] = { { client, 0, ZMQ_POLLIN, 0 } };    int request_nbr = 0;    while (true) {        //  Tick once per second, pulling in arriving messages        int centitick;        for (centitick = 0; centitick < 100; centitick++) {            zmq_poll (items, 1, 10 * ZMQ_POLL_MSEC);            if (items [0].revents & ZMQ_POLLIN) {                zmsg_t *msg = zmsg_recv (client);                zframe_print (zmsg_last (msg), identity);                zmsg_destroy (&msg);            }        }        zstr_send (client, "request #%d");    }    zctx_destroy (&ctx);    return NULL;}

int main(){    czmqpp::context ctx;    assert(ctx.self());    //  Create a few sockets    czmqpp::socket vent(ctx, ZMQ_PUSH);    int rc = vent.bind("tcp://*:9000");    assert(rc != -1);    czmqpp::socket sink(ctx, ZMQ_PULL);    rc = sink.connect("tcp://localhost:9000");    assert(rc != -1);    czmqpp::socket bowl(ctx, ZMQ_PULL);    czmqpp::socket dish(ctx, ZMQ_PULL);    //  Set-up poller    czmqpp::poller poller(bowl, sink, dish);    assert(poller.self());    zstr_send(vent.self(), "Hello, World");    //  We expect a message only on the sink    czmqpp::socket which = poller.wait(-1);    assert(which == sink);    assert(poller.expired() == false);    assert(poller.terminated() == false);    char *message = zstr_recv(which.self());    assert(streq(message, "Hello, World"));    free(message);    return 0;}

intzthread_test (Bool verbose){    printf (" * zthread: ");    //  @selftest    zctx_t *ctx = zctx_new ();    //  Create a detached thread, let it run    zthread_new (ctx, s_test_detached, NULL);    zclock_sleep (100);    //  Create an attached thread, check it's safely alive    void *pipe = zthread_fork (ctx, s_test_attached, NULL);    zstr_send (pipe, "ping");    char *pong = zstr_recv (pipe);    assert (streq (pong, "pong"));    free (pong);    //  Everything should be cleanly closed now    zctx_destroy (&ctx);    //  @end    printf ("OK/n");    return 0;}

static void* client_routine(void* arg){  zctx_t* ctx = zctx_new();  void* client = zsocket_new(ctx, ZMQ_REQ);  zsocket_connect(client, "ipc://frontend.ipc");  while (1) {    char* reply;    zstr_send(client, "Hello");    reply = zstr_recv(client);    if (NULL == reply)      break;    fprintf(stdout, "client: %s/n", reply);    free(reply);    sleep(1);  }  zctx_destroy(&ctx);  return NULL;}

static intzyre_node_ping_peer (const char *key, void *item, void *argument){    zyre_peer_t *peer = (zyre_peer_t *) item;    zyre_node_t *self = (zyre_node_t *) argument;    if (zclock_mono () >= zyre_peer_expired_at (peer)) {        if (self->verbose)            zsys_info ("(%s) peer expired name=%s endpoint=%s",                self->name, zyre_peer_name (peer), zyre_peer_endpoint (peer));        zyre_node_remove_peer (self, peer);    }    else    if (zclock_mono () >= zyre_peer_evasive_at (peer)) {        //  If peer is being evasive, force a TCP ping.        //  TODO: do this only once for a peer in this state;        //  it would be nicer to use a proper state machine        //  for peer management.        if (self->verbose)            zsys_info ("(%s) peer seems dead/slow name=%s endpoint=%s",                self->name, zyre_peer_name (peer), zyre_peer_endpoint (peer));        zre_msg_t *msg = zre_msg_new (ZRE_MSG_PING);        zyre_peer_send (peer, &msg);        // Inform the calling application this peer is being evasive    	zstr_sendm (self->outbox, "EVASIVE");	zstr_sendm (self->outbox, zyre_peer_identity (peer));    	zstr_send (self->outbox, zyre_peer_name (peer));    }    return 0;}

static intzyre_node_stop (zyre_node_t *self){    if (self->beacon) {        //  Stop broadcast/listen beacon        beacon_t beacon;        beacon.protocol [0] = 'Z';        beacon.protocol [1] = 'R';        beacon.protocol [2] = 'E';        beacon.version = BEACON_VERSION;        beacon.port = 0;            //  Zero means we're stopping        zuuid_export (self->uuid, beacon.uuid);        zsock_send (self->beacon, "sbi", "PUBLISH",            (byte *) &beacon, sizeof (beacon_t), self->interval);        zclock_sleep (1);           //  Allow 1 msec for beacon to go out        zpoller_remove (self->poller, self->beacon);        zactor_destroy (&self->beacon);    }    //  Stop polling on inbox    zpoller_remove (self->poller, self->inbox);    zstr_sendm (self->outbox, "STOP");    zstr_sendm (self->outbox, zuuid_str (self->uuid));    zstr_send (self->outbox, self->name);    return 0;}

// Basic request-reply client using REQ socket//static void *client_task(void *args){	zctx_t *ctx = zctx_new();	void *client = zsocket_new(ctx, ZMQ_REQ);#if (defined (WIN32))	zsocket_connect(client, "tcp://localhost:5672"); // frontend#else	zsocket_connect(client, "ipc://frontend.ipc");#endif	// Send request, get reply	while (1) {		zstr_send(client, "HELLO");		char *reply = zstr_recv(client);		if (!reply) {			break;		}		printf("Client: %s/n", reply);		free(reply);		zclock_sleep(1);	}	zctx_destroy(&ctx);	return NULL;}

voidcurve_client_set_verbose (curve_client_t *self, bool verbose){    assert (self);    zstr_sendm (self->control, "VERBOSE");    zstr_send  (self->control, "%d", verbose);}

//  Checks whether client can connect to serverstatic bools_can_connect (zctx_t *ctx, void **server, void **client){    int port_nbr = zsocket_bind (*server, "tcp://127.0.0.1:*");    assert (port_nbr > 0);    int rc = zsocket_connect (*client, "tcp://127.0.0.1:%d", port_nbr);    assert (rc == 0);    //  Give the connection time to fail if that's the plan    zclock_sleep (200);    //  By default PUSH sockets block if there's no peer    zsock_set_sndtimeo (*server, 200);    zstr_send (*server, "Hello, World");    zpoller_t *poller = zpoller_new (*client, NULL);    bool success = (zpoller_wait (poller, 400) == *client);    zpoller_destroy (&poller);    zsocket_destroy (ctx, *client);    zsocket_destroy (ctx, *server);    *server = zsocket_new (ctx, ZMQ_PUSH);    assert (*server);    *client = zsocket_new (ctx, ZMQ_PULL);    assert (*client);    return success;}

static voidtest_tcp_pub (void *args, zctx_t *ctx, void *pipe){    vtx_t *vtx = vtx_new (ctx);    int rc = vtx_tcp_load (vtx, FALSE);    assert (rc == 0);    char *port = zstr_recv (pipe);    //  Create publisher socket and bind to all network interfaces    void *publisher = vtx_socket (vtx, ZMQ_PUB);    assert (publisher);    rc = vtx_bind (vtx, publisher, "tcp://*:%s", port);    assert (rc == 0);    int sent = 0;    while (!zctx_interrupted) {        zstr_sendf (publisher, "NOM %04x", randof (0x10000));        sent++;        char *end = zstr_recv_nowait (pipe);        if (end) {            free (end);            zstr_send (pipe, "OK");            break;        }    }    zclock_log ("I: PUB: sent=%d", sent);    free (port);    vtx_destroy (&vtx);}

voidzauth_set_verbose (zauth_t *self, bool verbose){    assert (self);    zstr_sendm (self->pipe, "VERBOSE");    zstr_send  (self->pipe, "%d", verbose);}

static bools_can_connect (void *server, void *client){    //  We'll do each test on a new port number since otherwise we have to    //  destroy and recreate the sockets each time.    static int port_nbr = 9000;    int rc = zsocket_bind (server, "tcp://*:%d", port_nbr);    assert (rc == port_nbr);    rc = zsocket_connect (client, "tcp://localhost:%d", port_nbr);    assert (rc == 0);        zpoller_t *poller = zpoller_new (client, NULL);    zstr_send (server, "Hello, World");    //  Need up to half a second if running under Valgrind    bool success = zpoller_wait (poller, 500) == client;    if (success)        free (zstr_recv (client));    zpoller_destroy (&poller);    rc = zsocket_unbind (server, "tcp://*:%d", port_nbr);    assert (rc != -1);    rc = zsocket_disconnect (client, "tcp://localhost:%d", port_nbr);    assert (rc != -1);    port_nbr++;    return success;}

 virtual void run() {     char* msg = zstr_recv(socket_);     std::cout << id_ << ": Streaming data for member TODO." << std::endl;     zstr_send(backendSocket_, msg);     free(msg); }

voidfmq_server_setoption (fmq_server_t *self, const char *path, const char *value){    zstr_sendm (self->pipe, "SETOPTION");    zstr_sendm (self->pipe, path);    zstr_send  (self->pipe, value);}

void MessageProcessor::processSocket(){    char* msg = zstr_recv(zmqSocket_);    if (msg == nullptr) {        Log::Error("zeromq recv() failed/n");        return;    }    std::string requestString(msg);    zstr_free(&msg);    std::string responseString;    bool error = processMessage(requestString, responseString);    if (error) {        responseString = "";    }    int rc = zstr_send(zmqSocket_, responseString.c_str());    if (rc != 0) {        Log::vError("MessageProcessor: failed to send response/n"                    "request:/n%s/n/n"                    "response:/n%s/n/n",                    requestString.c_str(), responseString.c_str());    }}