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

// timer callback; updates stock data and publishes new infostatic int onloop (zloop_t *loop, int timer, void *arg){  // get list of stocks and publisher socket  zloop_data_t *loopdata = (zloop_data_t *)arg;  // for each stock ...  zframe_t *frame = zframe_new_empty();  tick_t *stock = (tick_t *)zlist_first(loopdata->stocks);  while (stock != NULL)  {    // update point-in-time data    stock->timestamp  = time(NULL);    stock->value      = revalue(stock->value);    // publish point-in-time-data (each tick field is a seperate frame)        // Frame 1: stock symbol (to facilitate topic filtering)    frame = zframe_new(stock->symbol,strlen(stock->symbol));    zframe_send(&frame,loopdata->socket,ZFRAME_MORE);    // Frame 2: timestamp of last update    frame = zframe_new(&(stock->timestamp),sizeof(stock->timestamp));    zframe_send(&frame,loopdata->socket,ZFRAME_MORE);        // Frame 3: actual stock value    frame = zframe_new(&(stock->value),sizeof(stock->value));    zframe_send(&frame,loopdata->socket,0);    stock = zlist_next(loopdata->stocks);  }  zframe_destroy(&frame);    return 0;}

intmain (int argc, char *argv[]){    if (argc != 3) {        exit (-1);    }    int numb_msgs = atoi (argv[2]);    zctx_t *ctx = zctx_new ();    void *dealer = zsocket_new (ctx, ZMQ_DEALER);    zsocket_set_linger (dealer, -1);    zsocket_connect (dealer, "%s:9000", argv[1]);    void *sub = zsocket_new (ctx, ZMQ_SUB);    zsocket_connect (sub, "%s:9002", argv[1]);    zmq_setsockopt (sub, ZMQ_SUBSCRIBE, "all", 4);    int64_t time[2];    zmq_pollitem_t pollitem[1] = { {sub, 0, ZMQ_POLLIN}    };    zmq_poll (pollitem, 1, -1);    zmsg_t *signal = zmsg_recv (sub);    zmsg_destroy (&signal);    char blob[SIZE] = { 0 };    zmsg_t *msg = zmsg_new ();    zframe_t *frame = zframe_new (blob, SIZE);    zmsg_add (msg, frame);    time[0] = zclock_time ();    int i;    for (i = 0; i < numb_msgs; i++) {        zmsg_t *nmsg = zmsg_dup (msg);        zmsg_send (&nmsg, dealer);    }    time[1] = zclock_time ();    zmsg_destroy (&msg);    zmq_poll (pollitem, 1, -1);    msg = zmsg_recv (sub);    zmsg_destroy (&msg);    msg = zmsg_new ();    frame = zframe_new (time, sizeof (int64_t) * 2);    zmsg_add (msg, frame);    zmsg_send (&msg, dealer);    zctx_destroy (&ctx);}

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

intzmsg_addstr (zmsg_t *self, const char *format, ...){    assert (self);    assert (format);    //  Format string into buffer    va_list argptr;    va_start (argptr, format);    int size = 255 + 1;    char *string = (char *) malloc (size);    if (!string) {        va_end (argptr);        return -1;    }    int required = vsnprintf (string, size, format, argptr);    if (required >= size) {        size = required + 1;        string = (char *) realloc (string, size);        if (!string) {            va_end (argptr);            return -1;        }        vsnprintf (string, size, format, argptr);    }    va_end (argptr);    self->content_size += strlen (string);    zlist_append (self->frames, zframe_new (string, strlen (string)));    free (string);    return 0;}

/* 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 gbooleando_heartbeat (GPPWorker *self){  GPPWorkerPrivate *priv = GET_PRIV (self);  if (--priv->liveness == 0) {    g_warning ("W: heartbeat failure, can't reach queue/n");    g_warning ("W: reconnecting in %zd msec.../n", priv->interval);    g_source_remove (priv->frontend_source);    priv->frontend_source = 0;    g_io_channel_unref (priv->frontend_channel);    if (priv->interval < INTERVAL_MAX)      priv->interval *= 2;    zsocket_destroy (priv->ctx, priv->frontend);    g_timeout_add (priv->interval, (GSourceFunc) do_start, self);    return FALSE;  }  zframe_t *frame = zframe_new (PPP_HEARTBEAT, 1);  zframe_send (&frame, priv->frontend, 0);  /* We need to do that for some reason ... */  check_socket_activity (priv->frontend_channel, G_IO_IN, self);  return TRUE;}

