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

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

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

voidzre_node_publish (zre_node_t *self, char *logical, char *physical){    zstr_sendm (self->pipe, "PUBLISH");    zstr_sendm (self->pipe, logical);    zstr_send  (self->pipe, physical);}

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

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

//  Publish file into virtual spacevoidzre_interface_publish (zre_interface_t *self,                       char *filename, char *external){    zstr_sendm (self->pipe, "PUBLISH");    zstr_sendm (self->pipe, filename);  //  Real file name    zstr_send  (self->pipe, external);  //  Location in virtual space}

intzyre_whisper (zyre_t *self, const char *peer, zmsg_t **msg_p){    assert (self);    assert (peer);    zstr_sendm (self->actor, "WHISPER");    zstr_sendm (self->actor, peer);    zmsg_send (msg_p, self->actor);    return 0;}

intzyre_shout (zyre_t *self, const char *group, zmsg_t **msg_p){    assert (self);    assert (group);    zstr_sendm (self->actor, "SHOUT");    zstr_sendm (self->actor, group);    zmsg_send (msg_p, self->actor);    return 0;}

voidcurve_client_connect (curve_client_t *self, char *endpoint, byte *server_key){    assert (self);    assert (endpoint);    assert (server_key);    zstr_sendm (self->control, "CONNECT");    zstr_sendm (self->control, endpoint);    zmq_send (self->control, server_key, 32, 0);}

char *zyre_peer_header_value (zyre_t *self, const char *peer, const char *name){    assert (self);    assert (peer);    assert (name);    zstr_sendm (self->actor, "PEER HEADER");    zstr_sendm (self->actor, peer);    zstr_send (self->actor, name);    return zstr_recv (self->actor);}

