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

int main (int argc, char *argv []){    if (argc == 1) {        printf ("I: syntax: %s <endpoint> .../n", argv [0]);        exit (EXIT_SUCCESS);    }    //  Create new freelance client object    flclient_t *client = flclient_new ();        //  Connect to each endpoint    int argn;    for (argn = 1; argn < argc; argn++)        flclient_connect (client, argv [argn]);        //  Send a bunch of name resolution 'requests', measure time    int requests = 10000;    uint64_t start = s_clock ();    while (requests--) {        zmsg_t *request = zmsg_new ("random name");        zmsg_t *reply = flclient_request (client, &request);        if (!reply) {            printf ("E: name service not available, aborting/n");            exit (EXIT_FAILURE);        }        zmsg_destroy (&reply);    }    printf ("Average round trip cost: %d usec/n",         (int) (s_clock () - start) / 10);        flclient_destroy (&client);    return 0;}

END_TEST//  --------------------------------------------------------------------------/// _pop () multiple times. Mainly used to test the garbage collection.START_TEST(test_msg_pop_successively){    sam_selftest_introduce ("test_msg_pop_successively");    zmsg_t *zmsg = zmsg_new ();    zmsg_pushstr (zmsg, "three");    zmsg_pushstr (zmsg, "two");    zmsg_pushstr (zmsg, "one");    char payload = '0';    zframe_t *frame = zframe_new (&payload, sizeof (payload));    zmsg_push (zmsg, frame);    sam_msg_t *msg = sam_msg_new (&zmsg);    zframe_t *zero;    char *one;    int rc = sam_msg_pop (msg, "fs", &zero, &one);    ck_assert_int_eq (rc, 0);    ck_assert_int_eq (sam_msg_size (msg), 2);    char *two, *three;    rc = sam_msg_pop (msg, "ss", &two, &three);    ck_assert_int_eq (rc, 0);    ck_assert_int_eq (sam_msg_size (msg), 0);    sam_msg_destroy (&msg);}

zmsg_t *zmsg_load (zmsg_t *self, FILE *file){    assert (file);    if (!self)        self = zmsg_new ();    if (!self)        return NULL;    while (true) {        size_t frame_size;        size_t rc = fread (&frame_size, sizeof (frame_size), 1, file);        if (rc == 1) {            zframe_t *frame = zframe_new (NULL, frame_size);            rc = fread (zframe_data (frame), frame_size, 1, file);            if (frame_size > 0 && rc != 1) {                zframe_destroy (&frame);                break;          //  Unable to read properly, quit            }            zmsg_append (self, &frame);        }        else            break;              //  Unable to read properly, quit    }    if (!zmsg_size (self)) {        zmsg_destroy (&self);        self = NULL;    }    return self;}

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) zhash_size (self->tuples));    }    else        zsys_error ("unknown zgossip method '%s'", method);    return reply;}

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

static void *client_task (void *args) {    void *context = zmq_init (1);    void *client = zmq_socket (context, ZMQ_XREQ);    //  Generate printable identity for the client    char identity [5];    sprintf (identity, "%04X", randof (0x10000));    zmq_setsockopt (client, ZMQ_IDENTITY, identity, strlen (identity));    zmq_connect (client, "tcp://localhost:5570");    zmq_pollitem_t items [] = { { client, 0, ZMQ_POLLIN, 0 } };    int request_nbr = 0;    while (1) {        //  Tick once per second, pulling in arriving messages        int centitick;        for (centitick = 0; centitick < 100; centitick++) {            zmq_poll (items, 1, 10000);            if (items [0].revents & ZMQ_POLLIN) {                zmsg_t *zmsg = zmsg_recv (client);                printf ("%s: %s/n", identity, zmsg_body (zmsg));                zmsg_destroy (&zmsg);            }        }        zmsg_t *zmsg = zmsg_new ();        zmsg_body_fmt (zmsg, "request #%d", ++request_nbr);        zmsg_send (&zmsg, client);    }    //  Clean up and end task properly    zmq_close (client);    zmq_term (context);    return (NULL);}

static voidhandle_mmi (client_t *self, const char *service_name) {    const char *result = "501";    zmsg_t *mmibody = mdp_msg_get_body(self->message);    if(mmibody) {        if(strstr(service_name, "mmi.service")) {            char *svc_lookup = zmsg_popstr(mmibody);            if(svc_lookup) {                service_t *service = (service_t *) zhash_lookup(self->server->services, svc_lookup);                result = service && service->workers ? "200" : "404";                zstr_free(&svc_lookup);            }        }        zmsg_destroy(&mmibody);    }    // Set routing id, messageid, service, body    mdp_msg_t *client_msg = mdp_msg_new();    mdp_msg_set_routing_id(client_msg, mdp_msg_routing_id(self->message));    mdp_msg_set_id(client_msg, MDP_MSG_CLIENT_FINAL);    mdp_msg_set_service(client_msg, service_name);    zmsg_t *rep_body = zmsg_new();    zmsg_pushstr(rep_body, result);    mdp_msg_set_body(client_msg, &rep_body);    mdp_msg_send(client_msg, self->server->router);    mdp_msg_destroy(&client_msg);}

void send_czmq (char *buf, int len){    zctx_t *zctx;    void *zs;    zmsg_t *zmsg;    if (!(zctx = zctx_new ()))        log_err_exit ("C: zctx_new");    if (lopt) /* zctx linger default = 0 (flush none) */        zctx_set_linger (zctx, linger);     if (!(zs = zsocket_new (zctx, ZMQ_DEALER)))        log_err_exit ("C: zsocket_new");    //if (lopt) // doesn't work here     //    zsocket_set_linger (zs, linger);     if (iopt)        zsocket_set_immediate (zs, imm);    //zsocket_set_sndhwm (zs, 0); /* unlimited */    if (zsocket_connect (zs, "%s", uri) < 0)        log_err_exit ("C: zsocket_connect");    if (!(zmsg = zmsg_new ()))        oom ();    if (zmsg_pushmem (zmsg, buf, bufsize) < 0)        oom ();    if (zmsg_send (&zmsg, zs) < 0)        log_err_exit ("C: zmsg_send");    if (sleep_usec > 0)        usleep (sleep_usec);    zctx_destroy (&zctx);}

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

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

static void _devio_destroy_smio_all (devio_t *self){#if 0    unsigned i;    for (i = 0; i < self->nnodes; ++i) {        /* This cannot fail at this point... but it can */        zmsg_t *msg = zmsg_new ();        /* An empty message means to selfdestruct */        zmsg_pushstr (msg, "");        zmsg_send (&msg, self->pipes [i]);    }#endif    /* Get all hash keys */    zlist_t *hash_keys = zhash_keys (self->sm_io_h);    ASSERT_ALLOC (hash_keys, err_hash_keys_alloc);    char *hash_item = zlist_first (hash_keys);    /* Iterate over all keys removing each of one */    for (; hash_item != NULL; hash_item = zlist_next (hash_keys)) {        /* FIXME: Usage of stroul fucntion for reconverting the string         * into a uint32_t */        _devio_destroy_smio (self, (uint32_t) strtoul (hash_item,                    (char **) NULL, 16));    }    zlist_destroy (&hash_keys);err_hash_keys_alloc:    return;}

static void _devio_destroy_smio (devio_t *self, uint32_t smio_id){    assert (self);    /* Stringify ID */    char *key_c = halutils_stringify_key (smio_id);    ASSERT_ALLOC (key_c, err_key_alloc);    /* Lookup SMIO reference in hash table */    void *pipe = zhash_lookup (self->sm_io_h, key_c);    ASSERT_TEST (pipe != NULL, "Could not find SMIO registered with this ID",            err_hash_lookup);    /* Send message to SMIO informing it to destroy itself */    /* This cannot fail at this point... but it can */    zmsg_t *send_msg = zmsg_new ();    ASSERT_ALLOC (send_msg, err_msg_alloc);    /* An empty message means to selfdestruct */    zmsg_pushstr (send_msg, "");    int zerr = zmsg_send (&send_msg, pipe);    ASSERT_TEST (zerr == 0, "Could not send self-destruct message to SMIO instance",            err_send_msg);    /* Finally, remove the pipe from hash */    zhash_delete (self->sm_io_h, key_c);err_send_msg:    zmsg_destroy (&send_msg);err_msg_alloc:err_hash_lookup:    free (key_c);err_key_alloc:    return;}

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 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 void zmqwriter (flux_reactor_t *r, flux_watcher_t *w,                       int revents, void *arg){    void *sock = flux_zmq_watcher_get_zsock (w);    static int count = 0;    if (revents & FLUX_POLLERR) {        fprintf (stderr, "%s: FLUX_POLLERR is set/n", __FUNCTION__);        goto error;    }    if (revents & FLUX_POLLOUT) {        uint8_t blob[64];        zmsg_t *zmsg = zmsg_new ();        if (!zmsg || zmsg_addmem (zmsg, blob, sizeof (blob)) < 0) {            fprintf (stderr, "%s: failed to create message: %s/n",                     __FUNCTION__, strerror (errno));            goto error;        }        if (zmsg_send (&zmsg, sock) < 0) {            fprintf (stderr, "%s: zmsg_send: %s/n",                     __FUNCTION__, strerror (errno));            goto error;        }        count++;        if (count == zmqwriter_msgcount)            flux_watcher_stop (w);    }    return;error:    flux_reactor_stop_error (r);}

void val_msg(void* line_in, char * msg, int k) {  zmsg_t * n = zmsg_new();  zmsg_pushstr(n, "foo");  // pushmem appears to copy the data.  zmsg_pushmem(n, &k,sizeof(int));  zmsg_send(&n, line_in);}

void read_serial(void * cvoid, zctx_t * context, void * pipe) {  char * buf;  serialconfig_t * config = (serialconfig_t*)cvoid;    FILE * in = config->in; // fopen("/dev/ttyO1", "r");  size_t nbytes=2047;  //  Prepare our context and publisher  buf = (char *) malloc(nbytes+1) ;  fprintf(stderr, "bound/n");  // first line is always garbage  getline(&buf, &nbytes, in);  child_handshake(pipe);  zsocket_destroy(context, pipe);  void* socket = zsocket_new(context, ZMQ_PUB);  zsocket_bind(socket, "inproc://raw_serial");    while ( getline(&buf, &nbytes, in) != -1 ) {#ifdef DEBUG    puts("line:");    puts(buf);#endif    zmsg_t * msg = zmsg_new();    zmsg_pushstr(msg, buf); // does buf need to be copied?    zmsg_send(&msg, socket);  }  fprintf(stderr, "error reading from stdin/n");  zsocket_destroy(context, socket);}

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 ssize_t _thsafe_zmq_client_read_generic (smio_t *self, loff_t offs, uint8_t *data,        uint32_t size){    assert (self);    ssize_t ret_size = -1;    zmsg_t *send_msg = zmsg_new ();    ASSERT_ALLOC(send_msg, err_msg_alloc);    uint32_t opcode;    DBE_DEBUG (DBG_MSG | DBG_LVL_TRACE, "[smio_thsafe_client:zmq] Calling _thsafe_read_generic/n");    switch (size) {        case THSAFE_READ_16_DSIZE:            opcode = THSAFE_READ_16;        break;        case THSAFE_READ_32_DSIZE:            opcode = THSAFE_READ_32;        break;        case THSAFE_READ_64_DSIZE:            opcode = THSAFE_READ_64;        break;        default:            opcode = THSAFE_READ_32;    }    /* Message is:     * frame 0: READ<size> opcode     * frame 1: offset */    int zerr = zmsg_addmem (send_msg, &opcode, sizeof (opcode));    ASSERT_TEST(zerr == 0, "Could not add READ opcode in message",            err_add_opcode);    zerr = zmsg_addmem (send_msg, &offs, sizeof (offs));    ASSERT_TEST(zerr == 0, "Could not add offset in message",            err_add_offset);    DBE_DEBUG (DBG_MSG | DBG_LVL_TRACE, "[smio_thsafe_client:zmq] Sending message:/n");#ifdef LOCAL_MSG_DBG    zmsg_print (send_msg);#endif    zerr = zmsg_send (&send_msg, self->pipe);    ASSERT_TEST(zerr == 0, "Could not send message", err_send_msg);    /* Message is:     * frame 0: reply code     * frame 1: return code     * frame 2: data */    ret_size = _thsafe_zmq_client_recv_read (self, data, size);err_send_msg:err_add_offset:err_add_opcode:    zmsg_destroy (&send_msg);err_msg_alloc:    return ret_size;}