//  Worker using REQ socket to do load-balancing//static void *worker_task(void *args){	zctx_t *ctx = zctx_new();	void *worker = zsocket_new(ctx, ZMQ_REQ);#if (defined (WIN32))	zsocket_connect(worker, "tcp://localhost:5673"); // backend#else	zsocket_connect(worker, "ipc://backend.ipc");#endif	//  Tell broker we're ready for work	zframe_t *frame = zframe_new(WORKER_READY, strlen(WORKER_READY));	zframe_send(&frame, worker, 0);	//  Process messages as they arrive	while (1) {		zmsg_t *msg = zmsg_recv(worker);		if (!msg)			break;              //  Interrupted		zframe_print(zmsg_last(msg), "Worker: ");		zframe_reset(zmsg_last(msg), "OK", 2);		zmsg_send(&msg, worker);	}	zctx_destroy(&ctx);	return NULL;}

void send_outgoing_messages(client_state* state, void * socket){    for(zchat_message_vector_t::iterator         it = state->out_messages.begin();        it != state->out_messages.end();        it++)    {        zchat_string_t serialised;        zchat_message * message = *it;                serialize_message_to_string(message, &serialised);        zframe_t* content = zframe_new (serialised.c_str(),                                        serialised.length());                zclock_sleep (randof (1000) + 1);                zframe_send (&content, socket, ZFRAME_REUSE);        if(message->type() == zchat_message_message_type_PING)        {            client_state_set_heartbeat_time(state);        }                zframe_destroy (&content);        zchat_message_destroy(message);    }        state->out_messages.clear();}

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

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

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 zframe_t *s_upstream_create_content (upstream_t *self){    int msgsize = self->size + randof (self->variance) - randof (self->variance);    zframe_t *content = zframe_new (NULL, msgsize);    return content;}

static void* worker_routine(void* arg){  zmsg_t* msg;  zframe_t* frame;  zctx_t* ctx = zctx_new();  void* worker = zsocket_new(ctx, ZMQ_REQ);  zsocket_connect(worker, "ipc://%s-localbe.ipc", self);  frame = zframe_new(WORKER_READY, 1);  zframe_send(&frame, worker, 0);  while (1) {    msg = zmsg_recv(worker);    if (!msg)      break;    zframe_print(zmsg_last(msg), "Worker: ");    zframe_reset(zmsg_last(msg), "OK", 2);    zmsg_send(&msg, worker);  }  zctx_destroy(&ctx);  return NULL;}

voidzmsg_addmem (zmsg_t *self, const void *src, size_t size){    assert (self);    zframe_t *frame = zframe_new (src, size);    self->content_size += size;    zlist_append (self->frames, frame);}

zframe_t *zchunk_pack (zchunk_t *self){    assert(self);    assert(zchunk_is(self));    return zframe_new (self->data, self->max_size);}

