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

static voids_agent_task (void *args, zctx_t *ctx, void *pipe){    char *endpoint = zstr_recv (pipe);    assert (endpoint);    agent_t *self = s_agent_new (ctx, pipe, endpoint);    zpoller_t *poller = zpoller_new (self->pipe, self->socket, NULL);    while (!zctx_interrupted) {        //  Poll on API pipe and on monitor socket        void *result = zpoller_wait (poller, -1);        if (result == NULL)            break; // Interrupted        else        if (result == self->pipe)            s_api_command (self);        else        if (result == self->socket)            s_socket_event (self);        if (self->terminated)            break;    }    zpoller_destroy (&poller);    s_agent_destroy (&self);}

voiddrops_agent_main (void *args, zctx_t *ctx, void *pipe){    //  Create agent instance to pass around    s_agent_t *self = s_agent_new (ctx, pipe);    if (!self)                  //  Interrupted        return;    zstr_send (self->pipe, "OK");    //  These are the sockets we will monitor for activity    zpoller_t *poller = zpoller_new (        self->pipe, zyre_socket (self->zyre), NULL);    while (!zpoller_terminated (poller)) {        //  Check directory once a second; this is a pretty nasty way of        //  doing it, but portable and simple. Later I'd like to use file        //  system monitoring library and get events back over a socket.        void *which = zpoller_wait (poller, 1000);        if (which == self->pipe)            s_recv_from_api (self);        else        if (which == zyre_socket (self->zyre))            s_recv_from_zyre (self);                if (self->terminated)            break;                s_check_directory (self);    }    zpoller_destroy (&poller);    s_agent_destroy (&self);}

static zyre_node_t *zyre_node_new (zsock_t *pipe, void *args){    zyre_node_t *self = (zyre_node_t *) zmalloc (sizeof (zyre_node_t));    self->inbox = zsock_new (ZMQ_ROUTER);    if (self->inbox == NULL) {        free (self);        return NULL;            //  Could not create new socket    }    //  Use ZMQ_ROUTER_HANDOVER so that when a peer disconnects and    //  then reconnects, the new client connection is treated as the    //  canonical one, and any old trailing commands are discarded.    zsock_set_router_handover (self->inbox, 1);        self->pipe = pipe;    self->outbox = (zsock_t *) args;    self->poller = zpoller_new (self->pipe, NULL);    self->beacon_port = ZRE_DISCOVERY_PORT;    self->interval = 0;         //  Use default    self->uuid = zuuid_new ();    self->peers = zhash_new ();    self->peer_groups = zhash_new ();    self->own_groups = zhash_new ();    self->headers = zhash_new ();    zhash_autofree (self->headers);    //  Default name for node is first 6 characters of UUID:    //  the shorter string is more readable in logs    self->name = (char *) zmalloc (7);    memcpy (self->name, zuuid_str (self->uuid), 6);    return self;}

static self_t *s_self_new (zsock_t *pipe, zcertstore_t *certstore){    self_t *self = (self_t *) zmalloc (sizeof (self_t));    assert (self);    if (certstore) {        self->certstore = certstore;        self->allow_any = false;    }    self->pipe = pipe;    self->whitelist = zhashx_new ();    assert (self->whitelist);    self->blacklist = zhashx_new ();    //  Create ZAP handler and get ready for requests    assert (self->blacklist);    self->handler = zsock_new (ZMQ_REP);    assert (self->handler);    int rc = zsock_bind (self->handler, ZAP_ENDPOINT);    assert (rc == 0);    self->poller = zpoller_new (self->pipe, self->handler, NULL);    assert (self->poller);    return self;}

//  Checks whether client can connect to serverstatic bools_can_connect (zctx_t *ctx, void **server, void **client){    int port_nbr = zsocket_bind (*server, "tcp://127.0.0.1:*");    assert (port_nbr > 0);    int rc = zsocket_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);    bool success = (zpoller_wait (poller, 400) == *client);    zpoller_destroy (&poller);    zsocket_destroy (ctx, *client);    zsocket_destroy (ctx, *server);    *server = zsocket_new (ctx, ZMQ_PUSH);    assert (*server);    *client = zsocket_new (ctx, ZMQ_PULL);    assert (*client);    return success;}

static bools_can_connect (void *server, void *client){    //  We'll do each test on a new port number since otherwise we have to    //  destroy and recreate the sockets each time.    static int port_nbr = 9000;    int rc = zsocket_bind (server, "tcp://*:%d", port_nbr);    assert (rc == port_nbr);    rc = zsocket_connect (client, "tcp://localhost:%d", port_nbr);    assert (rc == 0);        zpoller_t *poller = zpoller_new (client, NULL);    zstr_send (server, "Hello, World");    //  Need up to half a second if running under Valgrind    bool success = zpoller_wait (poller, 500) == client;    if (success)        free (zstr_recv (client));    zpoller_destroy (&poller);    rc = zsocket_unbind (server, "tcp://*:%d", port_nbr);    assert (rc != -1);    rc = zsocket_disconnect (client, "tcp://localhost:%d", port_nbr);    assert (rc != -1);    port_nbr++;    return success;}

broker_t *broker_new (const char *contexts_uri, const char *executors_uri){    broker_t *self = (broker_t *) malloc (sizeof (broker_t));    assert (self);    printf ("[BROKER] binding to frontend %s/n", contexts_uri);    self->contexts = zsock_new_router (contexts_uri);    assert(self->contexts);    printf ("[BROKER] binding to backend %s/n", executors_uri);    self->executors = zsock_new_router (executors_uri);    assert (self->executors);    zsock_set_router_mandatory (self->executors, true);    // Only poll on executors until we have executors available.    self->poller = zpoller_new (self->executors, NULL);    assert (self->poller);    self->executor_lb = zlist_new ();    assert (self->executor_lb);    self->backlog = zlist_new ();    assert (self->backlog);    return self;}

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

JNIEXPORT jlong JNICALLJava_org_zeromq_czmq_Zpoller__1_1new (JNIEnv *env, jclass c, jlong reader){    //  Disable CZMQ signal handling; allow Java to deal with it    zsys_handler_set (NULL);    jlong new_ = (jlong) (intptr_t) zpoller_new ((void *) (intptr_t) reader);    return new_;}

static self_t *s_self_new (zsock_t *pipe, zsock_t *sock){    self_t *self = (self_t *) zmalloc (sizeof (self_t));    self->pipe = pipe;    self->monitored = zsock_resolve (sock);    self->poller = zpoller_new (self->pipe, NULL);    return self;}

static self_t *s_self_new (zsock_t *pipe){    self_t *self = (self_t *) zmalloc (sizeof (self_t));    if (self) {        self->pipe = pipe;        self->poller = zpoller_new (self->pipe, NULL);        if (!self->poller)            s_self_destroy (&self);    }    return self;}

static someactor_t *someactor_new (zsock_t *pipe, void *args){    someactor_t *self = (someactor_t *) zmalloc (sizeof (someactor_t));    assert (self);    self->pipe = pipe;    self->terminated = false;    self->poller = zpoller_new (self->pipe, NULL);    //  TODO: Initialize properties    return self;}

static self_t *s_self_new (zsock_t *pipe, void *sock){    self_t *self = (self_t *) zmalloc (sizeof (self_t));    if (!self)        return NULL;    self->pipe = pipe;    self->monitored = zsock_resolve (sock);    self->poller = zpoller_new (self->pipe, NULL);    if (!self->poller)        s_self_destroy (&self);    return self;}

static voids_alerts (    zsock_t *pipe,    void *args) {    const char *name = "ALERT";    mlm_client_t *cl = mlm_client_new ();    mlm_client_connect (cl, endpoint, 5000, __PRETTY_FUNCTION__);    mlm_client_set_producer (cl, stream);    zsock_t *msgpipe = mlm_client_msgpipe (cl);    zpoller_t *poller = zpoller_new (pipe, msgpipe, NULL);    char *alert_state = strdup ("NEW");    zsock_signal (pipe, 0);    while (!zsys_interrupted) {        zsock_t *which = zpoller_wait (poller, 1000);        if (!which) {            mlm_client_sendx (cl, "alert://[email
C++ zpool_close函数代码示例
C++ zpoller_destroy函数代码示例