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

static void *worker_task (void *args){    zctx_t *ctx = zctx_new ();    void *worker = zsocket_new (ctx, ZMQ_DEALER);    zsocket_connect (worker, "tcp://localhost:5556");        while (true) {        zmsg_t *msg = zmsg_recv (worker);        zmsg_send (&msg, worker);    }    zctx_destroy (&ctx);    return NULL;}

voidnom_server_destroy (nom_server_t **self_p){    assert (self_p);    if (*self_p) {        nom_server_t *self = *self_p;        zstr_send (self->pipe, "STOP");        char *string = zstr_recv (self->pipe);        free (string);        zctx_destroy (&self->ctx);        free (self);        *self_p = NULL;    }}

//  Thread shim for UNIX calls the real thread and cleans up afterwards.static void *s_thread_shim (void *args){    assert (args);    shim_t *shim = (shim_t *) args;    if (shim->attached)        shim->attached (shim->args, shim->ctx, shim->pipe);    else        shim->detached (shim->args);    zctx_destroy (&shim->ctx);    free (shim);    return NULL;}

void kosmonaut_destroy(kosmonaut_t** self_p){	if (!*self_p)		return;  	kosmonaut_t* self = *self_p;  	kosmonaut_stop_listening(self);	zctx_destroy(&self->ctx);	pthread_mutex_destroy(&self->lmtx);	pthread_mutex_destroy(&self->tmtx);	free(self);	*self_p = NULL;}

voidfmq_client_destroy (fmq_client_t **self_p){    assert (self_p);    if (*self_p) {        fmq_client_t *self = *self_p;        zstr_send (self->pipe, "STOP");        char *string = zstr_recv (self->pipe);        free (string);        zctx_destroy (&self->ctx);        free (self);        *self_p = NULL;    }}

static voids_broker_destroy (broker_t **self_p){    assert (self_p);    if (*self_p) {        broker_t *self = *self_p;        zctx_destroy (&self->ctx);        zhash_destroy (&self->services);        zhash_destroy (&self->workers);        zlist_destroy (&self->waiting);        free (self);        *self_p = NULL;    }}

