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

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

zyre_t *zyre_new (const char *name){    zyre_t *self = (zyre_t *) zmalloc (sizeof (zyre_t));    assert (self);    //  Create front-to-back pipe pair for data traffic    self->inbox = zsock_new (ZMQ_PAIR);    assert (self->inbox);    char endpoint [32];    while (true) {        sprintf (endpoint, "inproc://zyre-%04x-%04x/n",                 randof (0x10000), randof (0x10000));        if (zsock_bind (self->inbox, "%s", endpoint) == 0)            break;    }    //  Create other half of traffic pipe    zsock_t *outbox = zsock_new_pair (endpoint);    assert (outbox);        //  Start node engine and wait for it to be ready    self->actor = zactor_new (zyre_node_actor, outbox);    assert (self->actor);    //  Send name, if any, to node ending    if (name)        zstr_sendx (self->actor, "SET NAME", name, NULL);        return self;}

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

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

MessageProcessor::MessageProcessor(ServerLoader& loader)    : server_(loader.getServer())    , zmqSocket_(zsock_new_rep(NULL))    , zmqAuth_(zactor_new(zauth, NULL))    , zmqPoller_(zpoller_new(zmqSocket_, NULL)){    init(loader.getPort(), loader.getTransportKey());}

zhttp_client_t *zhttp_client_new (bool verbose) {    zhttp_client_t *self = NULL;#ifdef HAVE_LIBCURL    self = (zhttp_client_t *) zactor_new (zhttp_client_actor, &verbose);#endif    return self;}

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

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

/* start */int czmq_ppworker_start(ubx_block_t *b){        struct czmq_ppworker_info *inf = (struct czmq_ppworker_info*) b->private_data;        int ret = 0;	// incoming data is handled by the actor thread        zactor_t* actor = zactor_new (ppworker_actor, b);        assert(actor);        inf->actor = actor;        return ret;}

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

/* start */int zmq_receiver_start(ubx_block_t *b){        struct zmq_receiver_info *inf = (struct zmq_receiver_info*) b->private_data;		// incoming data is handled by the actor thread        zactor_t* actor = zactor_new (receiver_actor, b);	inf->actor = actor;        int ret = 0;        return ret;}

static zactor_t *s_broker(char *endpoint){    //1. start malamute    zactor_t *broker = zactor_new (mlm_server, NULL);    assert (broker);    zsys_info ("malamute->bind(/"%s/")", endpoint);    zsock_send (broker, "ss", "BIND", endpoint);    zstr_send (broker, "VERBOSE");    return broker;}

//  If we haven't already set-up the gossip network, do sostatic voidzyre_node_gossip_start (zyre_node_t *self){    if (!self->gossip) {        self->beacon_port = 0;      //  Disable UDP beaconing        self->gossip = zactor_new (zgossip, self->name);        if (self->verbose)            zstr_send (self->gossip, "VERBOSE");        assert (self->gossip);    }}

static int twps_create_ticket_printer(twps_server_t *self, zmsg_t *msg) {    char *type = zmsg_popstr(msg);    char *id = zmsg_popstr(msg);    zstr_free(&id);    char *path = zmsg_popstr(msg);    char *model = zmsg_popstr(msg);    wchar_t *manufacture = (wchar_t *) zmsg_popstr(msg);    wchar_t *product = (wchar_t *) zmsg_popstr(msg);    zactor_t *printer = NULL;    if (streq(type, "HID")) {        printer = zactor_new(ticket_hid_printer, path);    }#ifdef __WIN32__    else if (streq(type, "SERIAL")) {        printer = zactor_new(ticket_serial_printer, path);    }#endif    if (printer != NULL) {        if (self->verbose) {            zstr_send(printer, "VERBOSE");        }        if (self->diagnostic)            zstr_sendx(printer, "SETDIAGNOSTIC", NULL);        zstr_sendx(printer, "START", TICKET_STORE_REP_ENDPOINT, NULL);        zsock_wait(printer);        zstr_sendx(printer, "SETPRINTERSTORE", PRINTER_STORE_REP_ENDPOINT, PRINTER_STORE_PUB_ENDPOINT, NULL);        zsock_wait(printer);        zsys_info("twps server: started ticket printer %s manufacturer|product|model %ls|%ls|%s", type, manufacture,                  product, model);        zlistx_add_end(self->ticket_printers, printer);    } else {        zsys_warning("twps server: printer not added %s, %s", type, path);        zmsg_print(msg);    }    zstr_free(&type);    zstr_free(&path);    zstr_free(&model);    zstr_free((char **) &manufacture);    zstr_free((char **) &product);    return 0;}

WorkerAgent::WorkerAgent(    const std::string &ecmKey,    const std::string &localEndpoint,    const std::string &publicEndpoint,    const std::function<void(const msgpack::unpacked &, msgpack::sbuffer &)> callback)    : _ecmKey(ecmKey)    , _localEndpoint(localEndpoint)    , _publicEndpoint(publicEndpoint)    , _callback(callback)    , _actor(zactor_new(actorTask, this)){    assert(_actor);}

static twps_server_t *twps_new(bool verbose, bool proxy_log) {    zsys_info("twps: TWPS is initializing.");    twps_server_t *self = (twps_server_t *) calloc(1, sizeof(twps_server_t));    self->verbose = verbose;    self->ticket_printers = zlistx_new();    self->ticket_store = zactor_new(ticket_store, NULL);    self->printer_store = zactor_new(printer_store, NULL);    self->client_proxy = zactor_new(client_proxy, NULL);    zsys_info("twps: actors created.");    if (verbose) {        zstr_send(self->ticket_store, "VERBOSE");        zstr_send(self->printer_store, "VERBOSE");    }    if (proxy_log)        zstr_send(self->client_proxy, "VERBOSE");    zstr_sendx(self->ticket_store, "START", TICKET_STORE_REP_ENDPOINT, TICKET_STORE_PUB_ENDPOINT, NULL);    zsock_wait(self->ticket_store);    zsys_info("twps: ticket store started listening on %s and publishing on %s", TICKET_STORE_REP_ENDPOINT,              TICKET_STORE_PUB_ENDPOINT);    zstr_sendx(self->printer_store, "START", PRINTER_STORE_REP_ENDPOINT, PRINTER_STORE_PUB_ENDPOINT, NULL);    zsock_wait(self->printer_store);    zsys_info("twps: printer store started listening on %s and publishing on %s", PRINTER_STORE_REP_ENDPOINT,              PRINTER_STORE_PUB_ENDPOINT);    zstr_sendx(self->client_proxy, "START", CLIENT_PROXY_ROUTE_ENDPOINT, CLIENT_PROXY_PUB_ENDPOINT, NULL);    zsock_wait(self->client_proxy);    zsys_info("twps: client proxy started listening on %s and publishing on %s", CLIENT_PROXY_ROUTE_ENDPOINT,              CLIENT_PROXY_PUB_ENDPOINT);    zstr_sendx(self->client_proxy, "SETTICKETSTORE", TICKET_STORE_REP_ENDPOINT, TICKET_STORE_PUB_ENDPOINT, NULL);    zsock_wait(self->client_proxy);    zstr_sendx(self->client_proxy, "SETPRINTERSTORE", PRINTER_STORE_REP_ENDPOINT, PRINTER_STORE_PUB_ENDPOINT, NULL);    zsock_wait(self->client_proxy);    self->printer_store_sub = zsock_new_sub(PRINTER_STORE_PUB_ENDPOINT, "");    zsys_info("twps: initialized with ticket store, printer store and client proxy");    zsys_info("twps: scanning for printers");    zstr_send(self->printer_store,"SCANPRINTERS");    return self;}

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

static intserver_initialize (server_t *self){    self->mailbox = zactor_new (mlm_mailbox_simple, NULL);    assert (self->mailbox);    self->streams = zhashx_new ();    assert (self->streams);    self->services = zhashx_new ();    assert (self->services);    self->clients = zhashx_new ();    assert (self->clients);    zhashx_set_destructor (self->streams, (czmq_destructor *) s_stream_destroy);    zhashx_set_destructor (self->services, (czmq_destructor *) s_service_destroy);    return 0;}

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

staticbool controller_create_actors(controller_state_t *state, zlist_t* devices, zlist_t *subscriptions, int rcv_hwm, int send_hwm){    // create subscriber    state->subscriber = graylog_forwarder_subscriber_new(state->config, devices, subscriptions, rcv_hwm, send_hwm);    // create the parsers    for (size_t i=0; i<num_parsers; i++) {        state->parsers[i] = graylog_forwarder_parser_new(state->config, i);    }    // create writer    state->writer = zactor_new(graylog_forwarder_writer, state->config);    return !zsys_interrupted;}

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

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

static intserver_initialize (server_t *self){    self->mailbox = zactor_new (mlm_mailbox_bounded, NULL);    assert (self->mailbox);    self->streams = zhashx_new ();    assert (self->streams);    self->services = zhashx_new ();    assert (self->services);    self->clients = zhashx_new ();    assert (self->clients);    zhashx_set_destructor (self->streams, (czmq_destructor *) s_stream_destroy);    zhashx_set_destructor (self->services, (czmq_destructor *) s_service_destroy);    zhashx_set_destructor (self->clients, (czmq_destructor *) s_client_local_destroy);    self->service_queue_cfg = mlm_msgq_cfg_new ("mlm_server/service/queue");    assert (self->service_queue_cfg);    return 0;}

static stream_t *s_stream_new (client_t *client, const char *name){    stream_t *self = (stream_t *) zmalloc (sizeof (stream_t));    if (self) {        zsock_t *backend;        self->name = strdup (name);        if (self->name)            self->msgpipe = zsys_create_pipe (&backend);        if (self->msgpipe) {            engine_handle_socket (client->server, self->msgpipe, s_forward_stream_traffic);            self->actor = zactor_new (mlm_stream_simple, backend);        }        if (!self->actor)            s_stream_destroy (&self);    }    return self;}

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

zyre_t *zyre_new (const char *name){    zyre_t *self = (zyre_t *) zmalloc (sizeof (zyre_t));    assert (self);    //  Create front-to-back pipe pair for data traffic    zsock_t *outbox;    self->inbox = zsys_create_pipe (&outbox);    //  Start node engine and wait for it to be ready    self->actor = zactor_new (zyre_node_actor, outbox);    assert (self->actor);    //  Send name, if any, to node ending    if (name)        zstr_sendx (self->actor, "SET NAME", name, NULL);    return self;}