static zyre_group_t *zyre_node_leave_peer_group (zyre_node_t *self, zyre_peer_t *peer, char *name){    zyre_group_t *group = zyre_node_require_peer_group (self, name);    zyre_group_leave (group, peer);    //  Now tell the caller about the peer left group    zstr_sendm (self->pipe, "LEAVE");    zstr_sendm (self->pipe, zyre_peer_identity (peer));    zstr_send (self->pipe, name);    return group;}

static zyre_group_t *zyre_node_join_peer_group (zyre_node_t *self, zyre_peer_t *peer, char *name){    zyre_group_t *group = zyre_node_require_peer_group (self, name);    zyre_group_join (group, peer);    //  Now tell the caller about the peer joined group    zstr_sendm (self->pipe, "JOIN");    zstr_sendm (self->pipe, zyre_peer_identity (peer));    zstr_send (self->pipe, name);    return group;}

voidzre_interface_header_set (zre_interface_t *self, char *name, char *format, ...){    assert (self);    va_list argptr;    va_start (argptr, format);    char *value = (char *) malloc (255 + 1);    vsnprintf (value, 255, format, argptr);    va_end (argptr);        zstr_sendm (self->pipe, "SET");    zstr_sendm (self->pipe, name);    zstr_send  (self->pipe, value);    free (value);}

static zre_peer_t *s_require_peer (agent_t *self, char *identity, char *address, uint16_t port){    zre_peer_t *peer = (zre_peer_t *) zhash_lookup (self->peers, identity);    if (!peer) {        //  Purge any previous peer on same endpoint        char endpoint [100];        snprintf (endpoint, 100, "%s:%hu", address, port);        zhash_foreach (self->peers, agent_peer_purge, endpoint);        peer = zre_peer_new (identity, self->peers, self->ctx);        zre_peer_connect (peer, self->identity, endpoint);        //  Handshake discovery by sending HELLO as first message        zre_msg_t *msg = zre_msg_new (ZRE_MSG_HELLO);        zre_msg_ipaddress_set (msg, zre_udp_host (self->udp));        zre_msg_mailbox_set (msg, self->port);        zre_msg_groups_set (msg, zhash_keys (self->own_groups));        zre_msg_status_set (msg, self->status);        zre_msg_headers_set (msg, zhash_dup (self->headers));        zre_peer_send (peer, &msg);                zre_log_info (self->log, ZRE_LOG_MSG_EVENT_ENTER,                      zre_peer_endpoint (peer), endpoint);        //  Now tell the caller about the peer        zstr_sendm (self->pipe, "ENTER");        zstr_send (self->pipe, identity);    }    return peer;}

voidfmq_server_set_anonymous (fmq_server_t *self, long enabled){    assert (self);    zstr_sendm (self->pipe, "SET ANONYMOUS");    zstr_sendf (self->pipe, "%ld", enabled);}

//  --------------------------------------------------------------------------//  Set the node expiration timeout, in milliseconds. Default is 30000.//  This can be tuned in order to deal with expected network conditions//  and the response time expected by the application. This is tied to//  the beacon interval and rate of messages received.voidzyre_set_expired_timeout (zyre_t *self, int interval){    assert (self);    zstr_sendm (self->actor, "SET EXPIRED TIMEOUT");    zstr_sendf (self->actor, "%d", interval);}

voidcurve_server_set_max_pending (curve_server_t *self, int limit){    assert (self);    zstr_sendm (self->control, "MAX PENDING");    zstr_sendf (self->control, "%d", limit);}

curve_client_t *curve_client_new (zcert_t **cert_p){    curve_client_t *self = (curve_client_t *) zmalloc (sizeof (curve_client_t));    assert (self);    self->ctx = zctx_new ();    self->control = zthread_fork (self->ctx, s_agent_task, NULL);    //  Create separate data socket, send address on control socket    self->data = zsocket_new (self->ctx, ZMQ_PAIR);    assert (self->data);    int rc = zsocket_bind (self->data, "inproc://data-%p", self->data);    assert (rc != -1);    zstr_sendm (self->control, "inproc://data-%p", self->data);       //  Now send cert on control socket as well    rc = zmq_send (self->control, zcert_public_key (*cert_p), 32, ZMQ_SNDMORE);    assert (rc == 32);    rc = zmq_send (self->control, zcert_secret_key (*cert_p), 32, 0);    assert (rc == 32);        zcert_destroy (cert_p);    return self;}

voidfmq_client_set_resync (fmq_client_t *self, long enabled){    assert (self);    zstr_sendm (self->pipe, "SET RESYNC");    zstr_sendf (self->pipe, "%ld", enabled);}

voidcurve_server_set_pending_ttl (curve_server_t *self, int limit){    assert (self);    zstr_sendm (self->control, "PENDING TTL");    zstr_sendf (self->control, "%d", limit);}

voidzyre_set_interval (zyre_t *self, size_t interval){    assert (self);    zstr_sendm (self->actor, "SET INTERVAL");    zstr_sendf (self->actor, "%zd", interval);}

voidzyre_set_port (zyre_t *self, int port_nbr){    assert (self);    zstr_sendm (self->actor, "SET PORT");    zstr_sendf (self->actor, "%d", port_nbr);}

voidzmonitor_set_verbose (zmonitor_t *self, bool verbose){    assert (self);    zstr_sendm (self->pipe, "VERBOSE");    zstr_sendf (self->pipe, "%d", verbose);}

voidcurve_server_set_verbose (curve_server_t *self, bool verbose){    assert (self);    zstr_sendm (self->control, "VERBOSE");    zstr_sendf (self->control, "%d", verbose);}

voidcurve_server_set_max_clients (curve_server_t *self, int limit){    assert (self);    zstr_sendm (self->control, "MAX CLIENTS");    zstr_sendf (self->control, "%d", limit);}

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

voidcurve_server_set_client_ttl (curve_server_t *self, int limit){    assert (self);    zstr_sendm (self->control, "CLIENT TTL");    zstr_sendf (self->control, "%d", limit);}

//  --------------------------------------------------------------------------//  Set the node evasiveness timeout, in milliseconds. Default is 5000.//  This can be tuned in order to deal with expected network conditions//  and the response time expected by the application. This is tied to//  the beacon interval and rate of messages received.voidzyre_set_evasive_timeout (zyre_t *self, int interval){    assert (self);    zstr_sendm (self->actor, "SET EVASIVE TIMEOUT");    zstr_sendf (self->actor, "%d", interval);}

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

gboolean ipcam_service_send_strings(IpcamService *service,                                    const gchar *name,                                    const gchar *strings[],                                    const gchar *client_id){    gboolean ret = FALSE;    gint type;    void *mq_socket = NULL;    IpcamServicePrivate *priv = ipcam_service_get_instance_private(service);    g_return_val_if_fail(ipcam_socket_manager_get_by_name(priv->socket_manager, name, &type, &mq_socket), FALSE);    switch(type)    {    case IPCAM_SOCKET_TYPE_SERVER:        g_return_val_if_fail(client_id, FALSE);        zstr_sendm(mq_socket, client_id);        zmq_send_strings(mq_socket, strings);        ret = TRUE;        break;    case IPCAM_SOCKET_TYPE_PUBLISHER:    case IPCAM_SOCKET_TYPE_CLIENT:        zmq_send_strings(mq_socket, strings);        ret = TRUE;        break;    default:        break;    }    return ret;}