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

int main (void){    puts ("Starting the zchat server");    zsock_t *responder = zsock_new_rep ("tcp://*:5555");    assert (responder);    zsock_t *publisher = zsock_new_pub ("tcp://*:5556");    assert (publisher);    while (!zsys_interrupted) {        //  Receive messages from client.        char *client_msg = zstr_recv (responder);        if (!client_msg)            break;                //  Let the zchat client know we got it.        zstr_send (responder, "OK");        //  Publish message to all zchat client subscribers        zstr_send (publisher, client_msg);        zstr_free (&client_msg);    }    puts ("Stopping the zchat server");    zsock_destroy (&publisher);    zsock_destroy (&responder);    return 0;}

int main (int argc, char *argv[]){    int rc;    zsock_t *zs;    pthread_t tid;    pthread_attr_t attr;    flux_msg_t *msg;    flux_sec_t *sec;    int n;    log_init (basename (argv[0]));    if (argc != 1 && argc != 2) {        fprintf (stderr, "Usage: tmunge [--fake]/n");        exit (1);    }    if (argc == 2) {        if (!strcmp (argv[1], "--fake"))            sec_typemask |= FLUX_SEC_FAKEMUNGE;        else            log_msg_exit ("unknown option %s", argv[1]);    }    if (!(sec = flux_sec_create (sec_typemask, NULL)))        log_err_exit ("flux_sec_create");    if (flux_sec_comms_init (sec) < 0)        log_err_exit ("flux_sec_comms_init: %s", flux_sec_errstr (sec));    if (!(zs = zsock_new_sub (uri, "")))        log_err_exit ("S: zsock_new_sub");    if (!(cs = zsock_new_pub (uri)))        log_err_exit ("S: zsock_new_pub");    if ((rc = pthread_attr_init (&attr)))        log_errn (rc, "S: pthread_attr_init");    if ((rc = pthread_create (&tid, &attr, thread, NULL)))        log_errn (rc, "S: pthread_create");    /* Handle one client message.     */    if (!(msg = flux_msg_recvzsock_munge (zs, sec)))        log_err_exit ("S: flux_msg_recvzsock_munge: %s", flux_sec_errstr (sec));    //zmsg_dump (zmsg);    if ((n = flux_msg_frames (msg)) != 4)        log_err_exit ("S: expected 4 frames, got %d", n);    flux_msg_destroy (msg);    /* Wait for thread to terminate, then clean up.     */    if ((rc = pthread_join (tid, NULL)))        log_errn (rc, "S: pthread_join");    zsock_destroy (&zs);    zsock_destroy (&cs);    flux_sec_destroy (sec);    log_fini ();    return 0;}

//  Checks whether client can connect to serverstatic bools_can_connect (zsock_t **server, zsock_t **client){    int port_nbr = zsock_bind (*server, "tcp://127.0.0.1:*");    assert (port_nbr > 0);    int rc = zsock_connect (*client, "tcp://127.0.0.1:%d", port_nbr);    assert (rc == 0);    //  Give the connection time to fail if that's the plan    zclock_sleep (200);    //  By default PUSH sockets block if there's no peer    zsock_set_sndtimeo (*server, 200);    zstr_send (*server, "Hello, World");    zpoller_t *poller = zpoller_new (*client, NULL);    assert (poller);    bool success = (zpoller_wait (poller, 400) == *client);    zpoller_destroy (&poller);    zsock_destroy (client);    zsock_destroy (server);    *server = zsock_new (ZMQ_PUSH);    assert (*server);    *client = zsock_new (ZMQ_PULL);    assert (*client);    return success;}

staticvoid subscriber_state_destroy(subscriber_state_t **state_p){    subscriber_state_t *state = *state_p;    zsock_destroy(&state->sub_socket);    zsock_destroy(&state->push_socket);    device_tracker_destroy(&state->tracker);    *state_p = NULL;}

voidzmailer_msg_test (bool verbose){    printf (" * zmailer_msg:");    if (verbose)        printf ("/n");    //  @selftest    //  Simple create/destroy test    zmailer_msg_t *self = zmailer_msg_new ();    assert (self);    zmailer_msg_destroy (&self);    //  Create pair of sockets we can send through    //  We must bind before connect if we wish to remain compatible with ZeroMQ < v4    zsock_t *output = zsock_new (ZMQ_DEALER);    assert (output);    int rc = zsock_bind (output, "inproc://selftest-zmailer_msg");    assert (rc == 0);    zsock_t *input = zsock_new (ZMQ_ROUTER);    assert (input);    rc = zsock_connect (input, "inproc://selftest-zmailer_msg");    assert (rc == 0);    //  Encode/send/decode and verify each message type    int instance;    self = zmailer_msg_new ();    zmailer_msg_set_id (self, ZMAILER_MSG_MAIL);    zmailer_msg_set_from (self, "Life is short but Now lasts for ever");    zmailer_msg_set_to (self, "Life is short but Now lasts for ever");    zmailer_msg_set_subject (self, " test subject ");    zmailer_msg_set_request (self, " this is the text to be sent ");    //  Send twice    zmailer_msg_send (self, output);    zmailer_msg_send (self, output);    for (instance = 0; instance < 2; instance++) {        zmailer_msg_recv (self, input);        assert (zmailer_msg_routing_id (self));        assert (streq (zmailer_msg_from (self), "Life is short but Now lasts for ever"));        assert (streq (zmailer_msg_to (self), "Life is short but Now lasts for ever"));        assert (streq (zmailer_msg_subject (self), " test subject "));        assert (streq (zmailer_msg_request (self), " this is the text to be sent "));    }    zmailer_msg_destroy (&self);    zsock_destroy (&input);    zsock_destroy (&output);    //  @end    printf ("OK/n");}

staticvoid parser_state_destroy(parser_state_t **state_p){    parser_state_t *state = *state_p;    // must not destroy the pipe, as it's owned by the actor    zsock_destroy(&state->pull_socket);    zsock_destroy(&state->push_socket);    zchunk_destroy(&state->decompression_buffer);    free(state);    *state_p = NULL;}

voidzchanneler_destroy (zchanneler_t **self_p){    assert (self_p);    if (*self_p) {        zchanneler_t *self = *self_p;        zsock_destroy (&self->client);        zsock_destroy (&self->server);        free (self);        *self_p = NULL;    }}

voidzmonitor_test (bool verbose){    printf (" * zmonitor: ");    if (verbose)        printf ("/n");#if defined (ZMQ_EVENT_ALL)    //  @selftest    zsock_t *client = zsock_new (ZMQ_DEALER);    assert (client);    zactor_t *clientmon = zactor_new (zmonitor, client);    assert (clientmon);    if (verbose)        zstr_sendx (clientmon, "VERBOSE", NULL);    zstr_sendx (clientmon, "LISTEN", "LISTENING", "ACCEPTED", NULL);    zstr_sendx (clientmon, "START", NULL);    zsock_wait (clientmon);    zsock_t *server = zsock_new (ZMQ_DEALER);    assert (server);    zactor_t *servermon = zactor_new (zmonitor, server);    assert (servermon);    if (verbose)        zstr_sendx (servermon, "VERBOSE", NULL);    zstr_sendx (servermon, "LISTEN", "CONNECTED", "DISCONNECTED", NULL);    zstr_sendx (servermon, "START", NULL);    zsock_wait (servermon);    //  Allow a brief time for the message to get there...    zmq_poll (NULL, 0, 200);    //  Check client is now listening    int port_nbr = zsock_bind (client, "tcp://127.0.0.1:*");    assert (port_nbr != -1);    s_assert_event (clientmon, "LISTENING");    //  Check server connected to client    zsock_connect (server, "tcp://127.0.0.1:%d", port_nbr);    s_assert_event (servermon, "CONNECTED");    //  Check client accepted connection    s_assert_event (clientmon, "ACCEPTED");    zactor_destroy (&clientmon);    zactor_destroy (&servermon);    zsock_destroy (&client);    zsock_destroy (&server);#endif    //  @end    printf ("OK/n");}

static voids_self_destroy (self_t **self_p){    assert (self_p);    if (*self_p) {        self_t *self = *self_p;        zsock_destroy (&self->frontend);        zsock_destroy (&self->backend);        zsock_destroy (&self->capture);        zpoller_destroy (&self->poller);        free (self);        *self_p = NULL;    }}

voidzloop_test (bool verbose){    printf (" * zloop: ");    int rc = 0;    //  @selftest    //  Create two PAIR sockets and connect over inproc    zsock_t *output = zsock_new (ZMQ_PAIR);    assert (output);    zsock_bind (output, "inproc://zloop.test");    zsock_t *input = zsock_new (ZMQ_PAIR);    assert (input);    zsock_connect (input, "inproc://zloop.test");    zloop_t *loop = zloop_new ();    assert (loop);    zloop_set_verbose (loop, verbose);    //  Create a timer that will be cancelled    int timer_id = zloop_timer (loop, 1000, 1, s_timer_event, NULL);    zloop_timer (loop, 5, 1, s_cancel_timer_event, &timer_id);    //  After 20 msecs, send a ping message to output3    zloop_timer (loop, 20, 1, s_timer_event, output);    //  Set up some tickets that will never expire    zloop_set_ticket_delay (loop, 10000);    void *ticket1 = zloop_ticket (loop, s_timer_event, NULL);    void *ticket2 = zloop_ticket (loop, s_timer_event, NULL);    void *ticket3 = zloop_ticket (loop, s_timer_event, NULL);    //  When we get the ping message, end the reactor    rc = zloop_reader (loop, input, s_socket_event, NULL);    assert (rc == 0);    zloop_reader_set_tolerant (loop, input);    zloop_start (loop);    zloop_ticket_delete (loop, ticket1);    zloop_ticket_delete (loop, ticket2);    zloop_ticket_delete (loop, ticket3);    zloop_destroy (&loop);    assert (loop == NULL);    zsock_destroy (&input);    zsock_destroy (&output);    //  @end    printf ("OK/n");}

int main (int argc, char** argv) {    zsock_t *client = zsock_new (ZMQ_DEALER);    int i;    for (i = 0; i < 255; i++)    {        int port = 5556;        // printf ("Seeking to server at '%s.%d:%d'/n", argv [1], i, port);        zsock_connect (client, "tcp://%s.%d:%d", argv [1], i, port);    }    zsock_set_rcvtimeo (client, 2000);    while ( 1 )    {        char *reply = zstr_recv (client);        if (reply) {            puts (reply);            free (reply);        }        else        {            puts ("-");            break;        }    }    zsock_destroy (&client);    return 0;}

voidzyre_peer_test (bool verbose){    printf (" * zyre_peer: ");    zsock_t *mailbox = zsock_new_dealer ("@tcp://127.0.0.1:5551");    zhash_t *peers = zhash_new ();    zuuid_t *you = zuuid_new ();    zuuid_t *me = zuuid_new ();    zyre_peer_t *peer = zyre_peer_new (peers, you);    assert (!zyre_peer_connected (peer));    zyre_peer_connect (peer, me, "tcp://127.0.0.1:5551");    assert (zyre_peer_connected (peer));    zyre_peer_set_name (peer, "peer");    assert (streq (zyre_peer_name (peer), "peer"));    zre_msg_t *msg = zre_msg_new (ZRE_MSG_HELLO);    zre_msg_set_endpoint (msg, "tcp://127.0.0.1:5552");    int rc = zyre_peer_send (peer, &msg);    assert (rc == 0);    msg = zre_msg_recv (mailbox);    assert (msg);    if (verbose)        zre_msg_print (msg);    zre_msg_destroy (&msg);    //  Destroying container destroys all peers it contains    zhash_destroy (&peers);    zuuid_destroy (&me);    zuuid_destroy (&you);    zsock_destroy (&mailbox);    printf ("OK/n");}

static voidserver_connect (server_t *self, const char *endpoint){    zsock_t *remote = zsock_new (ZMQ_DEALER);    assert (remote);          //  No recovery if exhausted    //  Never block on sending; we use an infinite HWM and buffer as many    //  messages as needed in outgoing pipes. Note that the maximum number    //  is the overall tuple set size.    zsock_set_sndhwm (remote, 0);    if (zsock_connect (remote, "%s", endpoint)) {        zsys_error ("bad zgossip endpoint '%s'", endpoint);        zsock_destroy (&remote);        return;    }    //  Send HELLO and then PUBLISH for each tuple we have    zgossip_msg_send_hello (remote);    tuple_t *tuple = (tuple_t *) zhash_first (self->tuples);    while (tuple) {        int rc = zgossip_msg_send_publish (remote, tuple->key, tuple->value, 0);        assert (rc == 0);        tuple = (tuple_t *) zhash_next (self->tuples);    }    //  Now monitor this remote for incoming messages    engine_handle_socket (self, remote, remote_handler);    zlist_append (self->remotes, remote);}

voidmdp_broker_test (bool verbose){    printf (" * mdp_broker: ");    if (verbose)        printf ("/n");        //  @selftest    zactor_t *server = zactor_new (mdp_broker, "server");    if (verbose)        zstr_send (server, "VERBOSE");    zstr_sendx (server, "BIND", "ipc://@/mdp_broker", NULL);    zsock_t *client = zsock_new (ZMQ_DEALER);    assert (client);    zsock_set_rcvtimeo (client, 2000);    zsock_connect (client, "ipc://@/mdp_broker");    //  TODO: fill this out    mdp_msg_t *request = mdp_msg_new ();    mdp_msg_destroy (&request);        zsock_destroy (&client);    zactor_destroy (&server);    //  @end    printf ("OK/n");}

MessageProcessor::~MessageProcessor(){    zpoller_remove(zmqPoller_, zmqSocket_);    zpoller_destroy(&zmqPoller_);    zactor_destroy(&zmqAuth_);    zsock_destroy(&zmqSocket_);}

static zsock_t *s_create_socket (char *type_name, char *endpoints){    //  This array matches ZMQ_XXX type definitions    assert (ZMQ_PAIR == 0);    char *type_names [] = {        "PAIR", "PUB", "SUB", "REQ", "REP",        "DEALER", "ROUTER", "PULL", "PUSH",        "XPUB", "XSUB", type_name    };    //  We always match type at least at end of table    int index;    for (index = 0; strneq (type_name, type_names [index]); index++) ;    if (index > ZMQ_XSUB) {        zsys_error ("zproxy: invalid socket type '%s'", type_name);        return NULL;    }    zsock_t *sock = zsock_new (index);    if (sock) {        if (zsock_attach (sock, endpoints, true)) {            zsys_error ("zproxy: invalid endpoints '%s'", endpoints);            zsock_destroy (&sock);        }    }    return sock;}

static voidserver_connect (server_t *self, const char *endpoint){    zsock_t *remote = zsock_new (ZMQ_DEALER);    assert (remote);          //  No recovery if exhausted    //  Never block on sending; we use an infinite HWM and buffer as many    //  messages as needed in outgoing pipes. Note that the maximum number    //  is the overall tuple set size.    zsock_set_unbounded (remote);    if (zsock_connect (remote, "%s", endpoint)) {        zsys_warning ("bad zgossip endpoint '%s'", endpoint);        zsock_destroy (&remote);        return;    }    //  Send HELLO and then PUBLISH for each tuple we have    zgossip_msg_t *gossip = zgossip_msg_new ();    zgossip_msg_set_id (gossip, ZGOSSIP_MSG_HELLO);    zgossip_msg_send (gossip, remote);        tuple_t *tuple = (tuple_t *) zhashx_first (self->tuples);    while (tuple) {        zgossip_msg_set_id (gossip, ZGOSSIP_MSG_PUBLISH);        zgossip_msg_set_key (gossip, tuple->key);        zgossip_msg_set_value (gossip, tuple->value);        zgossip_msg_send (gossip, remote);        tuple = (tuple_t *) zhashx_next (self->tuples);    }    //  Now monitor this remote for incoming messages    zgossip_msg_destroy (&gossip);    engine_handle_socket (self, remote, remote_handler);    zlistx_add_end (self->remotes, remote);}

zsock_t *zsock_new_dealer_ (const char *endpoints, const char *filename, size_t line_nbr){    zsock_t *sock = zsock_new_ (ZMQ_DEALER, filename, line_nbr);    if (zsock_attach (sock, endpoints, false))        zsock_destroy (&sock);    return sock;}

zsock_t *zsock_new_sub_ (const char *endpoints, const char *subscribe, const char *filename, size_t line_nbr){    zsock_t *sock = zsock_new_ (ZMQ_SUB, filename, line_nbr);    if (zsock_attach (sock, endpoints, false))        zsock_destroy (&sock);    return sock;}

zsock_t *zsock_new_pub_ (const char *endpoints, const char *filename, size_t line_nbr){    zsock_t *sock = zsock_new_ (ZMQ_PUB, filename, line_nbr);    if (zsock_attach (sock, endpoints, true))        zsock_destroy (&sock);    return sock;}

/* cleanup */void zmq_sender_cleanup(ubx_block_t *b){        struct zmq_sender_info *inf = (struct zmq_sender_info*) b->private_data;        delete inf->buffer;        // clean up the publisher socket	zsock_destroy(&inf->publisher);        free(b->private_data);}

intzpubsub_filter_test (bool verbose){    printf (" * zpubsub_filter: ");    //  @selftest    //  Simple create/destroy test    zpubsub_filter_t *self = zpubsub_filter_new ();    assert (self);    zpubsub_filter_destroy (&self);    //  Create pair of sockets we can send through    zsock_t *input = zsock_new (ZMQ_ROUTER);    assert (input);    zsock_connect (input, "inproc://selftest-zpubsub_filter");    zsock_t *output = zsock_new (ZMQ_DEALER);    assert (output);    zsock_bind (output, "inproc://selftest-zpubsub_filter");    //  Encode/send/decode and verify each message type    int instance;    self = zpubsub_filter_new ();    zpubsub_filter_set_id (self, ZPUBSUB_FILTER_FILTER);    zpubsub_filter_set_partition (self, "Life is short but Now lasts for ever");    zpubsub_filter_set_topic (self, "Life is short but Now lasts for ever");    //  Send twice    zpubsub_filter_send (self, output);    zpubsub_filter_send (self, output);    for (instance = 0; instance < 2; instance++) {        zpubsub_filter_recv (self, input);        assert (zpubsub_filter_routing_id (self));        assert (streq (zpubsub_filter_partition (self), "Life is short but Now lasts for ever"));        assert (streq (zpubsub_filter_topic (self), "Life is short but Now lasts for ever"));    }    zpubsub_filter_destroy (&self);    zsock_destroy (&input);    zsock_destroy (&output);    //  @end    printf ("OK/n");    return 0;}

//  Checks whether client can connect to serverstatic bools_can_connect (zsock_t **server, zsock_t **client){    int port_nbr = zsock_bind (*server, "tcp://127.0.0.1:*");    assert (port_nbr > 0);    int rc = zsock_connect (*client, "tcp://127.0.0.1:%d", port_nbr);    assert (rc == 0);    zstr_send (*server, "Hello, World");    zpoller_t *poller = zpoller_new (*client, NULL);    bool success = (zpoller_wait (poller, 200) == *client);    zpoller_destroy (&poller);    zsock_destroy (client);    zsock_destroy (server);    *server = zsock_new (ZMQ_PUSH);    *client = zsock_new (ZMQ_PULL);    return success;}

/* cleanup */void zmq_receiver_cleanup(ubx_block_t *b){	struct zmq_receiver_info *inf = (struct zmq_receiver_info*) b->private_data;	// clean up subscriber socket	zsock_destroy(&inf->subscriber);	// clean up actor thread	zactor_destroy(&inf->actor);        free(b->private_data);}

static voids_self_destroy (self_t **self_p){    assert (self_p);    if (*self_p) {        self_t *self = *self_p;        zsock_destroy (&self->frontend);        zsock_destroy (&self->backend);        zsock_destroy (&self->capture);        for (int index = 0; index < SOCKETS; index++) {            zstr_free (&self->domain [index]);            zstr_free (&self->public_key [index]);            zstr_free (&self->secret_key [index]);        }        zpoller_destroy (&self->poller);        free (self);        *self_p = NULL;    }}

static voidserver_terminate (server_t *self){    while (zlist_size (self->remotes) > 0) {        zsock_t *remote = (zsock_t *) zlist_pop (self->remotes);        zsock_destroy (&remote);    }    zlist_destroy (&self->remotes);    zhash_destroy (&self->tuples);}

void s_mdp_client_connect_to_broker (mdp_client_t *self){    if(self->client)        zsock_destroy (&self->client);    self->client = zsock_new (ZMQ_DEALER);    assert(0==zsock_connect (self->client, "%s", self->broker));    if (self->verbose)        zclock_log ("I: connecting to broker at %s...", self->broker);    zsock_set_rcvtimeo(self->client,self->timeout);}

//  Destroy the brokervoidbroker_destroy (broker_t **self_p){    assert (self_p);    if (*self_p) {        broker_t *self = *self_p;        zsock_destroy (&self->contexts);        zsock_destroy (&self->executors);        zpoller_destroy (&self->poller);        // TODO: delete contents of list.        zlist_destroy (&self->executor_lb);        zlist_destroy (&self->backlog);        //  Free object itself        free (self);        *self_p = NULL;    }}

int main(void){    zsock_t *sock = zsock_new(ZMQ_REQ);    zsock_destroy(&sock);#if defined(__WINDOWS__)    zsys_shutdown();#endif    return 0;}