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

void send_frames_at_server(zhash_t *frames, void *worker,  enum SEND_TYPE type, int n, ...) {    char *key;    va_list valist;    int i;    va_start(valist, n);    zlist_t *names = zlist_new();    for(i=0; i < n; i++ ) {        key = va_arg(valist, char *);        zframe_t *frame = (zframe_t *)zhash_lookup(frames, key);        assert(frame!=NULL);         assert(zframe_is(frame));        zlist_append(names, key);        if( i == n-1 && type==SEND_FINAL)  {            zframe_send(&frame, worker, ZFRAME_REUSE);        } else            zframe_send(&frame, worker, ZFRAME_REUSE + ZFRAME_MORE);    }    va_end(valist);    if(DEBUG_MODE) print_out_hash_in_order(frames, names);    zlist_purge(names);    zlist_destroy(&names);}

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

static voidserver_worker (void *args, zctx_t *ctx, void *pipe){    void *worker = zsocket_new (ctx, ZMQ_DEALER);    zsocket_connect (worker, "inproc://backend");    while (true) {        //  The DEALER socket gives us the reply envelope and message        zmsg_t *msg = zmsg_recv (worker);        zframe_t *identity = zmsg_pop (msg);        zframe_t *content = zmsg_pop (msg);        assert (content);        zmsg_destroy (&msg);        //  Send 0..4 replies back        int reply, replies = randof (5);        for (reply = 0; reply < replies; reply++) {            //  Sleep for some fraction of a second            zclock_sleep (randof (1000) + 1);            zframe_send (&identity, worker, ZFRAME_REUSE + ZFRAME_MORE);            zframe_send (&content, worker, ZFRAME_REUSE);        }        zframe_destroy (&identity);        zframe_destroy (&content);    }}

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

static ints_agent_handle_dealer (agent_t *self){    if (self->state == connecting) {        zframe_t *input = zframe_recv (self->dealer);        zframe_t *output = curve_codec_execute (self->codec, &input);        if (output)            zframe_send (&output, self->dealer, 0);        else        if (curve_codec_connected (self->codec))            self->state = connected;        else        if (curve_codec_exception (self->codec))            self->state = exception;    }    else    if (self->state == connected) {        zframe_t *encrypted = zframe_recv (self->dealer);        zframe_t *cleartext = curve_codec_decode (self->codec, &encrypted);        if (cleartext) {            int flags = zframe_more (cleartext)? ZFRAME_MORE: 0;            zframe_send (&cleartext, self->data, flags);        }        else            self->state = exception;    }    return 0;}

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

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

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

static ints_snapshots (zloop_t *loop, zmq_pollitem_t *poller, void *args){    clonesrv_t *self = (clonesrv_t *) args;    zframe_t *identity = zframe_recv (poller->socket);    if (identity) {        //  Request is in second frame of message        char *request = zstr_recv (poller->socket);        char *subtree = NULL;        if (streq (request, "ICANHAZ?")) {            free (request);            subtree = zstr_recv (poller->socket);        }        else            printf ("E: bad request, aborting/n");        if (subtree) {            //  Send state socket to client            kvroute_t routing = { poller->socket, identity, subtree };            zhash_foreach (self->kvmap, s_send_single, &routing);            //  Now send END message with sequence number            zclock_log ("I: sending shapshot=%d", (int) self->sequence);            zframe_send (&identity, poller->socket, ZFRAME_MORE);            kvmsg_t *kvmsg = kvmsg_new (self->sequence);            kvmsg_set_key  (kvmsg, "KTHXBAI");            kvmsg_set_body (kvmsg, (byte *) subtree, 0);            kvmsg_send     (kvmsg, poller->socket);            kvmsg_destroy (&kvmsg);            free (subtree);        }    }    return 0;}

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

intzmailer_msg_send (zmailer_msg_t *self, zsock_t *output){    assert (self);    assert (output);    if (zsock_type (output) == ZMQ_ROUTER)        zframe_send (&self->routing_id, output, ZFRAME_MORE + ZFRAME_REUSE);    size_t frame_size = 2 + 1;          //  Signature and message ID    switch (self->id) {        case ZMAILER_MSG_MAIL:            frame_size += 2;            //  version            frame_size += 1 + strlen (self->from);            frame_size += 4;            if (self->to)                frame_size += strlen (self->to);            frame_size += 4;            if (self->subject)                frame_size += strlen (self->subject);            frame_size += 4;            if (self->request)                frame_size += strlen (self->request);            break;    }    //  Now serialize message into the frame    zmq_msg_t frame;    zmq_msg_init_size (&frame, frame_size);    self->needle = (byte *) zmq_msg_data (&frame);    PUT_NUMBER2 (0xAAA0 | 0);    PUT_NUMBER1 (self->id);    size_t nbr_frames = 1;              //  Total number of frames to send    switch (self->id) {        case ZMAILER_MSG_MAIL:            PUT_NUMBER2 (1);            PUT_STRING (self->from);            if (self->to) {                PUT_LONGSTR (self->to);            }            else                PUT_NUMBER4 (0);    //  Empty string            if (self->subject) {                PUT_LONGSTR (self->subject);            }            else                PUT_NUMBER4 (0);    //  Empty string            if (self->request) {                PUT_LONGSTR (self->request);            }            else                PUT_NUMBER4 (0);    //  Empty string            break;    }    //  Now send the data frame    zmq_msg_send (&frame, zsock_resolve (output), --nbr_frames? ZMQ_SNDMORE: 0);    return 0;}

intzmsg_sendm (zmsg_t **self_p, void *dest){    assert (self_p);    assert (dest);    zmsg_t *self = *self_p;    int rc = 0;    if (self) {        assert (zmsg_is (self));        bool sent_some = false;        zframe_t *frame;        while ((frame = (zframe_t *) zlist_head (self->frames))) {            rc = zframe_send (&frame, dest, ZFRAME_MORE);            if (rc != 0) {                if (errno == EINTR && sent_some)                    continue;                else                    break;            }            sent_some = true;            (void) zlist_pop (self->frames);        }        if (rc == 0)            zmsg_destroy (self_p);    }    return rc;}

intzmsg_send (zmsg_t **self_p, void *dest){    assert (self_p);    assert (dest);    zmsg_t *self = *self_p;    int rc = 0;    void *handle = zsock_resolve (dest);    if (self) {        assert (zmsg_is (self));        if (zlist_size (self->frames) == 0)            return -1;          //  Sending an empty message is an error                zframe_t *frame = (zframe_t *) zlist_pop (self->frames);        while (frame) {            rc = zframe_send (&frame, handle,                zlist_size (self->frames)? ZFRAME_MORE: 0);            if (rc != 0)                break;            frame = (zframe_t *) zlist_pop (self->frames);        }        zmsg_destroy (self_p);    }    return rc;}

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

static voidstate_manager (void *args, zctx_t *ctx, void *pipe){    zhash_t *kvmap = zhash_new ();    zstr_send (pipe, "READY");    void *snapshot = zsocket_new (ctx, ZMQ_ROUTER);    zsocket_bind (snapshot, "tcp://*:5556");    zmq_pollitem_t items [] = {        { pipe, 0, ZMQ_POLLIN, 0 },        { snapshot, 0, ZMQ_POLLIN, 0 }    };    int64_t sequence = 0;       //  Current snapshot version number    while (!zctx_interrupted) {        int rc = zmq_poll (items, 2, -1);        if (rc == -1 && errno == ETERM)            break;              //  Context has been shut down        //  Apply state update from main thread        if (items [0].revents & ZMQ_POLLIN) {            kvmsg_t *kvmsg = kvmsg_recv (pipe);            if (!kvmsg)                break;          //  Interrupted            sequence = kvmsg_sequence (kvmsg);            kvmsg_store (&kvmsg, kvmap);        }        //  Execute state snapshot request        if (items [1].revents & ZMQ_POLLIN) {            zframe_t *identity = zframe_recv (snapshot);            if (!identity)                break;          //  Interrupted            //  Request is in second frame of message            char *request = zstr_recv (snapshot);            if (streq (request, "ICANHAZ?"))                free (request);            else {                printf ("E: bad request, aborting/n");                break;            }            //  Send state snapshot to client            kvroute_t routing = { snapshot, identity };            //  For each entry in kvmap, send kvmsg to client            zhash_foreach (kvmap, s_send_single, &routing);            //  Now send END message with sequence number            printf ("Sending state shapshot=%d/n", (int) sequence);            zframe_send (&identity, snapshot, ZFRAME_MORE);            kvmsg_t *kvmsg = kvmsg_new (sequence);            kvmsg_set_key  (kvmsg, "KTHXBAI");            kvmsg_set_body (kvmsg, (byte *) "", 0);            kvmsg_send     (kvmsg, snapshot);            kvmsg_destroy (&kvmsg);        }    }    zhash_destroy (&kvmap);}

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

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

//  Send one state snapshot key-value pair to a socket//  Hash item data is our kvmsg object, ready to sendstatic ints_send_single (const char *key, void *data, void *args){    kvroute_t *kvroute = (kvroute_t *) args;    //  Send identity of recipient first    zframe_send (&kvroute->identity,        kvroute->socket, ZFRAME_MORE + ZFRAME_REUSE);    kvmsg_t *kvmsg = (kvmsg_t *) data;    kvmsg_send (kvmsg, kvroute->socket);    return 0;}

static ints_agent_handle_control (agent_t *self){    //  Get the whole message off the control socket in one go    zmsg_t *request = zmsg_recv (self->control);    char *command = zmsg_popstr (request);    if (!command)        return -1;                  //  Interrupted    if (streq (command, "SET")) {        char *name = zmsg_popstr (request);        char *value = zmsg_popstr (request);        curve_codec_set_metadata (self->codec, name, value);        free (name);        free (value);    }    else    if (streq (command, "CONNECT")) {        assert (!self->endpoint);        self->endpoint = zmsg_popstr (request);        int rc = zsocket_connect (self->dealer, "%s", self->endpoint);        assert (rc != -1);        zframe_t *server_key = zmsg_pop (request);        zframe_t *output = curve_codec_execute (self->codec, &server_key);        zframe_send (&output, self->dealer, 0);        self->state = connecting;    }    else    if (streq (command, "DISCONNECT")) {        if (self->endpoint) {            int rc = zsocket_disconnect (self->dealer, "%s", self->endpoint);            assert (rc != -1);            free (self->endpoint);        }    }    else    if (streq (command, "VERBOSE")) {        char *verbose = zmsg_popstr (request);        curve_codec_set_verbose (self->codec, *verbose == '1');        free (verbose);    }    else    if (streq (command, "TERMINATE")) {        self->state = terminated;        zstr_send (self->control, "OK");    }    else {        puts ("E: invalid command from API");        assert (false);    }    free (command);    zmsg_destroy (&request);    return 0;}

static voidserver_thread (void *args, zctx_t *ctx, void *pipe){    FILE *file = fopen ("testdata", "r");    assert (file);    void *router = zsocket_new (ctx, ZMQ_ROUTER);    //  Default HWM is 1000, which will drop messages here    //  since we send more than 1,000 chunks of test data,    //  so set an infinite HWM as a simple, stupid solution:    zsocket_set_hwm (router, 0);    zsocket_bind (router, "tcp://*:6000");    while (true) {        //  First frame in each message is the sender identity        zframe_t *identity = zframe_recv (router);        if (!identity)            break;              //  Shutting down, quit                    //  Second frame is "fetch" command        char *command = zstr_recv (router);        assert (streq (command, "fetch"));        free (command);        while (true) {            byte *data = malloc (CHUNK_SIZE);            assert (data);            size_t size = fread (data, 1, CHUNK_SIZE, file);            zframe_t *chunk = zframe_new_zero_copy (                data, size, free_chunk, NULL);            zframe_send (&identity, router, ZFRAME_REUSE + ZFRAME_MORE);            zframe_send (&chunk, router, 0);            if (size == 0)                break;          //  Always end with a zero-size frame        }        zframe_destroy (&identity);    }    fclose (file);}

//  Send one state snapshot key-value pair to a socket//  Hash item data is our kvmsg object, ready to sendstatic ints_send_single (const char *key, void *data, void *args){    kvroute_t *kvroute = (kvroute_t *) args;    kvmsg_t *kvmsg = (kvmsg_t *) data;    if (strlen (kvroute->subtree) <= strlen (kvmsg_key (kvmsg))    &&  memcmp (kvroute->subtree,                kvmsg_key (kvmsg), strlen (kvroute->subtree)) == 0) {        zframe_send (&kvroute->identity,    //  Choose recipient            kvroute->socket, ZFRAME_MORE + ZFRAME_REUSE);        kvmsg_send (kvmsg, kvroute->socket);    }    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 voidcounter_task (void *args, zctx_t *ctx, void *pipe){    void *counter = zsocket_new (ctx, ZMQ_ROUTER);    zsocket_bind (counter, "tcp://*:6001");    //  Parameters for the census    int census_msec = 250;       //  Msecs to settle down    //  Calling thread tells us the population size    char *population = zstr_recv (pipe);        //  All activity happens on our counter socket    zmq_pollitem_t items [] = { { counter, 0, ZMQ_POLLIN, 0 } };    int headcount = 0;         //  Known target size    int positives = 0;         //  How many said "yes"        int64_t timer_end = zclock_time () + census_msec;    int still_waiting = atoi (population);    while (still_waiting) {        int64_t time_left = timer_end - zclock_time ();        if (time_left <= 0)            break;              //  We're done here        int rc = zmq_poll (items, 1, time_left * ZMQ_POLL_MSEC);        if (rc == -1)            break;              //  Interrupted        if (items [0].revents & ZMQ_POLLIN) {            zframe_t *address = zframe_recv (counter);            char *message = zstr_recv (counter);            if (streq (message, "Hello")) {                headcount++;                zframe_send (&address, counter, ZFRAME_MORE);                zstr_send (counter, "Who wants pizza?");            }            else            if (streq (message, "Yes"))                positives++;                        zframe_destroy (&address);            free (message);        }    }    printf ("Out of %d people, %d want pizza/n", headcount, positives);    zstr_send (pipe, "DONE");}

/** * We call this function for each key-value pair in our hash table */static ints_send_single (const char *key, void *data, void *args){    kvroute_t *kvroute = (kvroute_t *) args;    kvmsg_t *kvmsg = (kvmsg_t *) data;    DEBUG ("I: s_send_single %"PRId64" type:%s", kvmsg_sequence(kvmsg), kvmsg_get_prop (kvmsg, "type") );    if (strlen (kvroute->subtree) <= strlen (kvmsg_key (kvmsg))    &&  memcmp (kvroute->subtree,                kvmsg_key (kvmsg), strlen (kvroute->subtree)) == 0) {        zframe_send (&kvroute->identity,    //  Choose recipient            kvroute->socket, ZFRAME_MORE + ZFRAME_REUSE);        kvmsg_send (kvmsg, kvroute->socket);    }        return 0;}

voidzmsg_send (zmsg_t **self_p, void *socket){    assert (self_p);    assert (socket);    zmsg_t *self = *self_p;    if (self) {        zframe_t *frame = (zframe_t *) zlist_pop (self->frames);        while (frame) {            zframe_send (&frame, socket,                zlist_size (self->frames)? ZFRAME_MORE: 0);            frame = (zframe_t *) zlist_pop (self->frames);        }        zmsg_destroy (self_p);    }}

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 void worker_task (void *args, zctx_t *ctx, void *pipe){    // Send out heartbeats at regular intervals    client_state* state = (client_state*) args;    void *worker = zsocket_new (ctx, ZMQ_DEALER);    zsocket_connect (worker, "inproc://backend");        while (true) {        ECHO("please set input");                zframe_t* content = get_frame_from_stdin(state);                zclock_sleep (randof (1000) + 1);                zframe_send (&content, worker, ZFRAME_REUSE);        zframe_destroy (&content);    }}

int main (void){    zctx_t *ctx = zctx_new ();    void *worker = zsocket_new (ctx, ZMQ_REQ);    //  Set random identity to make tracing easier    srandom ((unsigned) time (NULL));    char identity [10];    sprintf (identity, "%04X-%04X", randof (0x10000), randof (0x10000));    zmq_setsockopt (worker, ZMQ_IDENTITY, identity, strlen (identity));    zsocket_connect (worker, "tcp://localhost:5556");    //  Tell broker we're ready for work    printf ("I: (%s) worker ready/n", identity);    zframe_t *frame = zframe_new (LRU_READY, 1);    zframe_send (&frame, worker, 0);    int cycles = 0;    while (true) {        zmsg_t *msg = zmsg_recv (worker);        if (!msg)            break;              //  Interrupted        //  Simulate various problems, after a few cycles        cycles++;        if (cycles > 3 && randof (5) == 0) {            printf ("I: (%s) simulating a crash/n", identity);            zmsg_destroy (&msg);            break;        }        else        if (cycles > 3 && randof (5) == 0) {            printf ("I: (%s) simulating CPU overload/n", identity);            sleep (3);            if (zctx_interrupted)                break;        }        printf ("I: (%s) normal reply/n", identity);        sleep (1);              //  Do some heavy work        zmsg_send (&msg, worker);    }    zctx_destroy (&ctx);    return 0;}