voidzmonitor_test (bool verbose){    printf (" * zmonitor: ");    if (verbose)        printf ("/n");    //  @selftest    zctx_t *ctx = zctx_new ();    bool result;    void *sink = zsocket_new (ctx, ZMQ_PULL);    zmonitor_t *sinkmon = zmonitor_new (ctx,        sink, ZMQ_EVENT_LISTENING | ZMQ_EVENT_ACCEPTED);    zmonitor_set_verbose (sinkmon, verbose);    //  Check sink is now listening    int port_nbr = zsocket_bind (sink, "tcp://127.0.0.1:*");    assert (port_nbr != -1);    result = s_check_event (sinkmon, ZMQ_EVENT_LISTENING);    assert (result);    void *source = zsocket_new (ctx, ZMQ_PUSH);    zmonitor_t *sourcemon = zmonitor_new (ctx,        source, ZMQ_EVENT_CONNECTED | ZMQ_EVENT_DISCONNECTED);    zmonitor_set_verbose (sourcemon, verbose);    zsocket_connect (source, "tcp://127.0.0.1:%d", port_nbr);    //  Check source connected to sink    result = s_check_event (sourcemon, ZMQ_EVENT_CONNECTED);    assert (result);    //  Check sink accepted connection    result = s_check_event (sinkmon, ZMQ_EVENT_ACCEPTED);    assert (result);    //  Destroy sink to trigger a disconnect event on the source    //  PH: disabled since this causes an access violation in    //  zmonitor_destroy as the socket is no longer valid.//     zsocket_destroy (ctx, sink);//     result = s_check_event (sourcemon, ZMQ_EVENT_DISCONNECTED);//     assert (result);    zmonitor_destroy (&sinkmon);    zmonitor_destroy (&sourcemon);    zctx_destroy (&ctx);    //  @end    printf ("OK/n");}

intzre_log_msg_test (bool verbose){    printf (" * zre_log_msg: ");    //  @selftest    //  Simple create/destroy test    zre_log_msg_t *self = zre_log_msg_new (0);    assert (self);    zre_log_msg_destroy (&self);    //  Create pair of sockets we can send through    zctx_t *ctx = zctx_new ();    assert (ctx);    void *output = zsocket_new (ctx, ZMQ_DEALER);    assert (output);    zsocket_bind (output, "inproc://selftest");    void *input = zsocket_new (ctx, ZMQ_ROUTER);    assert (input);    zsocket_connect (input, "inproc://selftest");        //  Encode/send/decode and verify each message type    self = zre_log_msg_new (ZRE_LOG_MSG_LOG);    zre_log_msg_set_level (self, 123);    zre_log_msg_set_event (self, 123);    zre_log_msg_set_node (self, 123);    zre_log_msg_set_peer (self, 123);    zre_log_msg_set_time (self, 123);    zre_log_msg_set_data (self, "Life is short but Now lasts for ever");    zre_log_msg_send (&self, output);        self = zre_log_msg_recv (input);    assert (self);    assert (zre_log_msg_level (self) == 123);    assert (zre_log_msg_event (self) == 123);    assert (zre_log_msg_node (self) == 123);    assert (zre_log_msg_peer (self) == 123);    assert (zre_log_msg_time (self) == 123);    assert (streq (zre_log_msg_data (self), "Life is short but Now lasts for ever"));    zre_log_msg_destroy (&self);    zctx_destroy (&ctx);    //  @end    printf ("OK/n");    return 0;}

int main(int argc, char *argv[]){    char *bind = strdup("tcp://127.0.0.1:12345");    size_t msg_size = 1024;    int msg_count = 100;    int perf_type = THROUGHPUT;    int opt;    while ((opt = getopt(argc, argv, "b:s:c:t:h")) != -1) {        switch (opt) {        case 'b':            free(bind);            bind = strdup(optarg);            break;        case 's':            msg_size = atoi(optarg);            break;        case 'c':            msg_count = atoi(optarg);            break;        case 't':            perf_type = atoi(optarg);            break;        case 'h':            usage();            exit(EXIT_SUCCESS);        default:            fprintf(stderr, "Try 'nanomsg_req_perf -h' for more information/n");            exit(EXIT_FAILURE);        }    }    thr_info *info = (thr_info *) malloc(sizeof(thr_info));    info->bind_to = bind;    info->msg_size = msg_size + 1;    info->msg_count = msg_count;    info->perf_type = perf_type;    zthread_new(server_task, info);    zctx_t *ctx = zctx_new();    void *client = zthread_fork(ctx, client_task, info);    char *signal = zstr_recv(client);    assert(strcmp(signal, "done") == 0);    free(signal);    zctx_destroy(&ctx);    return 0;}

//  @selfteststatic void *s_test_detached (void *args){	void *push;	zctx_t *ctx;    //  Create a socket to check it'll be automatically deleted    ctx = zctx_new ();    assert (ctx);    push = zsocket_new (ctx, ZMQ_PUSH);    assert (push);    zctx_destroy (&ctx);    return NULL;}

//  Thread shim for Windows that wraps a POSIX-style thread handler//  and does the _endthreadex for us automatically.unsigned __stdcalls_thread_shim (void *args){    assert (args);    shim_t *shim = (shim_t *) args;    if (shim->attached)        shim->attached (shim->args, shim->ctx, shim->pipe);    else        shim->detached (shim->args);    zctx_destroy (&shim->ctx);  //  Close any dangling sockets    free (shim);    _endthreadex (0);           //  Terminates thread    return 0;}

int main (void){    //  Create threads    zctx_t *ctx = zctx_new ();    void *client = zthread_fork (ctx, client_task, NULL);    zthread_new (worker_task, NULL);    zthread_new (broker_task, NULL);    //  Wait for signal on client pipe    char *signal = zstr_recv (client);    free (signal);    zctx_destroy (&ctx);    return 0;}

intzctx_test (bool verbose){    printf (" * zctx: ");    //  @selftest    //  Create and destroy a context without using it    zctx_t *ctx = zctx_new ();    assert (ctx);    zctx_destroy (&ctx);    assert (ctx == NULL);    //  Create a context with many busy sockets, destroy it    ctx = zctx_new ();    assert (ctx);    zctx_set_iothreads (ctx, 1);    zctx_set_linger (ctx, 5);       //  5 msecs    void *s1 = zctx__socket_new (ctx, ZMQ_PAIR);    void *s2 = zctx__socket_new (ctx, ZMQ_XREQ);    void *s3 = zctx__socket_new (ctx, ZMQ_REQ);    void *s4 = zctx__socket_new (ctx, ZMQ_REP);    void *s5 = zctx__socket_new (ctx, ZMQ_PUB);    void *s6 = zctx__socket_new (ctx, ZMQ_SUB);    zsocket_connect (s1, "tcp://127.0.0.1:5555");    zsocket_connect (s2, "tcp://127.0.0.1:5555");    zsocket_connect (s3, "tcp://127.0.0.1:5555");    zsocket_connect (s4, "tcp://127.0.0.1:5555");    zsocket_connect (s5, "tcp://127.0.0.1:5555");    zsocket_connect (s6, "tcp://127.0.0.1:5555");    assert (zctx_underlying (ctx));    zctx_destroy (&ctx);    //  @end    printf ("OK/n");    return 0;}

voiddrops_destroy (drops_t **self_p){    assert (self_p);    if (*self_p) {        drops_t *self = *self_p;        if (self->pipe) {            zstr_send (self->pipe, "TERMINATE");            char *reply = zstr_recv (self->pipe);            zstr_free (&reply);        }        zctx_destroy (&self->ctx);        free (self);        *self_p = NULL;    }}

void rrwrk_destroy(rrwrk_t **self_p){    assert(self_p);    if (*self_p) {        rrwrk_t *self = *self_p;        zctx_destroy(&self->ctx); //** Also destroy all the sockets created in this context        rrwrk_data_destroy(self);        zlist_destroy(&self->data);        free(self->pattern);        free(self->sinker);        free(self->broker);        free(self->uuid_str);        free(self);        *self_p = NULL;    }}

int main (int argc, char *argv []){    int verbose = (argc > 1 && streq (argv [1], "-v"));    zctx_t *ctx = zctx_new ();    //  Prepare server socket with predictable identity    char *bind_endpoint = "tcp://*:5555";    char *connect_endpoint = "tcp://localhost:5555";    void *server = zsocket_new (ctx, ZMQ_ROUTER);    zmq_setsockopt (server,        ZMQ_IDENTITY, connect_endpoint, strlen (connect_endpoint));    zsocket_bind (server, bind_endpoint);    printf ("I: service is ready at %s/n", bind_endpoint);    while (!zctx_interrupted) {        zmsg_t *request = zmsg_recv (server);        if (verbose && request)            zmsg_dump (request);        if (!request)            break;          //  Interrupted        //  Frame 0: identity of client        //  Frame 1: PING, or client control frame        //  Frame 2: request body        zframe_t *identity = zmsg_pop (request);        zframe_t *control = zmsg_pop (request);        zmsg_t *reply = zmsg_new ();        if (zframe_streq (control, "PING"))            zmsg_addstr (reply, "PONG");        else {            zmsg_add (reply, control);            zmsg_addstr (reply, "OK");        }        zmsg_destroy (&request);        zmsg_push (reply, identity);        if (verbose && reply)            zmsg_dump (reply);        zmsg_send (&reply, server);    }    if (zctx_interrupted)        printf ("W: interrupted/n");    zctx_destroy (&ctx);    return 0;}

/** * Deconstruct *   This destroys the context and any associated sockets */BoomStick::~BoomStick() {   if (mCtx != nullptr) {      zctx_destroy(&mCtx);   }   if (!mPendingReplies.empty()) {      LOG(WARNING) << "Pending replies never emptied " << mPendingReplies.size();      for (auto reply : mPendingReplies) {         LOG(WARNING) << reply.first;      }   }   if (!mUnreadReplies.empty()) {      LOG(WARNING) << "mUnreadReplies replies never emptied " << mUnreadReplies.size();      for (auto reply : mUnreadReplies) {         LOG(WARNING) << reply.first;      }   }}

int main (int argc, char *argv []){    zctx_t *ctx = zctx_new ();    //  Use the CZMQ zbeacon class to make sure we listen on the    //  same network interface as our peers    void *collector = zsocket_new (ctx, ZMQ_SUB);    zbeacon_t *beacon = zbeacon_new (ZRE_DISCOVERY_PORT);    char *host = zbeacon_hostname (beacon);    //  Bind to an ephemeral port    int port = zsocket_bind (collector, "tcp://%s:*", host);    //  Announce this to all peers we connect to    zre_node_t *node = zre_node_new ();    zre_node_header_set (node, "X-ZRELOG", "tcp://%s:%d", host, port);    //  Get all log messages (don't filter)    zsocket_set_subscribe (collector, "");    zmq_pollitem_t pollitems [] = {        { collector, 0, ZMQ_POLLIN, 0 },        { zre_node_handle (node), 0, ZMQ_POLLIN, 0 }    };    while (!zctx_interrupted) {        if (zmq_poll (pollitems, 2, 1000 * ZMQ_POLL_MSEC) == -1)            break;              //  Interrupted        //  Handle input on collector        if (pollitems [0].revents & ZMQ_POLLIN)            s_print_log_msg (collector);        //  Handle event from node (ignore it)        if (pollitems [1].revents & ZMQ_POLLIN) {            zmsg_t *msg = zre_node_recv (node);            if (!msg)                break;              //  Interrupted            zmsg_destroy (&msg);        }    }    zre_node_destroy (&node);    zbeacon_destroy (&beacon);    zctx_destroy (&ctx);    return 0;}

int main (int argn, char *argv []) {    if (argn < 2) {        puts ("syntax: ./zyrechat myname");        exit (0);    }    zctx_t *ctx = zctx_new ();    void *chat_pipe = zthread_fork (ctx, chat_task, NULL);    while (!zctx_interrupted) {        char message [1024];        if (!fgets (message, 1024, stdin))            break;        zstr_sendf (chat_pipe, "%s:%s", argv [1], message);    }    zctx_destroy (&ctx);    return 0;}

int main (void){    zctx_t *ctx = zctx_new ();    void *subscriber = zsocket_new (ctx, ZMQ_SUB);    zsocket_connect (subscriber, "tcp://localhost:30001");    zsocket_set_subscribe (subscriber, "");        while (true) {        zmsg_t *msg = zmsg_recv (subscriber);        if (!msg)            break;        zmsg_dump (msg);        zmsg_destroy (&msg);    }    zctx_destroy (&ctx);    return 0;}

char *  test_line() {  zctx_t *context = zctx_new();  lineconfig_t config;  config.line_id = 2;  config_t base = { "uuid", "inproc:broker", "inproc:portwatcher",                         "inproc:registration", "http://something.com" };  config.base_config = &base;  void * line_in = zsocket_new(context, ZMQ_PUB);  zsocket_bind (line_in, "inproc://line");  // filter(identity, serial_reader, remote, context);    zclock_log("pre-fork");  void * pipe = zthread_fork(context, line_listener, &config);  // TODO a real test  zstr_send(pipe, "ping");  char * result = zstr_recv(pipe);  mu_assert_str("pong response", "pong", result);  free(result);  // do some channel switching  int i,j;  for(i=0;i<100;i++) {    // first channel    for(j=0;j<3; j++) {      val_msg(line_in, "foo", 12);    }    for(j=0;j<3;j++) {      val_msg(line_in, "bar", 99);    }  }  zstr_send(line_in, "DESTROY");  // temp  printf("destroy message sent/n");  result = zstr_recv(pipe);  mu_assert_str("destroy response", "ok", result);  free(result);  zsocket_destroy(context,pipe);  printf("freed pipe/n");  zsocket_destroy(context, line_in);  zctx_destroy(&context);  return NULL;}

intzsocket_test (Bool verbose){    printf (" * zsocket: ");    //  @selftest    zctx_t *ctx = zctx_new ();    assert (ctx);    //  Create a detached thread, let it run    char *interf = "*";    char *domain = "localhost";    int service = 5560;    void *writer = zsocket_new (ctx, ZMQ_PUSH);    assert (writer);    void *reader = zsocket_new (ctx, ZMQ_PULL);    assert (reader);    assert (streq (zsocket_type_str (writer), "PUSH"));    assert (streq (zsocket_type_str (reader), "PULL"));    int rc = zsocket_bind (writer, "tcp://%s:%d", interf, service);    assert (rc == service);    rc = zsocket_connect (reader, "tcp://%s:%d", domain, service);    assert (rc == 0);    zstr_send (writer, "HELLO");    char *message = zstr_recv (reader);    assert (message);    assert (streq (message, "HELLO"));    free (message);    int port = zsocket_bind (writer, "tcp://%s:*", interf);    assert (port >= ZSOCKET_DYNFROM && port <= ZSOCKET_DYNTO);    assert (zsocket_poll (writer, 100) == FALSE);    rc = zsocket_connect (reader, "txp://%s:%d", domain, service);    assert (rc == -1);    zsocket_destroy (ctx, writer);    zctx_destroy (&ctx);    //  @end    printf ("OK/n");    return 0;}

static void client_task (void *args){    client_state* state = (client_state*) args;    void *frontend = zsocket_new (state->context, ZMQ_DEALER);    zsocket_set_identity (frontend, state->identity);    zsocket_connect (frontend,state->server_url);        void *backend = zsocket_new (state->context, ZMQ_DEALER);    zsocket_bind (backend, "inproc://backend");    zthread_fork (state->context, worker_task, state);        client_loop(state, frontend, backend);        zsocket_destroy(state->context, &frontend);    zsocket_destroy(state->context, &backend);    zctx_destroy (&state->context);}

//  Request-reply client using REQ socket//  To simulate load, clients issue a burst of requests and then//  sleep for a random period.//static void *client_task (void *args){    zctx_t *ctx = zctx_new ();    void *client = zsocket_new (ctx, ZMQ_REQ);    zsocket_connect (client, "ipc://%s-localfe.ipc", self);    void *monitor = zsocket_new (ctx, ZMQ_PUSH);    zsocket_connect (monitor, "ipc://%s-monitor.ipc", self);    while (1) {        sleep (randof (5));        int burst = randof (15);        while (burst--) {            char task_id [5];            sprintf (task_id, "%04X", randof (0x10000));            //  Send request with random hex ID            zstr_send (client, task_id);            //  Wait max ten seconds for a reply, then complain            zmq_pollitem_t pollset [1] = { { client, 0, ZMQ_POLLIN, 0 } };            int rc = zmq_poll (pollset, 1, 10 * 1000 * ZMQ_POLL_MSEC);            if (rc == -1)                break;          //  Interrupted            if (pollset [0].revents & ZMQ_POLLIN) {                char *reply = zstr_recv (client);                if (!reply)                    break;              //  Interrupted                //  Worker is supposed to answer us with our task id                puts (reply);                assert (streq (reply, task_id));                free (reply);            }            else {                zstr_sendf (monitor,                            "E: CLIENT EXIT - lost task %s", task_id);                return NULL;            }        }    }    zctx_destroy (&ctx);    return NULL;}

int main (int argc, char *argv []){    //  Initialize context for talking to tasks    zctx_t *ctx = zctx_new ();    zctx_set_linger (ctx, 100);    //  Get number of nodes to simulate, default 100    int max_node = 100;    int nbr_nodes = 0;    int max_iterations = -1;    int nbr_iterations = 0;    if (argc > 1)        max_node = atoi (argv [1]);    if (argc > 2)        max_iterations = atoi (argv [2]);    //  We address nodes as an array of pipes    void **pipes = zmalloc (sizeof (void *) * max_node);    //  We will randomly start and stop node threads    while (!zctx_interrupted) {        uint index = randof (max_node);        //  Toggle node thread        if (pipes [index]) {            zstr_send (pipes [index], "STOP");            zsocket_destroy (ctx, pipes [index]);            pipes [index] = NULL;            zclock_log ("I: Stopped node (%d running)", --nbr_nodes);        }        else {            pipes [index] = zthread_fork (ctx, node_task, NULL);            zclock_log ("I: Started node (%d running)", ++nbr_nodes);        }        nbr_iterations++;        if (max_iterations > 0 && nbr_iterations >= max_iterations)            break;        //  Sleep ~750 msecs randomly so we smooth out activity        zclock_sleep (randof (500) + 500);    }    zclock_log ("I: Stopped tester (%d iterations)", nbr_iterations);    zctx_destroy (&ctx);    free (pipes);    return 0;}

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

voidzyre_log_test (bool verbose){    printf (" * zyre_log: ");    //  @selftest    zctx_t *ctx = zctx_new ();    //  Get all incoming log messages    void *collector = zsocket_new (ctx, ZMQ_SUB);    zsocket_bind (collector, "tcp://127.0.0.1:5555");    zsocket_set_subscribe (collector, "");    //  Create a log instance to send log messages    zyre_log_t *log = zyre_log_new (ctx, "this is me");    zyre_log_connect (log, "tcp://127.0.0.1:5555");    //  Workaround for issue 270; give time for connect to    //  happen and subscriptions to go to pub socket; 200    //  msec should be enough for under valgrind on a slow PC    zpoller_t *poller = zpoller_new (collector, NULL);    zpoller_wait (poller, 200);    //  Send some messages    zyre_log_info (log, ZRE_LOG_MSG_EVENT_JOIN, NULL, "this is you");    zyre_log_info (log, ZRE_LOG_MSG_EVENT_EXIT, "Pizza time", "this is you");    zyre_log_warning (log, "this is you", "Time flies like an %s", "arrow");    zyre_log_error (log, "this is you", "Fruit flies like a %s", "banana");    int count = 0;    while (count < 4) {        zre_log_msg_t *msg = zre_log_msg_recv (collector);        assert (msg);        if (verbose)            zre_log_msg_dump (msg);        zre_log_msg_destroy (&msg);        count++;    }    zpoller_destroy (&poller);    zyre_log_destroy (&log);    zctx_destroy (&ctx);    //  @end    printf ("OK/n");}

int main (void){    //  Start child threads    zctx_t *ctx = zctx_new ();    zthread_fork (ctx, publisher_thread, NULL);    zthread_fork (ctx, subscriber_thread, NULL);    void *subscriber = zsocket_new (ctx, ZMQ_XSUB);    zsocket_connect (subscriber, "tcp://localhost:6000");    void *publisher = zsocket_new (ctx, ZMQ_XPUB);    zsocket_bind (publisher, "tcp://*:6001");    void *listener = zthread_fork (ctx, listener_thread, NULL);    zmq_proxy (subscriber, publisher, listener);    puts (" interrupted");    //  Tell attached threads to exit    zctx_destroy (&ctx);    return 0;}

int main(void){	zctx_t *ctx = zctx_new();	lbbroker_t *self = (lbbroker_t *)zmalloc(sizeof(lbbroker_t));	self->frontend = zsocket_new(ctx, ZMQ_ROUTER);	self->backend = zsocket_new(ctx, ZMQ_ROUTER);#if (defined (WIN32))	zsocket_bind(self->frontend, "tcp://*:5672"); // frontend	zsocket_bind(self->backend, "tcp://*:5673"); // backend#else	zsocket_bind(self->frontend, "ipc://frontend.ipc");	zsocket_bind(self->backend, "ipc://backend.ipc");#endif	int client_nbr;	for (client_nbr = 0; client_nbr < NBR_CLIENTS; client_nbr++)		zthread_new(client_task, NULL);	int worker_nbr;	for (worker_nbr = 0; worker_nbr < NBR_WORKERS; worker_nbr++)		zthread_new(worker_task, NULL);	// Queue of available workers	self->workers = zlist_new();	// Prepare reactor and fire it up	zloop_t *reactor = zloop_new();	zmq_pollitem_t poller = { self->backend, 0, ZMQ_POLLIN };	zloop_poller(reactor, &poller, s_handle_backend, self);	zloop_start(reactor);	zloop_destroy(&reactor);	// When we're done, clean up properly	while (zlist_size(self->workers)) {		zframe_t *frame = (zframe_t *)zlist_pop(self->workers);		zframe_destroy(&frame);	}	zlist_destroy(&self->workers);	zctx_destroy(&ctx);	free(self);	return 0;}

int main (void){  zctx_t *ctx = zctx_new ();  void *dealer = zsocket_new (ctx, ZMQ_DEALER);  zsocket_connect (dealer, "tcp://127.0.0.1:6000");  // We'll allow up to N chunks in transit at once  size_t credit = PIPELINE;  size_t total = 0;     // Total bytes received  size_t chunks = 0;    // Total chunks received  size_t offset = 0;    // Offset of next chunk request  size_t offset_expc = 0;    while (true) {    while (credit) {      // Ask for next chunk      zstr_sendfm (dealer, "fetch");      zstr_sendfm (dealer, "%ld", offset);      zstr_sendf (dealer, "%ld", CHUNK_SIZE);      offset += CHUNK_SIZE;      credit--;    }    zframe_t *chunk = zframe_recv (dealer);    if (!chunk)      break; // Shutting down, quit        chunks++;    credit++;        size_t size = zframe_size (chunk);    zframe_destroy (&chunk);    total += size;    if (size < CHUNK_SIZE)      break; // Last chunk received; exit  }  printf ("%zd chunks received, %zd bytes/n", chunks, total);    zctx_destroy (&ctx);  return total == 102400? 0: -1;}