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

static ints_agent_handle_data (agent_t *self){    //  First frame is client address (hashkey)    //  If caller sends unknown client address, we discard the message    //  For testing, we'll abort in this case, since it cannot happen    //  The assert disappears when we start to timeout clients...    zmsg_t *request = zmsg_recv (self->data);    char *hashkey = zmsg_popstr (request);    client_t *client = (client_t *) zhash_lookup (self->clients, hashkey);    free (hashkey);    if (client) {        //  Encrypt and send all frames of request        //  Each frame is a full ZMQ message with identity frame        while (zmsg_size (request)) {            zframe_t *cleartext = zmsg_pop (request);            if (zmsg_size (request))                zframe_set_more (cleartext, 1);            zframe_t *encrypted = curve_codec_encode (client->codec, &cleartext);            if (encrypted) {                zframe_send (&client->address, self->router, ZFRAME_MORE + ZFRAME_REUSE);                zframe_send (&encrypted, self->router, 0);            }            else                client_set_exception (client);        }    }    zmsg_destroy (&request);    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 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);}

intzsock_wait (void *self){    assert (self);    //  A signal is a message containing one frame with our 8-byte magic     //  value. If we get anything else, we discard it and continue to look    //  for the signal message    while (true) {        zmsg_t *msg = zmsg_recv (self);        if (!msg)            return -1;        if (zmsg_size (msg) == 1        &&  zmsg_content_size (msg) == 8) {            zframe_t *frame = zmsg_first (msg);            int64_t signal_value = *((int64_t *) zframe_data (frame));            if ((signal_value & 0xFFFFFFFFFFFFFF00L) == 0x7766554433221100L) {                zmsg_destroy (&msg);                return signal_value & 255;            }        }        zmsg_destroy (&msg);    }    return -1;}

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_engine_handle_request (engine_t *self, zmsg_t *request, zframe_t *reply_to){    assert (zmsg_size (request) >= 3);    zframe_t *operation = zmsg_pop (request);    zframe_t *price     = zmsg_pop (request);    zframe_t *volume    = zmsg_pop (request);    if (zframe_streq (operation, "SELL"))        s_engine_handle_sell_request (self, price, volume, reply_to);    else    if (zframe_streq (operation, "BUY"))        s_engine_handle_buy_request (self, price, volume, reply_to);    else {        zclock_log ("E: invalid message: ");        zmsg_dump (request);    }    zframe_destroy (&operation);    zframe_destroy (&price);    zframe_destroy (&volume);    zmsg_destroy (&request);}

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

/** *  * @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 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);}

intcurve_server_send (curve_server_t *self, zmsg_t **msg_p){    assert (self);    assert (zmsg_size (*msg_p) > 0);    zmsg_send (msg_p, self->data);    return 0;}

static ints_agent_handle_data (agent_t *self){    //  Encrypt and send all frames of request    zmsg_t *request = zmsg_recv (self->data);    while (zmsg_size (request)) {        zframe_t *cleartext = zmsg_pop (request);        if (zmsg_size (request))            zframe_set_more (cleartext, 1);        zframe_t *encrypted = curve_codec_encode (self->codec, &cleartext);        if (encrypted)            zframe_send (&encrypted, self->dealer, 0);        else            self->state = exception;    }    zmsg_destroy (&request);    return 0;}

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

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 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++ zmw_get_wlan_dev函数代码示例
C++ zmsg_send函数代码示例