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

    void unregister(std::string const &name) {	directoryd::ServiceRequest request;	request.set_type(directoryd::UNREGISTER);	auto *r = request.mutable_unregister();	r->set_name(name);	zframe_t *sf = zframe_new(NULL, request.ByteSize());	assert (sf != NULL);	request.SerializeToArray(zframe_data(sf),zframe_size(sf));	int retval = zframe_send(&sf,				 DDClient::instance().register_socket(), 0);	assert(retval == 0);	zframe_t *rf = zframe_recv (DDClient::instance().register_socket());	directoryd::ServiceReply reply;	reply.ParseFromArray(zframe_data(rf),zframe_size(rf));	zframe_destroy(&rf);	if (reply.type() != directoryd::UNREGISTER) {	    throw RegistrationError("Got back incorrect message type when trying to unregister.");	}	if (reply.success() != true) {	    throw RegistrationError(reply.result());	}    }

zhashx_t *zhashx_unpack (zframe_t *frame){    zhashx_t *self = zhashx_new ();    if (!self)        return NULL;    assert (frame);    if (zframe_size (frame) < 4)        return self;            //  Arguable...    byte *needle = zframe_data (frame);    byte *ceiling = needle + zframe_size (frame);    size_t nbr_items = ntohl (*(uint32_t *) needle);    needle += 4;    while (nbr_items && needle < ceiling) {        //  Get key as string        size_t key_size = *needle++;        if (needle + key_size <= ceiling) {            char key [256];            memcpy (key, needle, key_size);            key [key_size] = 0;            needle += key_size;            //  Get value as longstr            if (needle + 4 <= ceiling) {                size_t value_size = ntohl (*(uint32_t *) needle);                needle += 4;                //  Be wary of malformed frames                if (needle + value_size <= ceiling) {                    char *value = (char *) zmalloc (value_size + 1);                    if (!value) {                        zhashx_destroy (&self);                        return NULL;                    }                    memcpy (value, needle, value_size);                    value [value_size] = 0;                    needle += value_size;                    //  Hash takes ownership of value                    if (zhashx_insert (self, key, value)) {                        zhashx_destroy (&self);                        break;                    }                }            }        }    }    //  Hash will free values in destructor    if (self)        zhashx_autofree (self);    return self;}

static ints_self_handle_pipe (self_t *self){    //  Get just the command off the pipe    char *command = zstr_recv (self->pipe);    if (!command)        return -1;                  //  Interrupted    if (self->verbose)        zsys_info ("zbeacon: API command=%s", command);    if (streq (command, "VERBOSE"))        self->verbose = true;    else    if (streq (command, "CONFIGURE")) {        int port;        int rc = zsock_recv (self->pipe, "i", &port);        assert (rc == 0);        s_self_configure (self, port);    }    else    if (streq (command, "PUBLISH")) {        zframe_destroy (&self->transmit);        zsock_recv (self->pipe, "fi", &self->transmit, &self->interval);        assert (zframe_size (self->transmit) <= UDP_FRAME_MAX);        if (self->interval == 0)            self->interval = INTERVAL_DFLT;        //  Start broadcasting immediately        self->ping_at = zclock_mono ();    }    else    if (streq (command, "SILENCE"))        zframe_destroy (&self->transmit);    else    if (streq (command, "SUBSCRIBE")) {        zframe_destroy (&self->filter);        self->filter = zframe_recv (self->pipe);        assert (zframe_size (self->filter) <= UDP_FRAME_MAX);    }    else    if (streq (command, "UNSUBSCRIBE"))        zframe_destroy (&self->filter);    else    if (streq (command, "$TERM"))        self->terminated = true;    else {        zsys_error ("zbeacon: - invalid command: %s", command);        assert (false);    }    zstr_free (&command);    return 0;}

voidinterval_minit (interval_t ** interval, zmsg_t * msg){    *interval = malloc (sizeof (interval_t));    zframe_t *frame = zmsg_first (msg);    memcpy (&((*interval)->start), zframe_data (frame), zframe_size (frame));    frame = zmsg_next (msg);    memcpy (&((*interval)->end), zframe_data (frame), zframe_size (frame));}

static VALUE rb_czmq_frame_cmp(VALUE obj, VALUE other_frame){    long diff;    zframe_t *other = NULL;    if (obj == other_frame) return INT2NUM(0);    ZmqGetFrame(obj);    ZmqAssertFrame(other_frame);    Data_Get_Struct(other_frame, zframe_t, other);    if (!other) rb_raise(rb_eTypeError, "uninitialized ZMQ frame!"); /    if (!(st_lookup(frames_map, (st_data_t)other, 0))) rb_raise(rb_eZmqError, "object %p has been destroyed by the ZMQ framework", (void *)other_frame);    diff = (zframe_size(frame) - zframe_size(other));    if (diff == 0) return INT2NUM(0);    if (diff > 0) return INT2NUM(1);    return INT2NUM(-1);}

static ssize_t _thsafe_zmq_client_recv_read (smio_t *self, uint8_t *data,        uint32_t size){    ssize_t ret_size = -1;    /* Returns NULL if confirmation was not OK or in case of error.     * Returns the original message if the confirmation was OK */    zmsg_t *recv_msg = _thsafe_zmq_client_recv_confirmation (self);    ASSERT_TEST(recv_msg != NULL, "Could not receive confirmation code", err_null_recv_msg);    /* If we are here, confirmation code was OK. Check for second frame */    zframe_t *reply_frame = zmsg_pop (recv_msg);    zframe_destroy (&reply_frame); /* Don't do anything with the reply code */    zframe_t *return_frame = zmsg_pop (recv_msg);    ASSERT_TEST(return_frame != NULL, "Could not receive retrurn code", err_null_ret_code_frame);    if (zframe_size (return_frame) != THSAFE_RETURN_SIZE) {        DBE_DEBUG (DBG_MSG | DBG_LVL_ERR, "[smio_thsafe_client:zmq] Return frame size is wrong/n");        goto err_wrong_size_ret_frame;    }    zframe_t *data_frame = zmsg_pop (recv_msg);    ASSERT_TEST(data_frame != NULL, "Could not receive data", err_recv_data);    /* Check if the frame has the number of bytes requested.     * For now, we consider a success only when the number of     * bytes requested is the same as the actually read*/    if ((ssize_t) zframe_size (data_frame) != *(THSAFE_RETURN_TYPE *) zframe_data (return_frame)) {        DBE_DEBUG (DBG_MSG | DBG_LVL_TRACE, "[smio_thsafe_client:zmq] Data frame size is wrong/n");        goto err_recv_data;    }    uint8_t* raw_data = (uint8_t *) zframe_data (data_frame);    memcpy (data, raw_data, size);    ret_size = size;    DBE_DEBUG (DBG_MSG | DBG_LVL_TRACE, "[smio_thsafe_client:zmq] Received data successfully/n");    zframe_destroy (&data_frame);err_recv_data:err_wrong_size_ret_frame:    zframe_destroy (&return_frame);err_null_ret_code_frame:    zmsg_destroy (&recv_msg);err_null_recv_msg:    return ret_size;}

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

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

/**************** Helper Functions ***************/static devio_err_e _devio_do_smio_op (devio_t *self, void *msg){    zmq_server_args_t *server_args = (zmq_server_args_t *) msg;    /* Message is:     * frame 0: opcode     * frame 1: payload */    /* Extract the first frame and determine the opcode */    zframe_t *opcode = zmsg_pop (*server_args->msg);    devio_err_e err = (opcode == NULL) ? DEVIO_ERR_BAD_MSG : DEVIO_SUCCESS;    ASSERT_TEST(opcode != NULL, "Could not receive opcode", err_null_opcode);    if (zframe_size (opcode) != THSAFE_OPCODE_SIZE) {        DBE_DEBUG (DBG_DEV_IO | DBG_LVL_ERR,                "[dev_io_core:poll_all_sm] Invalid opcode size received/n");        err = DEVIO_ERR_BAD_MSG;        goto err_wrong_opcode_size;    }    uint32_t opcode_data = *(uint32_t *) zframe_data (opcode);    if (opcode_data > THSAFE_OPCODE_END-1) {        DBE_DEBUG (DBG_DEV_IO | DBG_LVL_ERR,                "[dev_io_core:poll_all_sm] Invalid opcode received/n");        err = DEVIO_ERR_BAD_MSG;        goto err_invalid_opcode;    }    /* Do the actual work... */    disp_table_call (self->disp_table_thsafe_ops, opcode_data, self, server_args);err_invalid_opcode:err_wrong_opcode_size:    zframe_destroy (&opcode);err_null_opcode:    return err;}

void  SNetDistribZMQUnpack(zframe_t **srcframe, void *dst, size_t count){  if (*srcframe != NULL) {    size_t dst_size = count;    size_t srcframe_size = zframe_size(*srcframe);    byte *srcframe_data = zframe_data(*srcframe);    if(dst_size > srcframe_size) {      dst = NULL;    } else {      memcpy(dst, srcframe_data, dst_size);      if ((srcframe_size - dst_size) != 0) {        byte *newdst = SNetMemAlloc(srcframe_size - dst_size);        memcpy(newdst, srcframe_data + count, srcframe_size - dst_size);        zframe_reset(*srcframe, newdst, srcframe_size - dst_size);        SNetMemFree(newdst);      } else {        zframe_destroy(srcframe);        *srcframe = NULL;      }    }  } else {    dst = NULL;  }}

voidzmsg_remove (zmsg_t *self, zframe_t *frame){    assert (self);    self->content_size -= zframe_size (frame);    zlist_remove (self->frames, frame);}

zgossip_msg_t *zgossip_msg_decode (zmsg_t **msg_p){    assert (msg_p);    zmsg_t *msg = *msg_p;    if (msg == NULL)        return NULL;            zgossip_msg_t *self = zgossip_msg_new (0);    //  Read and parse command in frame    zframe_t *frame = zmsg_pop (msg);    if (!frame)         goto empty;             //  Malformed or empty    //  Get and check protocol signature    self->needle = zframe_data (frame);    self->ceiling = self->needle + zframe_size (frame);    uint16_t signature;    GET_NUMBER2 (signature);    if (signature != (0xAAA0 | 0))        goto empty;             //  Invalid signature    //  Get message id and parse per message type    GET_NUMBER1 (self->id);    switch (self->id) {        case ZGOSSIP_MSG_HELLO:            break;        case ZGOSSIP_MSG_ANNOUNCE:            GET_STRING (self->endpoint);            GET_STRING (self->service);            break;        case ZGOSSIP_MSG_PING:            break;        case ZGOSSIP_MSG_PONG:            break;        case ZGOSSIP_MSG_INVALID:            break;        default:            goto malformed;    }    //  Successful return    zframe_destroy (&frame);    zmsg_destroy (msg_p);    return self;    //  Error returns    malformed:        printf ("E: malformed message '%d'/n", self->id);    empty:        zframe_destroy (&frame);        zmsg_destroy (msg_p);        zgossip_msg_destroy (&self);        return (NULL);}

static ssize_t _thsafe_zmq_client_recv_write (smio_t *self){    ssize_t ret_size = -1;    /* Returns NULL if confirmation was not OK or in case of error.     * Returns the original message if the confirmation was OK */    zmsg_t *recv_msg = _thsafe_zmq_client_recv_confirmation (self);    ASSERT_TEST(recv_msg != NULL, "Could not receive confirmation code", err_null_recv_msg);    /* If we are here, confirmation code was OK. Check for second frame */    zframe_t *reply_frame = zmsg_pop (recv_msg);    zframe_destroy (&reply_frame); /* Don't do anything with the reply code */    zframe_t *return_frame = zmsg_pop (recv_msg);    ASSERT_TEST(return_frame != NULL, "Could not receive retrurn code", err_null_ret_code_frame);    if (zframe_size (return_frame) != THSAFE_RETURN_SIZE) {        DBE_DEBUG (DBG_MSG | DBG_LVL_ERR, "[smio_thsafe_client:zmq] Return frame size is wrong/n");        goto err_wrong_size_ret_frame;    }    ret_size = *(THSAFE_RETURN_TYPE *) zframe_data (return_frame);    DBE_DEBUG (DBG_MSG | DBG_LVL_TRACE, "[smio_thsafe_client:zmq] Received code: %zd/n", ret_size);err_wrong_size_ret_frame:    zframe_destroy (&return_frame);err_null_ret_code_frame:err_null_recv_msg:    return ret_size;}

static VALUE rb_czmq_frame_size(VALUE obj){    size_t size;    ZmqGetFrame(obj);    size = zframe_size(frame);    return LONG2FIX(size);}

static VALUE rb_czmq_frame_data(VALUE obj){    size_t size;    ZmqGetFrame(obj);    size = zframe_size(frame);    return ZmqEncode(rb_str_new((char *)zframe_data(frame), (long)size));}

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

int handle_event(zloop_t *loop, zsock_t *reader, void *args) {         // initialization        ubx_block_t *b = (ubx_block_t *) args;        struct zmq_receiver_info *inf = (struct zmq_receiver_info*) b->private_data;        printf("zmq_receiver: data available./n");        zframe_t *frame = zframe_recv (reader);        // print out frame data        zframe_print (frame, NULL);        // move to step function?        ubx_type_t* type =  ubx_type_get(b->ni, "unsigned char");        ubx_data_t msg;        msg.data = (void *)zframe_data(frame);        msg.len = zframe_size(frame);        msg.type = type;        //hexdump((unsigned char *)msg.data, msg.len, 16);        __port_write(inf->ports.zmq_in, &msg);        /* Inform potential observers ? */        // clean up temporary frame object        zframe_destroy (&frame);        return 1;}

/* step */void zmq_sender_step(ubx_block_t *b){        struct zmq_sender_info *inf = (struct zmq_sender_info*) b->private_data;//        std::cout << "zmq_sender: Processing a port update" << std::endl;	/* Read data from port */	ubx_port_t* port = inf->ports.zmq_out;	assert(port != 0);	ubx_data_t msg;	checktype(port->block->ni, port->in_type, "unsigned char", port->name, 1);	msg.type = port->in_type;	msg.len = inf->buffer_length;	msg.data = inf->buffer;//	std::cout << "zmq_sender: Reading from port" << std::endl;	int read_bytes = __port_read(port, &msg);	if (read_bytes <= 0) {//		std::cout << "zmq_sender: No data recieved from port" << std::endl;		return;	}	std::cout << "zmq_sender: read bytes = " << read_bytes << std::endl;	/* Setup ZMQ frame. At this point only single frames are sent. This can be replaced by zmsg_t messages            if multi-part messages become necessary*/	zframe_t* message = zframe_new(msg.data, read_bytes);	std::cout << "Created frame of length " << zframe_size(message) << std::endl;	/* Send the message */	int result = zframe_send(&message, inf->publisher,0);	std::cout << "send message with result " << result << std::endl;}

static int event(zloop_t *loop, zmq_pollitem_t *item, void *arg){	if (interrupt)		return -1;	zmsg_t *msg = zmsg_recv(dealer);	zframe_t *payload = zmsg_pop(msg);	zmsg_destroy(&msg);	msgpack_unpacked object;	msgpack_unpacked_init(&object);		if (msgpack_unpack_next(&object, (char*)zframe_data(payload), zframe_size(payload) , NULL))	{		//zclock_log("message");		//msgpack_object_print(stdout, object.data);		char *command = (char*)m_lookup(object.data, "command");		if (command) {			//zclock_log("command: %s", command);			if (streq(command, "exception")) {			    failed++;			}			if (streq(command, "result")) {				success++;			}			free(command);		}	}	msgpack_unpacked_destroy(&object);	zframe_destroy(&payload);	return 0;}

    void register_service(std::string const &name,			  int port,			  std::map<std::string, std::string> const &txt)    {	directoryd::ServiceRequest request;	request.set_type(directoryd::REGISTER);	auto *r = request.mutable_register_();	auto l = r->add_location();	l->set_port(port);	l->set_type("_hotdec._tcp");	r->set_name(name);	for (auto &t : txt) {	    auto txtfield = r->add_txt();	    txtfield->set_key(t.first);	    txtfield->set_value(t.second);	}	zframe_t *sf = zframe_new(NULL, request.ByteSize());	assert (sf != NULL);	request.SerializeToArray(zframe_data(sf),zframe_size(sf));	string buffer;	if (debug && TextFormat::PrintToString(request, &buffer)) {	    fprintf(stderr, "request: %s/n", buffer.c_str());	}	int retval = zframe_send(&sf, DDClient::instance().register_socket(), 0);	assert(retval == 0);	zframe_t *rf = zframe_recv (DDClient::instance().register_socket());	directoryd::ServiceReply reply;	reply.ParseFromArray(zframe_data(rf),zframe_size(rf));	if (debug && TextFormat::PrintToString(reply, &buffer)) {	    fprintf(stderr, "reply: %s/n", buffer.c_str());	}	zframe_destroy(&rf);	if (reply.type() != directoryd::REGISTER) {	    throw RegistrationError("Got back incorrect message type when trying to register.");	}	if (reply.success() != true) {	    throw RegistrationError(reply.result());	}	RegistrationManager::instance().add(name);    }

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

zchunk_t *zchunk_unpack (zframe_t *frame){    assert(frame);    assert(zframe_is(frame));    return zchunk_new (zframe_data (frame), zframe_size (frame) );}

int16_t message_get_msgtype(zmsg_t *msg){    zframe_t *frame_msgtype = zmsg_first(msg);    if ( frame_msgtype != NULL && zframe_size(frame_msgtype) == sizeof(int16_t) ){        return *(int16_t*)zframe_data(frame_msgtype);    }     return MSGTYPE_UNKNOWN;}

intzmsg_push (zmsg_t *self, zframe_t *frame){    assert (self);    assert (frame);    self->content_size += zframe_size (frame);    return zlist_push (self->frames, (void *) frame);}

intzmsg_add (zmsg_t *self, zframe_t *frame){    assert (self);    assert (frame);    self->content_size += zframe_size (frame);    return zlist_append (self->frames, frame);}

size_tzmsg_encode (zmsg_t *self, byte **buffer){    assert (self);    assert (zmsg_is (self));    //  Calculate real size of buffer    size_t buffer_size = 0;    zframe_t *frame = zmsg_first (self);    while (frame) {        size_t frame_size = zframe_size (frame);        if (frame_size < 255)            buffer_size += frame_size + 1;        else            buffer_size += frame_size + 1 + 4;        frame = zmsg_next (self);    }    *buffer = (byte *) zmalloc (buffer_size);    if (*buffer) {        //  Encode message now        byte *dest = *buffer;        frame = zmsg_first (self);        while (frame) {            size_t frame_size = zframe_size (frame);            if (frame_size < 255) {                *dest++ = (byte) frame_size;                memcpy (dest, zframe_data (frame), frame_size);                dest += frame_size;            }            else {                *dest++ = 0xFF;                *dest++ = (frame_size >> 24) & 255;                *dest++ = (frame_size >> 16) & 255;                *dest++ = (frame_size >>  8) & 255;                *dest++ =  frame_size        & 255;                memcpy (dest, zframe_data (frame), frame_size);                dest += frame_size;            }            frame = zmsg_next (self);        }        assert ((dest - *buffer) == buffer_size);    }    return buffer_size;}

JNIEXPORT jbyteArray JNICALLJava_org_zeromq_czmq_Zframe__1_1data (JNIEnv *env, jclass c, jlong self){    jbyte *data_ = (jbyte *) zframe_data ((zframe_t *) (intptr_t) self);    jint return_size_ = (jint) zframe_size ((zframe_t *) (intptr_t) self);    jbyteArray return_data_ = (*env)->NewByteArray (env, return_size_);    (*env)->SetByteArrayRegion (env, return_data_, 0, return_size_, (jbyte *) data_);    return return_data_;}

zframe_t *zmsg_pop (zmsg_t *self){    assert (self);    zframe_t *frame = zlist_pop (self->frames);    if (frame)        self->content_size -= zframe_size (frame);    return frame;}

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