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

static intzyre_handler (zloop_t *loop, zsock_t *reader, void *argument){    server_t *self = (server_t *) argument;    zmsg_t *msg = zyre_recv (self->zyre);    if (!msg)        return -1;              //  Interrupted    char *command = zmsg_popstr (msg);    char *peer_id = zmsg_popstr (msg);    char *peer_name = zmsg_popstr (msg);    if (streq (command, "ENTER"))        zsys_info ("ZPIPES server appeared at %s", peer_name);    else    if (streq (command, "EXIT"))        zsys_info ("ZPIPES server vanished from %s", peer_name);    else    if (streq (command, "SHOUT")) {        char *group = zmsg_popstr (msg);        if (streq (group, "ZPIPES"))            server_process_cluster_command (self, peer_id, peer_name, msg, false);        zstr_free (&group);    }    else    if (streq (command, "WHISPER"))        server_process_cluster_command (self, peer_id, peer_name, msg, true);        zstr_free (&command);    zstr_free (&peer_id);    zstr_free (&peer_name);    zmsg_destroy (&msg);        return 0;}

static bools_node_recv (zre_node_t *node, char* command, char* expected){    bool result = false;    zmsg_t *incoming = zre_node_recv (node);    assert (incoming);    char *event = zmsg_popstr (incoming);    if (streq (event, command)) {        char *peer = zmsg_popstr (incoming);        char *group = NULL;        if (streq (command, "SHOUT"))            group = zmsg_popstr (incoming);        char *cookie = zmsg_popstr (incoming);        if (streq (cookie, expected)) {            result = true;        }        free (peer);        if (group)            free (group);        free (cookie);    }    free (event);    zmsg_destroy (&incoming);    return result;}

/** *  * @param foundId * @param foundReply * @return  */bool BoomStick::ReadFromReadySocket(std::string& foundId, std::string& foundReply) {   if (0 == mUtilizedThread) {      mUtilizedThread = pthread_self();   } else {      CHECK(pthread_self() == mUtilizedThread);   }   if (!mChamber) {      LOG(WARNING) << "Invalid socket";      return false;   }   bool success = false;   zmsg_t* msg = zmsg_recv(mChamber);   if (!msg) {      foundReply = zmq_strerror(zmq_errno());   } else if (zmsg_size(msg) == 2) {      char* msgChar;      msgChar = zmsg_popstr(msg);      foundId = msgChar;      free(msgChar);      msgChar = zmsg_popstr(msg);      foundReply = msgChar;      free(msgChar);      success = true;   } else {      foundReply = "Malformed reply, expecting 2 parts";   }   if (msg) {      zmsg_destroy(&msg);   }   return success;}

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

static ints_recv_from_zyre (s_agent_t *self){    zyre_event_t *event = zyre_event_new (self->zyre);    if (zyre_event_type (event) == ZYRE_EVENT_SHOUT    && streq (zyre_event_group (event), "DROPS")) {        zmsg_t *msg = zyre_event_msg (event);        char *operation = zmsg_popstr (msg);                if (streq (operation, "CREATE")) {            char *filename = zmsg_popstr (msg);            zframe_t *frame = zmsg_pop (msg);            zfile_t *file = zfile_new (self->path, filename);            zfile_output (file);            fwrite (zframe_data (frame), 1, zframe_size (frame), zfile_handle (file));            zfile_destroy (&file);            zframe_destroy (&frame);            zstr_send (self->pipe, filename);            free (filename);        }        free (operation);    }    zyre_event_destroy (&event);    return 0;}

int get_request(zsock_t *sock, struct request *req) {    zmsg_t *msg;    int size;    char *str;    int Y, m, d, H, M, S;    struct tm tm;    msg = zmsg_recv(sock);    size = zmsg_size(msg);    if (size != 2) {        // something's wrong        zmsg_destroy(&msg);        return -1;    }    str = zmsg_popstr(msg);    req->ip = inet_addr(str);    str = zmsg_popstr(msg);    sscanf(str, "%04d%02d%02d%02d%02d%02d", &Y, &m, &d, &H, &M, &S);    tm.tm_year = Y - 1900;    tm.tm_mon = m;    tm.tm_mday = d;    tm.tm_hour = H;    tm.tm_min = M;    tm.tm_sec = S;    req->timestamp = mktime(&tm);    zmsg_destroy(&msg);    return 0;}

static voidserver_process_cluster_command (    server_t *self,    const char *peer_id,    const char *peer_name,    zmsg_t *msg,    bool unicast){    char *request = zmsg_popstr (msg);    char *pipename = zmsg_popstr (msg);    zsys_info ("peer=%s command=%s pipe=%s unicast=%d",                peer_name, request, pipename? pipename: "-", unicast);    //  Lookup or create pipe    //  TODO: remote pipes need cleaning up with some timeout    pipe_t *pipe = NULL;    if (pipename) {        pipe = (pipe_t *) zhash_lookup (self->pipes, pipename);        if (!pipe)            pipe = pipe_new (self, pipename);    }    if (pipe && streq (request, "HAVE WRITER"))        pipe_attach_remote_writer (pipe, peer_id, unicast);    else    if (pipe && streq (request, "HAVE READER"))        pipe_attach_remote_reader (pipe, peer_id, unicast);    else    if (pipe && streq (request, "DATA")) {        //  TODO encode these commands as proper protocol        zframe_t *frame = zmsg_pop (msg);        zchunk_t *chunk = zchunk_new (zframe_data (frame), zframe_size (frame));        if (pipe->writer == REMOTE_NODE && pipe->reader) {            zsys_info ("send %d bytes to pipe", (int) zchunk_size (chunk));            pipe_send_data (pipe, &chunk);         }        else            zsys_info ("discard %d bytes, unroutable", (int) zchunk_size (chunk));                    zframe_destroy (&frame);        zchunk_destroy (&chunk);    }    else    if (pipe && streq (request, "DROP READER"))        pipe_drop_remote_reader (&pipe, peer_id);    else    if (pipe && streq (request, "DROP WRITER"))        pipe_drop_remote_writer (&pipe, peer_id);    else    if (streq (request, "DUMP"))        zyre_dump (self->zyre);    else        zsys_warning ("bad request %s from %s", request, peer_name);    zstr_free (&pipename);    zstr_free (&request);}

static ints_agent_handle_control (agent_t *self){    //  Get the whole message off the control socket in one go    zmsg_t *request = zmsg_recv (self->control);    char *command = zmsg_popstr (request);    if (!command)        return -1;                  //  Interrupted    if (streq (command, "SET")) {        char *name = zmsg_popstr (request);        char *value = zmsg_popstr (request);        curve_codec_set_metadata (self->codec, name, value);        free (name);        free (value);    }    else    if (streq (command, "CONNECT")) {        assert (!self->endpoint);        self->endpoint = zmsg_popstr (request);        int rc = zsocket_connect (self->dealer, "%s", self->endpoint);        assert (rc != -1);        zframe_t *server_key = zmsg_pop (request);        zframe_t *output = curve_codec_execute (self->codec, &server_key);        zframe_send (&output, self->dealer, 0);        self->state = connecting;    }    else    if (streq (command, "DISCONNECT")) {        if (self->endpoint) {            int rc = zsocket_disconnect (self->dealer, "%s", self->endpoint);            assert (rc != -1);            free (self->endpoint);        }    }    else    if (streq (command, "VERBOSE")) {        char *verbose = zmsg_popstr (request);        curve_codec_set_verbose (self->codec, *verbose == '1');        free (verbose);    }    else    if (streq (command, "TERMINATE")) {        self->state = terminated;        zstr_send (self->control, "OK");    }    else {        puts ("E: invalid command from API");        assert (false);    }    free (command);    zmsg_destroy (&request);    return 0;}

static voids_self_configure (self_t *self, zsock_t **sock_p, zmsg_t *request, char *name){    char *type_name = zmsg_popstr (request);    char *endpoints = zmsg_popstr (request);    if (self->verbose)        zsys_info ("zmonitor: - %s type=%s attach=%s", name, type_name, endpoints);    assert (*sock_p == NULL);    *sock_p = s_create_socket (type_name, endpoints);    assert (*sock_p);    zpoller_add (self->poller, *sock_p);    zstr_free (&type_name);    zstr_free (&endpoints);}

process_item *utils_msg2processitem(zmsg_t *message){  process_item *item = malloc(sizeof(process_item));  char *msgid = zmsg_popstr(message);  char *command = zmsg_popstr(message);  zframe_t *frame = zmsg_pop(message);  pid_t *pid = (pid_t*)frame;  memcpy(&item->pid,pid,sizeof(pid_t));  item->message_id = msgid;  item->command = command;  return item;}

static voids_alerts (    zsock_t *pipe,    void *args) {    const char *name = "ALERT";    mlm_client_t *cl = mlm_client_new ();    mlm_client_connect (cl, endpoint, 5000, __PRETTY_FUNCTION__);    mlm_client_set_producer (cl, stream);    zsock_t *msgpipe = mlm_client_msgpipe (cl);    zpoller_t *poller = zpoller_new (pipe, msgpipe, NULL);    char *alert_state = strdup ("NEW");    zsock_signal (pipe, 0);    while (!zsys_interrupted) {        zsock_t *which = zpoller_wait (poller, 1000);        if (!which) {            mlm_client_sendx (cl, "alert://[email
C++ zmsg_pushstr函数代码示例
C++ zmsg_new函数代码示例