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

static intpipe_attach_remote_writer (pipe_t *self, const char *remote, bool unicast){    assert (self);    if (self->reader == REMOTE_NODE) {        //  We're witnessing two nodes chatting, so we can drop the pipe        //  and forget all about it        pipe_destroy (&self);        return 0;    }    else    if (self->writer == NULL) {        //  This is how we indicate a remote writer        self->writer = REMOTE_NODE;        self->remote = strdup (remote);        zsys_info ("%s: attach remote writer", self->name);        if (self->reader && !unicast) {            //  Tell remote node we're acting as reader, if we got a            //  broadcast message. If we got a unicast message, the peer            //  already knows about us, so don't re-echo the message            zmsg_t *msg = zmsg_new ();            zmsg_addstr (msg, "HAVE READER");            zmsg_addstr (msg, self->name);            zyre_whisper (self->server->zyre, self->remote, &msg);            zsys_info ("%s: tell peer we are now reader", self->name);        }        return 0;    }    zsys_info ("%s: pipe already has writer: ignored", self->name);    return -1;}

static voids_check_directory (s_agent_t *self){    //  Get latest snapshot and build a patches list for any changes    //  All patches are built using a virtual path starting at "/"    zdir_t *dir = zdir_new (self->path, NULL);    zlist_t *patches = zdir_diff (self->dir, dir, "/");    //  Drop old directory and replace with latest version    zdir_destroy (&self->dir);    self->dir = dir;    while (zlist_size (patches)) {        zdir_patch_t *patch = (zdir_patch_t *) zlist_pop (patches);        if (zdir_patch_op (patch) == patch_create) {            //  Shout new files to DROPS group            //  Stupidest possible approach: send whole file as one frame            //  Truncate file at arbitrary limit of 10MB            zfile_t *file = zdir_patch_file (patch);            if (zfile_input (file) == 0) {                zchunk_t *chunk = zfile_read (file, 10 * 1024 * 1024, 0);                assert (chunk);                zmsg_t *msg = zmsg_new ();                zmsg_addstr (msg, "CREATE");                zmsg_addstr (msg, zdir_patch_vpath (patch));                zmsg_add (msg, zframe_new (zchunk_data (chunk), zchunk_size (chunk)));                zchunk_destroy (&chunk);                zyre_shout (self->zyre, "DROPS", &msg);            }        }        zdir_patch_destroy (&patch);    }    zlist_destroy (&patches);}

static ints_zap_request_reply (zap_request_t *self, char *status_code, char *status_text, unsigned char *metadata, size_t metasize){    if (self->verbose)        zsys_info ("zauth: - ZAP reply status_code=%s status_text=%s",                   status_code, status_text);    zmsg_t *msg = zmsg_new ();    int rc = zmsg_addstr(msg, "1.0");    assert (rc == 0);    rc = zmsg_addstr(msg, self->sequence);    assert (rc == 0);    rc = zmsg_addstr(msg, status_code);    assert (rc == 0);    rc = zmsg_addstr(msg, status_text);    assert (rc == 0);    rc = zmsg_addstr(msg, "");    assert (rc == 0);    rc = zmsg_addmem(msg, metadata, metasize);    assert (rc == 0);    rc = zmsg_send(&msg, self->handler);    assert (rc == 0);    return 0;}

intmdp_client_send (mdp_client_t **self_p, void *socket){    assert (socket);    assert (self_p);    assert (*self_p);    mdp_client_t *self = *self_p;    //  If we're sending to a ROUTER, we send the address first    zmsg_t *msg = zmsg_new ();    if (zsocket_type (socket) == ZMQ_ROUTER) {        assert (self->address);        zmsg_add (msg, self->address);        self->address = NULL;       //  Owned by msg now    }    //  Send header fields    zmsg_addstr (msg, "");    zmsg_addstr (msg, "MDPC01");    //  All messages have the same structure    zmsg_addstr (msg, self->service);    zmsg_add (msg, self->body);    self->body = NULL;    //  Send the message and destroy mdp_client object    int rc = zmsg_send (&msg, socket);    mdp_client_destroy (self_p);    return rc;}

static zmsg_t *server_method (server_t *self, const char *method, zmsg_t *msg){    if (streq (method, "CLIENTLIST")) {        zmsg_t *reply = zmsg_new ();        zmsg_addstr (reply, "CLIENTLIST");        void *item = zhashx_first (self->clients);        while (item) {            zmsg_addstr (reply, (const char *) zhashx_cursor (self->clients));            item = (void *) zhashx_next (self->clients);        }        return reply;    }    if (streq (method, "STREAMLIST")) {        zmsg_t *reply = zmsg_new();        zmsg_addstr (reply, "STREAMLIST");        stream_t *stream = (stream_t *) zhashx_first (self->streams);        while (stream) {            zmsg_addstr (reply, stream->name);            stream = (stream_t *) zhashx_next (self->streams);        }        return reply;    }    if (streq (method, "SLOW_TEST_MODE")) {        //  selftest: Tell all stream engines to enable SLOW_TEST_MODE        stream_t *stream = (stream_t *) zhashx_first (self->streams);        while (stream) {            zsock_send (stream->actor, "s", method);            stream = (stream_t *) zhashx_next (self->streams);        }        return NULL;    }    return NULL;}

static voidpipe_drop_local_writer (pipe_t **self_p){    assert (self_p);    if (*self_p) {        pipe_t *self = *self_p;        //  TODO: what if self->writer is REMOTE_NODE?        self->writer = NULL;        if (self->reader) {            if (self->reader == REMOTE_NODE) {                //  Tell remote node we're dropping off                zmsg_t *msg = zmsg_new ();                zmsg_addstr (msg, "DROP WRITER");                zmsg_addstr (msg, self->name);                zyre_whisper (self->server->zyre, self->remote, &msg);                zsys_info ("%s: tell peer we stopped being writer", self->name);            }            else {                engine_send_event (self->reader, writer_dropped_event);                //  Don't destroy pipe yet - reader is still using it                *self_p = NULL;            }        }        pipe_destroy (self_p);    }}

voidztask_job_request_clean (ztask_job_request_t *self, ztask_node_manager_t *node_mgr){    assert (self);//    zclock_log ("Cleaning job request from %s ...", zyre_event_sender (self->request));    zlist_t *keys = zhash_keys (self->processes);    char *key = (char *) zlist_first (keys);    ztask_job_proc_t *p;    int pid;    while (key) {        p = (ztask_job_proc_t *) zhash_lookup (self->processes, key);        zhash_delete (self->processes, key);        pid = ztask_job_proc_pid(p);//        assert (pid);        if (pid) {            zclock_log("Killing pid=%d ...", ztask_job_proc_pid(p));            kill (ztask_job_proc_pid(p), SIGKILL);            zmsg_t *msg_report = zmsg_new ();            zmsg_addstr (msg_report, "REPORT");            zmsg_addstr (msg_report, ztask_job_proc_jobid(p));            zmsg_addstr (msg_report, "-100");            zyre_whisper (ztask_node_manager_zyre_node(node_mgr), zyre_event_sender(self->request), &msg_report);            zmsg_destroy (&msg_report);            zhash_delete (ztask_node_manager_list_running_processes (node_mgr), ztask_job_proc_jobid(p));            zlist_append (ztask_node_manager_list_available_processes (node_mgr), p);            ztask_job_proc_reset(p);        }        key = (char *) zlist_next (keys);    }    zlist_destroy (&keys);}

static voidclient_execute (client_t *self, int event){    self->next_event = event;    while (self->next_event) {        self->event = self->next_event;        self->next_event = 0;        printf ("State=%s, event=%s/n",            s_state_name [self->state], s_event_name [self->event]);        switch (self->state) {            case start_state:                if (self->event == ohai_event) {                check_credentials_action (self);                    self->state = authenticated_state;                }                break;                            case authenticated_state:                if (self->event == ok_event) {                    zmsg_addstr (self->reply, "OHAI-OK");                    self->state = ready_state;                }                else                if (self->event == error_event) {                    zmsg_addstr (self->reply, "WTF");                    self->state = start_state;                }                break;                            case ready_state:                if (self->event == icanhaz_event) {                    zmsg_addstr (self->reply, "CHEEZBURGER");                }                else                if (self->event == hugz_event) {                    zmsg_addstr (self->reply, "HUGZ-OK");                }                else                if (self->event == heartbeat_event) {                    zmsg_addstr (self->reply, "HUGZ");                }                break;                            case stopped_state:                //  Discard all events silently                break;        }        if (zmsg_size (self->reply) > 1) {            puts ("Send message to client");            zmsg_dump (self->reply);            zmsg_send (&self->reply, self->router);            self->reply = zmsg_new ();            zmsg_add (self->reply, zframe_dup (self->address));        }    }}

int timer_event(zloop_t *loop, int timer_id, void *pusher){  int rc;  static int message_count = 0;  zmsg_t *message = zmsg_new();  zmsg_addstr(message, "request-stream-test-development");  zmsg_addstr(message, "routing-key");  zmsg_addstrf(message, "message-body %d", message_count++);  rc = zmsg_send(&message, pusher);  if (!zsys_interrupted) {    assert(rc==0);  }  return 0;}

int main (int argc, char *argv []){    zre_interface_t *interface = zre_interface_new (false);    zre_interface_join (interface, "GLOBAL");    while (true) {        zmsg_t *incoming = zre_interface_recv (interface);        if (!incoming)            break;              //  Interrupted        //  If new peer, say hello to it and wait for it to answer us        char *event = zmsg_popstr (incoming);        if (streq (event, "ENTER")) {            char *peer = zmsg_popstr (incoming);            printf ("I: [%s] peer entered/n", peer);            zmsg_t *outgoing = zmsg_new ();            zmsg_addstr (outgoing, peer);            zmsg_addstr (outgoing, "Hello");            zre_interface_whisper (interface, &outgoing);            free (peer);        }        else if (streq (event, "EXIT")) {            char *peer = zmsg_popstr (incoming);            printf ("I: [%s] peer exited/n", peer);            free (peer);        }        else if (streq (event, "WHISPER")) {            char *peer = zmsg_popstr (incoming);            printf ("I: [%s] received ping (WHISPER)/n", peer);            free (peer);            zmsg_t *outgoing = zmsg_new ();            zmsg_addstr (outgoing, "GLOBAL");            zmsg_addstr (outgoing, "Hello");            zre_interface_shout (interface, &outgoing);        }        else if (streq (event, "SHOUT")) {            char *peer = zmsg_popstr (incoming);            char *group = zmsg_popstr (incoming);            printf ("I: [%s](%s) received ping (SHOUT)/n", peer, group);            free (peer);            free (group);        }        free (event);        zmsg_destroy (&incoming);    }    zre_interface_destroy (&interface);    return 0;}

int main (int argc, char *argv []){    int verbose = (argc > 1 && streq (argv [1], "-v"));    mdcli_t *session = mdcli_new ("tcp://localhost:5555", verbose);    //  1. Send 'echo' request to Titanic    zmsg_t *request = zmsg_new ();    zmsg_addstr (request, "echo");    zmsg_addstr (request, "Hello world");    zmsg_t *reply = s_service_call (        session, "titanic.request", &request);    zframe_t *uuid = NULL;    if (reply) {        uuid = zmsg_pop (reply);        zmsg_destroy (&reply);        zframe_print (uuid, "I: request UUID ");    }    //  2. Wait until we get a reply    while (!zctx_interrupted) {        zclock_sleep (100);        request = zmsg_new ();        zmsg_add (request, zframe_dup (uuid));        zmsg_t *reply = s_service_call (            session, "titanic.reply", &request);        if (reply) {            char *reply_string = zframe_strdup (zmsg_last (reply));            printf ("Reply: %s/n", reply_string);            free (reply_string);            zmsg_destroy (&reply);            //  3. Close request            request = zmsg_new ();            zmsg_add (request, zframe_dup (uuid));            reply = s_service_call (session, "titanic.close", &request);            zmsg_destroy (&reply);            break;        }        else {            printf ("I: no reply yet, trying again.../n");            zclock_sleep (5000);     //  Try again in 5 seconds        }    }    zframe_destroy (&uuid);    mdcli_destroy (&session);    return 0;}

static voids_pub_metric(mlm_client_t * mlm, const char *key, const char *dev_name){    char topic[100], value[100];    snprintf (topic, 100, "%s.%s", key, dev_name );    zmsg_t *msg = zmsg_new ();    zmsg_addstr (msg, dev_name);    zmsg_addstr (msg, key);    snprintf (value, 100, "%li", random() % 16 + 10 );    zmsg_addstr (msg, value);    mlm_client_send (mlm, topic, &msg);    printf ("sending %s/%s/%s/n", dev_name, key, value);}

zmsg_t* kosmonaut_request(kosmonaut_t* self, char* request){	if (!request || !self)		return NULL;	zmq_pollitem_t items[] = {{self->req, 0, ZMQ_POLLOUT|ZMQ_POLLIN, 0}};	zmsg_t* res = NULL;	zmsg_t* req = NULL;	pthread_mutex_lock(&self->tmtx);	int rc = zmq_poll(items, 1, REQUEST_TIMEOUT * ZMQ_POLL_MSEC);	if (rc == 0 && items[0].revents & ZMQ_POLLOUT) {		req = zmsg_new();		zmsg_addstr(req, request);		zmsg_send(&req, self->req);		zmsg_destroy(&req);			rc = zmq_poll(items, 1, RESPONSE_TIMEOUT * ZMQ_POLL_MSEC);		if (rc == 0 && items[0].revents & ZMQ_POLLIN) {			res = zmsg_recv(self->req);		}	}  	pthread_mutex_lock(&self->tmtx);	return res;}

void graph_response_newNodeResponse(json_t * request, json_t * response,				    int32_t requestId, void *spss,				    req_store_t * req_store){	json_t *pss_request = json_object();	json_object_set_new(pss_request, "type", json_string("newNodeRequest"));	json_t *node = json_object_get(request,				       "node");	//set the id of the node	json_object_set(json_object_get(node, "node"),			"id", json_object_get(response, "id"));	json_object_set(node, "id", json_object_get(response, "id"));	json_t *pnode = json_object();	json_object_set(pnode, "posX", json_object_get(node, "posX"));	json_object_set(pnode, "posY", json_object_get(node, "posY"));	json_object_set(pnode, "id", json_object_get(response, "id"));	json_object_set(pss_request, "node", pnode);	json_t *pss_req = json_object();	json_object_set_new(pss_req, "requestId", json_integer(requestId));	json_object_set_new(pss_req, "request", pss_request);	zmsg_t *req = zmsg_new();	char *pss_req_str = json_dumps(pss_req,				       JSON_COMPACT);	printf("/nbroker:spss sent: %s/n", pss_req_str);	zmsg_addstr(req, pss_req_str);	free(pss_req_str);	zmsg_send(&req, spss);	json_decref(pss_req);}

int main (void){    //  Create new freelance client object    flcliapi_t *client = flcliapi_new ();    //  Connect to several endpoints    flcliapi_connect (client, "tcp://localhost:5555");    flcliapi_connect (client, "tcp://localhost:5556");    flcliapi_connect (client, "tcp://localhost:5557");    //  Send a bunch of name resolution 'requests', measure time    int requests = 1000;    uint64_t start = zclock_time ();    while (requests--) {        zmsg_t *request = zmsg_new ();        zmsg_addstr (request, "random name");        zmsg_t *reply = flcliapi_request (client, &request);        if (!reply) {            printf ("E: name service not available, aborting/n");            break;        }        zmsg_destroy (&reply);    }    printf ("Average round trip cost: %d usec/n",        (int) (zclock_time () - start) / 10);    flcliapi_destroy (&client);    return 0;}

static zmsg_t *server_method (server_t *self, const char *method, zmsg_t *msg){    //  Connect to a remote    zmsg_t *reply = NULL;    if (streq (method, "CONNECT")) {        char *endpoint = zmsg_popstr (msg);        assert (endpoint);        server_connect (self, endpoint);        zstr_free (&endpoint);    }    else    if (streq (method, "PUBLISH")) {        char *key = zmsg_popstr (msg);        char *value = zmsg_popstr (msg);        server_accept (self, key, value);        zstr_free (&key);        zstr_free (&value);    }    else    if (streq (method, "STATUS")) {        //  Return number of tuples we have stored        reply = zmsg_new ();        assert (reply);        zmsg_addstr (reply, "STATUS");        zmsg_addstrf (reply, "%d", (int) zhashx_size (self->tuples));    }    else        zsys_error ("unknown zgossip method '%s'", method);    return reply;}

zmsg_t *create_action_message(const char *action){    zmsg_t *msg = create_base_message(MSGTYPE_ACTION);    zmsg_addstr(msg, action);    return msg;}

intmdp_worker_send (mdp_worker_t **self_p, void *socket){    assert (socket);    assert (self_p);    assert (*self_p);    mdp_worker_t *self = *self_p;    //  If we're sending to a ROUTER, we send the address first    zmsg_t *msg = zmsg_new ();    if (zsockopt_type (socket) == ZMQ_ROUTER) {        assert (self->address);        zmsg_add (msg, self->address);        self->address = NULL;       //  Owned by msg now    }    //  Send header fields    zmsg_addstr (msg, "");    zmsg_addstr (msg, "MDPW01");    zmsg_addmem (msg, &self->id, 1);    switch (self->id) {        case MDP_WORKER_READY:            zmsg_addstr (msg, self->service);            break;        case MDP_WORKER_REQUEST:            zmsg_add (msg, self->client);            self->client = NULL;            zmsg_add (msg, self->body);            self->body = NULL;            break;        case MDP_WORKER_REPLY:            zmsg_add (msg, self->client);            self->client = NULL;            zmsg_add (msg, self->body);            self->body = NULL;            break;        case MDP_WORKER_HEARBEAT:            break;        case MDP_WORKER_DISCONNECT:            break;    }    //  Send the message and destroy mdp_worker object    int rc = zmsg_send (&msg, socket);    mdp_worker_destroy (self_p);    return rc;}

int main (int argc, char *argv []){    int verbose = (argc > 1 && streq (argv [1], "-v"));    zctx_t *ctx = zctx_new ();    //  Prepare server socket with predictable identity    char *bind_endpoint = "tcp://*:5555";    char *connect_endpoint = "tcp://localhost:5555";    void *server = zsocket_new (ctx, ZMQ_ROUTER);    zmq_setsockopt (server,        ZMQ_IDENTITY, connect_endpoint, strlen (connect_endpoint));    zsocket_bind (server, bind_endpoint);    printf ("I: service is ready at %s/n", bind_endpoint);    while (!zctx_interrupted) {        zmsg_t *request = zmsg_recv (server);        if (verbose && request)            zmsg_dump (request);        if (!request)            break;          //  Interrupted        //  Frame 0: identity of client        //  Frame 1: PING, or client control frame        //  Frame 2: request body        zframe_t *identity = zmsg_pop (request);        zframe_t *control = zmsg_pop (request);        zmsg_t *reply = zmsg_new ();        if (zframe_streq (control, "PING"))            zmsg_addstr (reply, "PONG");        else {            zmsg_add (reply, control);            zmsg_addstr (reply, "OK");        }        zmsg_destroy (&request);        zmsg_push (reply, identity);        if (verbose && reply)            zmsg_dump (reply);        zmsg_send (&reply, server);    }    if (zctx_interrupted)        printf ("W: interrupted/n");    zctx_destroy (&ctx);    return 0;}

voidprob_terminate(prob_client_t pc){    zmsg_t *request = zmsg_new();    zmsg_addstr(request, "terminate");    zmsg_addstrf(request, "%d", pc->id_count);    zmsg_send(&request, pc->zocket);    zmsg_t *response = zmsg_recv(pc->zocket);    zmsg_destroy(&response);    zmsg_destroy(&request);}

int main(int argc, char **argv){		zctx_t *ctx;	zwssock_t *sock;	char *l =  argc > 1 ? argv[1] : listen_on;	int major, minor, patch;	zmq_version (&major, &minor, &patch);	printf("built with: 
C++ zmsg_destroy函数代码示例
C++ zmq_version函数代码示例