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

staticvoid controller_destroy_actors(controller_state_t *state){    zactor_destroy(&state->subscriber);    zactor_destroy(&state->writer);    for (size_t i=0; i<num_parsers; i++) {        graylog_forwarder_parser_destroy(&state->parsers[i]);    }}

int main (int argc, char *argv []){    //  Get number of nodes N to simulate    //  We need 3 x N x N + 3N file handles    int max_nodes = 10;    int nbr_nodes = 0;    if (argc > 1)        max_nodes = atoi (argv [1]);    assert (max_nodes);    int max_iterations = -1;    int nbr_iterations = 0;    if (argc > 2)        max_iterations = atoi (argv [2]);    //  Our gossip network will use one fixed hub (not a Zyre node),    //  to which all nodes will connect    zactor_t *hub = zactor_new (zgossip, "hub");    zstr_sendx (hub, "BIND", "inproc://zyre-hub", NULL);            //  We address nodes as an array of actors    zactor_t **actors = (zactor_t **) zmalloc (sizeof (zactor_t *) * max_nodes);    //  We will randomly start and stop node threads    uint index;    while (!zsys_interrupted) {        index = randof (max_nodes);        //  Toggle node thread        if (actors [index]) {            zactor_destroy (&actors [index]);            actors [index] = NULL;            zsys_info ("stopped node (%d running)", --nbr_nodes);        }        else {            char node_name [10];            sprintf (node_name, "node-%d", index);            actors [index] = zactor_new (node_actor, strdup (node_name));            zsys_info ("started node (%d running)", ++nbr_nodes);        }        nbr_iterations++;        if (max_iterations > 0 && nbr_iterations >= max_iterations)            break;        //  Sleep ~300 msecs randomly so we smooth out activity        zclock_sleep (randof (100) + 100);    }    zsys_info ("stopped tester (%d iterations)", nbr_iterations);    //  Stop all remaining actors    for (index = 0; index < max_nodes; index++) {        if (actors [index])            zactor_destroy (&actors [index]);    }    free (actors);        zactor_destroy (&hub);    return 0;}

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

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

int main (void) {  zsys_set_ipv6 (1);  zactor_t *root = zactor_new (zgossip, "root");  assert (root);  int rc = 0;  rc = zstr_sendx (root, "BIND", "tcp://*:5670", NULL);  assert (rc == 0);  zloop_t *reactor = zloop_new ();  assert (reactor);  zloop_set_verbose (reactor, true);  rc = zloop_reader (reactor, root, handle_pipe, NULL);  assert (rc == 0);  zloop_start (reactor);  zloop_destroy (&reactor);  zactor_destroy (&root);  return 0;}

static intzyre_node_stop (zyre_node_t *self){    if (self->beacon) {        //  Stop broadcast/listen beacon        beacon_t beacon;        beacon.protocol [0] = 'Z';        beacon.protocol [1] = 'R';        beacon.protocol [2] = 'E';        beacon.version = BEACON_VERSION;        beacon.port = 0;            //  Zero means we're stopping        zuuid_export (self->uuid, beacon.uuid);        zsock_send (self->beacon, "sbi", "PUBLISH",            (byte *) &beacon, sizeof (beacon_t), self->interval);        zclock_sleep (1);           //  Allow 1 msec for beacon to go out        zpoller_remove (self->poller, self->beacon);        zactor_destroy (&self->beacon);    }    //  Stop polling on inbox    zpoller_remove (self->poller, self->inbox);    zstr_sendm (self->outbox, "STOP");    zstr_sendm (self->outbox, zuuid_str (self->uuid));    zstr_send (self->outbox, self->name);    return 0;}

int main(int argc, char **argv) {    //    //    const char *config_file = "malamute.cfg";    if (argc >= 2)        config_file = argv[1];    //  Start Malamute server instance    zactor_t *server = zactor_new (mlm_server, "Malamute");    zstr_sendx (server, "LOAD", config_file, NULL);    //  Accept and print any message back from server    while (true) {        char *message = zstr_recv (server);        if (message) {            puts (message);            free (message);        }        else {            puts ("interrupted");            break;        }    }    //  Shutdown all services    zactor_destroy (&server);    return EXIT_SUCCESS;}

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

int main (int argc, char *argv []){    //  Get number of nodes N to simulate    //  We need 3 x N x N + 3N file handles    int max_nodes = 10;    int nbr_nodes = 0;    if (argc > 1)        max_nodes = atoi (argv [1]);    int max_iterations = -1;    int nbr_iterations = 0;    if (argc > 2)        max_iterations = atoi (argv [2]);    //  We address nodes as an array of actors    zactor_t **actors = (zactor_t **) zmalloc (sizeof (zactor_t *) * max_nodes);    //  We will randomly start and stop node threads    uint index;    while (!zsys_interrupted) {        index = randof (max_nodes);        //  Toggle node thread        if (actors [index]) {            zactor_destroy (&actors [index]);            actors [index] = NULL;            zsys_info ("stopped node (%d running)", --nbr_nodes);        }        else {            actors [index] = zactor_new (node_actor, NULL);            zsys_info ("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);    }    zsys_info ("stopped tester (%d iterations)", nbr_iterations);    //  Stop all remaining actors    for (index = 0; index < max_nodes; index++) {        if (actors [index])            zactor_destroy (&actors [index]);    }    free (actors);    return 0;}

static voidserver_terminate (server_t *self){    zactor_destroy (&self->mailbox);    zhashx_destroy (&self->streams);    zhashx_destroy (&self->services);    zhashx_destroy (&self->clients);}

int main (void){    zactor_t *server = zactor_new (zccp_server, "zccpd");    zsock_send (server, "s", "VERBOSE");    zsock_send (server, "ss", "BIND", "ipc://@/zccp");    zsock_wait (server);    zactor_destroy (&server);    return 0;}

/* 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 voidserver_terminate (server_t *self){    zactor_destroy (&self->mailbox);    zhashx_destroy (&self->streams);    zhashx_destroy (&self->services);    zhashx_destroy (&self->clients);    mlm_msgq_cfg_destroy (&self->service_queue_cfg);}

int main() {    zactor_t *actor = zactor_new (s_alerts, NULL);    //XXX: this is UGLY    while (!zsys_interrupted) {        zclock_sleep(100);    }    zactor_destroy (&actor);}

voidzpipes_client_test (bool verbose){    printf (" * zpipes_client: ");    //  @selftest    zactor_t *server = zactor_new (zpipes_server, NULL);    zstr_sendx (server, "SET", "server/animate", verbose? "1": "0", NULL);    zstr_sendx (server, "BIND", "ipc://@/zpipes/local", NULL);        zpipes_client_t *reader = zpipes_client_new ("local", "test pipe");    zpipes_client_t *writer = zpipes_client_new ("local", ">test pipe");    byte buffer [100];    ssize_t bytes;    //  Expect timeout error, EAGAIN    bytes = zpipes_client_read (reader, buffer, 6, 100);    assert (bytes == -1);    assert (zpipes_client_error (reader) == EAGAIN);    bytes = zpipes_client_write (writer, "CHUNK1", 6, 100);    assert (bytes == 6);    bytes = zpipes_client_write (writer, "CHUNK2", 6, 100);    assert (bytes == 6);    bytes = zpipes_client_write (writer, "CHUNK3", 6, 100);    assert (bytes == 6);    bytes = zpipes_client_read (reader, buffer, 1, 100);    assert (bytes == 1);    bytes = zpipes_client_read (reader, buffer, 10, 100);    assert (bytes == 10);    //  Now close writer    zpipes_client_destroy (&writer);    //  Expect end of pipe (short read)    bytes = zpipes_client_read (reader, buffer, 50, 100);    assert (bytes == 7);        //  Expect end of pipe (empty chunk)    bytes = zpipes_client_read (reader, buffer, 50, 100);    assert (bytes == 0);    //  Expect illegal action (EBADF) writing on reader    bytes = zpipes_client_write (reader, "CHUNK1", 6, 100);    assert (bytes == -1);    assert (zpipes_client_error (reader) == EBADF);    zpipes_client_destroy (&reader);    zpipes_client_destroy (&writer);    zactor_destroy (&server);    //  @end    printf ("OK/n");}

static void twps_destroy(twps_server_t **self_p) {    assert(self_p);    if (*self_p) {        twps_server_t *self = *self_p;        if (self->verbose) {            zsys_info("twps server: shutdowning application server");        }        zactor_t *t_printer = zlistx_first(self->ticket_printers);        while (t_printer != NULL) {            zactor_destroy(&t_printer);            t_printer = zlistx_next(self->ticket_printers);        }        zlistx_destroy(&self->ticket_printers);        zsock_destroy(&self->printer_store_sub);        zactor_destroy(&self->ticket_store);        zactor_destroy(&self->printer_store);        zactor_destroy(&self->client_proxy);        free(self);        *self_p = NULL;    }}

static voids_stream_destroy (stream_t **self_p){    assert (self_p);    if (*self_p) {        stream_t *self = *self_p;        zactor_destroy (&self->actor);        zsock_destroy (&self->msgpipe);        free (self->name);        free (self);        *self_p = NULL;    }}

voidexample_peer_test (bool verbose){    printf (" * example_peer: ");    if (verbose)        printf ("/n");    //  @selftest    zactor_t *client = zactor_new (example_peer, NULL);    example_peer_verbose = verbose;    zactor_destroy (&client);    //  @end    printf ("OK/n");}

voidmdp_client_test (bool verbose){    printf (" * mdp_client: ");    if (verbose)        printf ("/n");        //  @selftest    zactor_t *client = zactor_new (mdp_client, NULL);    if (verbose)        zstr_send (client, "VERBOSE");    zactor_destroy (&client);    //  @end    printf ("OK/n");}

voidzyre_destroy (zyre_t **self_p){    assert (self_p);    if (*self_p) {        zyre_t *self = *self_p;        zactor_destroy (&self->actor);        zsock_destroy (&self->inbox);        zstr_free (&self->uuid);        zstr_free (&self->name);        zstr_free (&self->endpoint);        free (self);        *self_p = NULL;    }}

static voidzyre_node_destroy (zyre_node_t **self_p){    assert (self_p);    if (*self_p) {        zyre_node_t *self = *self_p;        zpoller_destroy (&self->poller);        zuuid_destroy (&self->uuid);        zhash_destroy (&self->peers);        zhash_destroy (&self->peer_groups);        zlist_destroy (&self->own_groups);        zhash_destroy (&self->headers);        zsock_destroy (&self->inbox);        zsock_destroy (&self->outbox);        zactor_destroy (&self->beacon);        zactor_destroy (&self->gossip);        zstr_free (&self->endpoint);        zstr_free (&self->gossip_bind);        zstr_free (&self->gossip_connect);        free (self->name);        free (self);        *self_p = NULL;    }}

zactor_t *zactor_new (zactor_fn *actor, void *args){    zactor_t *self = (zactor_t *) zmalloc (sizeof (zactor_t));    if (!self)        return NULL;    self->tag = ZACTOR_TAG;    shim_t *shim = (shim_t *) zmalloc (sizeof (shim_t));    if (!shim) {        zactor_destroy (&self);        return NULL;    }    shim->pipe = zsys_create_pipe (&self->pipe);    shim->handler = actor;    shim->args = args;#if defined (__UNIX__)    pthread_t thread;    pthread_create (&thread, NULL, s_thread_shim, shim);    pthread_detach (thread);#elif defined (__WINDOWS__)    HANDLE handle = (HANDLE) _beginthreadex (        NULL,                   //  Handle is private to this process        0,                      //  Use a default stack size for new thread        &s_thread_shim,         //  Start real thread function via this shim        shim,                   //  Which gets arguments shim        CREATE_SUSPENDED,       //  Set thread priority before starting it        NULL);                  //  We don't use the thread ID    assert (handle);    //  Set child thread priority to same as current    int priority = GetThreadPriority (GetCurrentThread ());    SetThreadPriority (handle, priority);    //  Start thread & release resources    ResumeThread (handle);    CloseHandle (handle);#endif    //  Mandatory handshake for new actor so that constructor returns only    //  when actor has also initialized. This eliminates timing issues at    //  application start up.    zsock_wait (self->pipe);    return self;}

voidzactor_test (bool verbose){    printf (" * zactor: ");    //  @selftest    zactor_t *actor = zactor_new (echo_actor, "Hello, World");    assert (actor);    zstr_sendx (actor, "ECHO", "This is a string", NULL);    char *string = zstr_recv (actor);    assert (streq (string, "This is a string"));    free (string);    zactor_destroy (&actor);    //  @end    printf ("OK/n");}

voidmlm_stream_simple_test (bool verbose){    printf (" * mlm_stream_simple: ");    if (verbose)        printf ("/n");    //  @selftest    zactor_t *stream = zactor_new (mlm_stream_simple, NULL);    assert (stream);    if (verbose)        zstr_sendx (stream, "VERBOSE", NULL);    zactor_destroy (&stream);    //  @end    printf ("OK/n");}

intmain (int argc, char *argv[]){    setvbuf(stdout, NULL, _IONBF, 0);    char executors_uri[256];    if (argc == 3)        snprintf (executors_uri, 256, "tcp://%s:%s", argv[1], argv[2]);    else        snprintf (executors_uri, 256, "tcp://127.0.0.1:%s", argv[1]);    zactor_t *executor = zactor_new (executor_actor, (void *) executors_uri);    zstr_send (executor, "START");    zsock_wait (executor);    // Wait for interrupted    zsock_wait (executor);    zactor_destroy (&executor);}

intmain (int argc, char *argv []){    if (argc < 2) {        puts ("syntax: ./chat myname");        exit (0);    }    zactor_t *actor = zactor_new (chat_actor, argv [1]);    assert (actor);        while (!zsys_interrupted) {        char message [1024];        if (!fgets (message, 1024, stdin))            break;        message [strlen (message) - 1] = 0;     // Drop the trailing linefeed        zstr_sendx (actor, "SHOUT", message, NULL);    }    zactor_destroy (&actor);    return 0;}

int main(int argc, char** argv) {    if (argc != 2) {        fprintf (stderr, "Usage: %s name/n", argv[0]);        exit (EXIT_FAILURE);    }    zactor_t *gs = zactor_new (zgossip, argv[1]);    zstr_send (zgossip, "VERBOSE");    zstr_sendx (gs, "BIND", "ipc://@/bios-alerts");    /*    char buf[1024];    snprintf(buf, 1024, "%d", random());    zsys_debug ("PUBLISH: %s", buf);    zstr_sendx (gs, "PUBLISH", "X-BIOS-PATH", buf, NULL);    */    while (!zsys_interrupted) {        char *method, *key, *value;        zmsg_t *msg = zactor_recv (gs);        if (!msg)            continue;        zmsg_print (msg);        method = zmsg_popstr (msg);        key = zmsg_popstr (msg);        value = zmsg_popstr (msg);        printf ("method: '%s', key: '%s', value: '%s'/n", method, key, value);        zstr_free (&key);        zstr_free (&value);        zstr_free (&method);        zmsg_destroy (&msg);    }    zactor_destroy (&gs);}

int main(int argc, char* argv[]) {    srand(time(NULL));    broker_t *broker = broker_new (        "ipc://contexts.ipc", "ipc://executors.ipc");    pthread_t broker_thread = broker_run_in_thread (&broker);    zactor_t *executor = zactor_new (executor_actor, NULL);    zstr_send (executor, "START");    zsock_wait (executor);    sprk_ctx_t *ctx = sprk_ctx_new("ipc://contexts.ipc");    sprk_dataset_t *dataset = sprk_ctx_load_dense (ctx, "/tmp/gen.dat", 400);    sleep (3);    pthread_cancel (broker_thread);    pthread_join (broker_thread, NULL);    sprk_dataset_destroy (&dataset);    zactor_destroy (&executor);    sprk_ctx_destroy(&ctx);}

static voids_create_test_sockets (zactor_t **proxy, zsock_t **faucet, zsock_t **sink, bool verbose){    if (*faucet)        zsock_destroy (faucet);    if (*sink)        zsock_destroy (sink);    if (*proxy)        zactor_destroy (proxy);    *faucet = zsock_new (ZMQ_PUSH);    assert (*faucet);    *sink = zsock_new (ZMQ_PULL);    assert (*sink);    *proxy = zactor_new (zproxy, NULL);    assert (*proxy);    if (verbose) {        zstr_sendx (*proxy, "VERBOSE", NULL);        zsock_wait (*proxy);    }}