int create_object(zhash_t *object_hash, char *obj_name, char *algorithm,                  char *init_data, Server_Status *status) {    void *item =NULL;    char tag_str[BUFSIZE];    Tag tag;    item = zhash_lookup(object_hash, obj_name);    if( item!= NULL) return 0;    init_tag(&tag);    tag_to_string(tag, tag_str);    if( strcmp(algorithm, ABD)==0) {        zframe_t *payload_frame =            zframe_new((void *)server_args->init_data, server_args->init_data_size);        store_payload(object_hash, obj_name, tag, payload_frame, yield) ;        status->metadata_memory += (float) strlen(tag_str);        status->data_memory += (float) server_args->init_data_size;        printf("/tCreated /"%s/" (size %d) /n", obj_name, (int)status->data_memory);        //add it to the main list        return 1;    }    if( strcmp(algorithm, SODAW)==0) {        zframe_t *payload_frame =            zframe_new((void *)server_args->init_data, server_args->init_data_size);        store_payload(object_hash, obj_name, tag, payload_frame, yield) ;        status->metadata_memory += (float) strlen(tag_str);        status->data_memory += (float) server_args->init_data_size;        printf("/tCreated /"%s/" (size %lu) /n", obj_name,  strlen(init_data));        //add it to the main list        return 1;    }    return 0;}

int master_finder(zloop_t *loop, zmq_pollitem_t *item, void *arg) {    maneater_client *cli = (maneater_client *)arg;    zframe_t *out;    MSG_DOPACK(        msgpack_pack_int(pk, MID_WANT_MASTER);        MSG_PACK_STR(pk, cli->sessid);        MSG_PACK_STR(pk, cli->myhostid);        out = zframe_new(buf->data, buf->size);    );

JNIEXPORT jlong JNICALLJava_org_zeromq_czmq_Zframe__1_1new (JNIEnv *env, jclass c, jbyteArray data, jlong size){    jbyte *data_ = (byte *) (*env)->GetByteArrayElements (env, data, 0);    //  Disable CZMQ signal handling; allow Java to deal with it    zsys_handler_set (NULL);    jlong new_ = (jlong) (intptr_t) zframe_new (data_, (size_t) size);    (*env)->ReleaseByteArrayElements (env, data, (jbyte *) data_, 0);    return new_;}

zframe_t* get_frame_from_stdin(client_state* state){    zchat_string_t serialisedMessage;    get_serialised_message_from_stdin(state, &serialisedMessage);    zframe_t* content = zframe_new (serialisedMessage.c_str(),                                    serialisedMessage.length());        //const char * data = serialisedMessage.c_str();    return content;}

static void *_worker_socket(zctx_t *ctx){  void *worker=zsocket_new(ctx, ZMQ_DEALER);  zsocket_connect(worker, "tcp://localhost:5556");  debug_log("I: worker ready/n");  zframe_t *frame=zframe_new(PPP_READY, 1);  zframe_send(&frame, worker, 0);  return worker;}

intzmsg_addstr (zmsg_t *self, const char *string){    assert (self);    assert (zmsg_is (self));    assert (string);    self->content_size += strlen (string);    zlist_append (self->frames, zframe_new (string, strlen (string)));    return 0;}

void PrimeWorker::SendReply(const proto::Reply& rep, zmsg_t** msg, void* socket) {		size_t fsize = rep.ByteSize();	zframe_t* frame = zframe_new(0, fsize);	byte* data = zframe_data(frame);	rep.SerializeToArray(data, fsize);		zmsg_append(*msg, &frame);	zmsg_send(msg, socket);	}

static zsock_t *s_worker_socket () {	zsock_t *worker = zsock_new_dealer("tcp://localhost:5556");    //  Tell queue we're ready for work    printf ("I: worker ready/n");    zframe_t *frame = zframe_new (PPP_READY, 1);    zframe_send (&frame, worker, 0);    return worker;}

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

intzmsg_addmem (zmsg_t *self, const void *src, size_t size){    assert (self);    zframe_t *frame = zframe_new (src, size);    if (frame) {        self->content_size += size;        return zlist_append (self->frames, frame);    }    else        return -1;}

static void Send(const C& req, void* socket) {		zmsg_t* msg = zmsg_new();	size_t fsize = req.ByteSize();	zframe_t* frame = zframe_new(0, fsize);	byte* data = zframe_data(frame);	req.SerializeToArray(data, fsize);		zmsg_append(msg, &frame);	zmsg_send(&msg, socket);	}

static VALUE rb_czmq_frame_s_new(int argc, VALUE *argv, VALUE frame){    VALUE data;    errno = 0;    zframe_t *fr;    rb_scan_args(argc, argv, "01", &data);    if (NIL_P(data)) {        fr = zframe_new(NULL, 0);    } else {        Check_Type(data, T_STRING);        fr = zframe_new(RSTRING_PTR(data), (size_t)RSTRING_LEN(data));    }    if (fr == NULL) {        ZmqAssertSysError();        rb_memerror();    }    ZmqRegisterFrame(fr);    frame = Data_Wrap_Struct(rb_cZmqFrame, 0, rb_czmq_free_frame_gc, fr);    rb_obj_call_init(frame, 0, NULL);    return frame;}