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

END_TEST//  --------------------------------------------------------------------------/// Test _free ().START_TEST(test_msg_free){    sam_selftest_introduce ("test_msg_free");    zmsg_t *zmsg = zmsg_new ();    int rc = zmsg_pushstr (zmsg, "one");    ck_assert_int_eq (rc, 0);    rc = zmsg_pushstr (zmsg, "two");    ck_assert_int_eq (rc, 0);    sam_msg_t *msg = sam_msg_new (&zmsg);    ck_assert_int_eq (sam_msg_size (msg), 2);    char *pic_str;    rc = sam_msg_pop (msg, "s", &pic_str);    ck_assert_int_eq (rc, 0);    ck_assert_int_eq (sam_msg_size (msg), 1);    ck_assert_str_eq (pic_str, "two");    sam_msg_free (msg);    rc = sam_msg_pop (msg, "s", &pic_str);    ck_assert_int_eq (rc, 0);    ck_assert_int_eq (sam_msg_size (msg), 0);    ck_assert_str_eq (pic_str, "one");    sam_msg_destroy (&msg);}

static void s_broker_worker_msg(broker_t *self, zframe_t *sender, zmsg_t *msg){    assert (zmsg_size(msg) >= 1);     //  At least, command    zframe_t *command = zmsg_pop(msg);    char *id_string = zframe_strhex(sender);    int worker_ready = (zhash_lookup(self->workers, id_string) != NULL);    free (id_string);	    worker_t *worker = s_worker_require(self, sender);    if (zframe_streq(command, MDPW_READY)) {        if (worker_ready) {               //  Not first command in session            s_worker_delete(worker, 1);			// Додумать, по идеи синоним сердцебиения		} else {			if (zframe_size(sender) >= 4 &&  memcmp(zframe_data (sender), "mmi.", 4) == 0) {				s_worker_delete(worker, 1);				// Додумать, по идеи синоним сердцебиения			} else {				//  Attach worker to service and mark as idle				zframe_t *service_frame = zmsg_pop(msg);				worker->service = s_service_require(self, service_frame);				worker->service->workers++;				s_worker_waiting(worker);				zframe_destroy(&service_frame);			}		}    } else if (zframe_streq(command, MDPW_REPLY)) {        if (worker_ready) {            //  Remove and save client return envelope and insert the            //  protocol header and service name, then rewrap envelope.            zframe_t *client = zmsg_unwrap(msg);            zmsg_pushstr(msg, worker->service->name);            zmsg_pushstr(msg, MDPC_CLIENT);            zmsg_wrap(msg, client);            zmsg_send(&msg, self->socket);            s_worker_waiting(worker);        } else {			// Просто обрыв связи между воркером и брокером			// синоним сердцебиения            s_worker_delete(worker, 1);		}    } else if (zframe_streq(command, MDPW_HEARTBEAT)) {        if (worker_ready) {            worker->expiry = zclock_time () + HEARTBEAT_EXPIRY;		} else {			// Просто обрыв связи между воркером и брокером			// синоним сердцебиения            s_worker_delete(worker, 1);		}    } else if (zframe_streq (command, MDPW_DISCONNECT)) {        s_worker_delete(worker, 0);	} else {        zclock_log ("E: invalid input message");        zmsg_dump (msg);    }    free (command);    zmsg_destroy (&msg);}

triggerconfig_t * add_trigger(rulepackage_t * rpkg, zmsg_t * request,zmsg_t * reply) {    char * rule_id  = zmsg_popstr(request);    zclock_log("new trigger!");        if (remove_rule(rpkg->context, rpkg->triggers, rule_id)) {        // already have a rule with that id!       zclock_log("Received duplicate rule %s - killing old trigger", rule_id);    }           triggerconfig_t * tconf = malloc(sizeof(triggerconfig_t));    tconf->base_config = rpkg->base_config;    create_triggerconfig(tconf, request, rpkg->channel, rule_id);          // when we create this trigger, what guarantee do we have that    // the line listener is active? FIX    // basically we have none. crap.    char * created = create_trigger(rpkg->triggers, rule_id, rpkg->context, tconf);    if(NULL == created) {      // happy path, so add to db      zmsg_pushstr(reply, "ok");    } else {      zclock_log("create_trigger failed: %s", created);      free(tconf);      tconf=NULL;      zmsg_pushstr(reply, created);    }    free(created);    return tconf;        }

static voids_send_proxy_msg (zactor_t *proxy, const char *command, proxy_socket selected_socket, zmsg_t *msg){    zmsg_pushstr (msg, s_self_selected_socket_name (selected_socket));    zmsg_pushstr (msg, command);    assert (zmsg_send (&msg, proxy) == 0);}

static voids_worker_process (broker_t *self, zframe_t *sender, zmsg_t *msg){    assert (zmsg_size (msg) >= 1);     //  At least, command    zframe_t *command = zmsg_pop (msg);    char *identity = zframe_strhex (sender);    int worker_ready = (zhash_lookup (self->workers, identity) != NULL);    free (identity);    worker_t *worker = s_worker_require (self, sender);    if (zframe_streq (command, MDPW_READY)) {        if (worker_ready)               //  Not first command in session            s_worker_delete (self, worker, 1);        else        if (zframe_size (sender) >= 4  //  Reserved service name        &&  memcmp (zframe_data (sender), "mmi.", 4) == 0)            s_worker_delete (self, worker, 1);        else {            //  Attach worker to service and mark as idle            zframe_t *service_frame = zmsg_pop (msg);            worker->service = s_service_require (self, service_frame);            worker->service->workers++;            s_worker_waiting (self, worker);            zframe_destroy (&service_frame);        }    }    else    if (zframe_streq (command, MDPW_REPLY)) {        if (worker_ready) {            //  Remove & save client return envelope and insert the            //  protocol header and service name, then rewrap envelope.            zframe_t *client = zmsg_unwrap (msg);            zmsg_pushstr (msg, worker->service->name);            zmsg_pushstr (msg, MDPC_CLIENT);            zmsg_wrap (msg, client);            zmsg_send (&msg, self->socket);            s_worker_waiting (self, worker);        }        else            s_worker_delete (self, worker, 1);    }    else    if (zframe_streq (command, MDPW_HEARTBEAT)) {        if (worker_ready)            worker->expiry = zclock_time () + HEARTBEAT_EXPIRY;        else            s_worker_delete (self, worker, 1);    }    else    if (zframe_streq (command, MDPW_DISCONNECT))        s_worker_delete (self, worker, 0);    else {        zclock_log ("E: invalid input message");        zmsg_dump (msg);    }    free (command);    zmsg_destroy (&msg);}

static voids_broker_worker_msg(broker_t *self, zframe_t *sender, zmsg_t *msg){	assert(zmsg_size(msg) >= 1);	zframe_t *command = zmsg_pop(msg);	char *id_string = zframe_strhex(sender);	int worker_ready = (zhash_lookup(self->workers, id_string) != NULL);	free(id_string);	worker_t *worker = s_worker_require(self, sender);	if (zframe_streq(command, MDPW_READY)){		if (worker_ready)			s_worker_delete(worker, 1);		else 		if (zframe_size(sender) >= 4 && memcmp(zframe_data(sender), "mmi.", 4) == 0)			s_worker_delete(worker, 1);		else {			zframe_t *service_frame = zmsg_pop(msg);			worker->service = s_service_require(self, service_frame);			worker->service->workers++;			s_worker_waiting(worker);			zframe_destroy(&service_frame);		}	}	else	if (zframe_streq(command, MDPW_REPLY)){		if (worker_ready){			zframe_t *client = zmsg_unwrap(msg);			zmsg_pushstr(msg, worker->service->name);			zmsg_pushstr(msg, MDPC_CLIENT);			zmsg_wrap(msg, client);			zmsg_send(&msg, self->socket);			s_worker_waiting(worker);		}		else 			s_worker_delete(worker, 1);	}	else	if (zframe_streq(command, MDPW_HEARTBEAT)){		if (worker_ready)			worker->expiry = zclock_time() + HEARTBEAT_EXPIRY;		else			s_worker_delete(worker, 1);	}	else	if (zframe_streq(command, MDPW_DISCONNECT))		s_worker_delete(worker, 0);	else {		zclock_log("E: invalid input message");		zmsg_dump(msg);	}	free(command);	zmsg_destroy(&msg);}

void _rrwrk_send(void *socket, char *command, zmsg_t *msg){    msg = msg ? zmsg_dup(msg):zmsg_new();    zmsg_pushstr(msg, command);    zmsg_pushstr(msg, RR_WORKER);    zmsg_pushstr(msg, "");    zmsg_send(&msg, socket);}

zmsg_t *utils_gen_msg(const char *device_id, const char *msgid, const char *msg, char *bytes, int len){  zmsg_t *answer = zmsg_new();  if (bytes != NULL) {    zframe_t *frame = zframe_new (bytes, len);    zmsg_push(answer, frame);  }  zmsg_pushstr(answer, "%s", msg);  zmsg_pushstr(answer, "%s", msgid);  zmsg_pushstr(answer, "%s", device_id);  return answer;}

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

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

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

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

static void s_worker_send(worker_t *self, char *command, zmsg_t *msg){    msg = (msg ? zmsg_dup(msg): zmsg_new());	    zmsg_pushstr(msg, command);    zmsg_pushstr(msg, MDPW_WORKER);    //  Stack routing envelope to start of message    zmsg_wrap(msg, zframe_dup(self->identity));    if (self->broker->verbose) {        zclock_log ("I: sending %s to worker", mdps_commands [(int) *command]);        zmsg_dump(msg);    }    zmsg_send(&msg, self->broker->socket);}

int main (int argc, char *argv []){    int verbose = (argc > 1 && streq (argv [1], "-v"));    mdcli_t *session = mdcli_new ("tcp://localhost:5555", verbose);	zmsg_t *request;	zmsg_t *reply;    int count;    //for (count = 0; count < 1; count++) {        request = zmsg_new ();        zmsg_pushstr (request, "KILL");        reply = mdcli_send (session, "echo", &request);        if (reply)		{			zframe_t * lastFrame = zmsg_last(reply);			char* msg = zframe_strdup(lastFrame);			printf("msg:%s/n", msg);			printf("size:%d/n", zmsg_content_size(reply));			//zframe_print(lastFrame,"->");			zmsg_destroy (&reply);		}        //else        //    break;              //  Interrupt or failure    //}    //printf ("%d requests/replies processed/n", count);    mdcli_destroy (&session);    return 0;}

static voids_broker_client_msg(broker_t *self, zframe_t *sender, zmsg_t *msg){	assert(zmsg_size(msg) >= 2);	zframe_t *service_frame = zmsg_pop(msg);	service_t *service = s_service_require(self, service_frame);	zmsg_wrap(msg, zframe_dup(sender));	if (zframe_size(service_frame) >= 4 && memcmp(zframe_data(service_frame), "mmi.", 4) == 0){		char *return_code;		if (zframe_streq(service_frame, "mmi.service")){			char *name = zframe_strdup(zmsg_last(msg));			service_t *service = (service_t *)zhash_lookup(self->services, name);			return_code = service && service->workers ? "200" : "404";			free(name);		}		else			return_code = "501";		zframe_reset(zmsg_last(msg), return_code, strlen(return_code));		zframe_t *client = zmsg_unwrap(msg);		zmsg_prepend(msg, &service_frame);		zmsg_pushstr(msg, MDPC_CLIENT);		zmsg_wrap(msg, client);		zmsg_send(&msg, self->socket);	}	else 		s_service_dispatch(service, msg);	zframe_destroy(&service_frame);}

static voids_order_update (order_t *self, int volume){    assert (self);    assert (volume <= self->volume);    self->volume -= volume;    //  Prepare and send report to the client    zmsg_t *report = zmsg_new ();    zmsg_pushstrf (report, "%d", volume);    if (self->volume == 0)        zmsg_pushstr (report, "FILL");    else        zmsg_pushstr (report, "PARTIAL_FILL");    mdp_worker_send (self->worker, &report, self->reply_to);}

static voids_worker_send(worker_t *self, char *command, char *option, zmsg_t *msg){	msg = msg ? zmsg_dup(msg) : zmsg_new();	if (option)		zmsg_pushstr(msg, option);	zmsg_pushstr(msg, command);	zmsg_pushstr(msg, MDPW_WORKER);	zmsg_wrap(msg, zframe_dup(self->identity));	if (self->broker->verbose){		zclock_log("I: sending %s to worker", mdps_commands[(int) *command]);		zmsg_dump(msg);	}	zmsg_send(&msg, self->broker->socket);}

void send_trigger(mdcli_t * client, char * target_worker, char * rule_id,  int ival, char * user_id) {  zclock_log("activating trigger/ntarget=%s/nvalue=%d/nuser=%s",             target_worker, ival, user_id);  struct timeval tval;  gettimeofday(&tval, NULL);  // make a messagepack hash  msgpack_sbuffer * buffer =  msgpack_sbuffer_new();  msgpack_packer* pk = msgpack_packer_new(buffer, msgpack_sbuffer_write);  // value chunk  msgpack_pack_map(pk, 3);  // key  msgpack_pack_raw(pk, 5);  msgpack_pack_raw_body(pk, "value", 5);  // value  msgpack_pack_int(pk, ival);  //time chunk  // key  msgpack_pack_raw(pk, 5);  msgpack_pack_raw_body(pk, "epoch", 5);  // time  msgpack_pack_int(pk, tval.tv_sec);  msgpack_pack_raw(pk, 6);  msgpack_pack_raw_body(pk, "micros", 6);  // time  msgpack_pack_int(pk, tval.tv_usec);  zmsg_t * msg = zmsg_new();  // really, the user_id should be being added by a  // gatekeeper, not the block itself, or it's a security  // hole. will do for now FIX    zmsg_pushstr(msg, user_id);  // zmsg_pushmem(msg, &trigger.line_id, sizeof(int));  zmsg_pushmem(msg, buffer->data, buffer->size);  zmsg_pushstr(msg, rule_id);  zmsg_pushstr(msg, "DoAction");  mdcli_send(client, target_worker, &msg);}

voidmdp_client_send (mdp_client_t *self, char *service, zmsg_t **request_p){    assert (self);    assert (request_p);    zmsg_t *request = *request_p;    //  Prefix request with protocol frames    //  Frame 1: empty frame (delimiter)    //  Frame 2: "MDPCxy" (six bytes, MDP/Client x.y)    //  Frame 3: Service name (printable string)    zmsg_pushstr (request, service);    zmsg_pushstr (request, MDPC_CLIENT);    zmsg_pushstr (request, "");    if (self->verbose) {        zclock_log ("I: send request to '%s' service:", service);        zmsg_dump (request);    }    zmsg_send (request_p, self->client);}

int main(int argc, const char *argv[]) {    zsock_t *sock = zsock_new_req("tcp://127.0.0.1:5555");    int count = 10000;    zpoller_t *poller = zpoller_new(sock, NULL);    while (count--) {        zmsg_t *request = zmsg_new();        zmsg_pushstr(request, "HELLO");        zmsg_pushstr(request, "com.tw.echo");        zmsg_pushstr(request, MDPC_CLIENT);        zmsg_send(&request, sock);        zmsg_t *msg = zmsg_recv(sock);        zsys_info("GET RESPONSE: %d ", count);        zmsg_print(msg);        zmsg_destroy(&msg);    }    zsock_destroy(&sock);    return 0;}