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

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

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

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

zpipes_client_t *zpipes_client_new (const char *server_name, const char *pipe_name){    //  Create new pipe API instance    zpipes_client_t *self = (zpipes_client_t *) zmalloc (sizeof (zpipes_client_t));    assert (self);        //  Create dealer socket and connect to server IPC port    self->dealer = zsock_new (ZMQ_DEALER);    assert (self->dealer);    int rc = zsock_connect (self->dealer, "ipc://@/zpipes/%s", server_name);    assert (rc == 0);    //  Open pipe for reading or writing    if (*pipe_name == '>') {        zpipes_msg_send_output (self->dealer, pipe_name + 1);        if (s_expect_reply (self, ZPIPES_MSG_OUTPUT_OK))            zpipes_client_destroy (&self);    }    else {        zpipes_msg_send_input (self->dealer, pipe_name);        if (s_expect_reply (self, ZPIPES_MSG_INPUT_OK))            zpipes_client_destroy (&self);    }    return self;}

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

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)        return NULL;    //  Create front-to-back pipe pair    self->pipe = zsock_new (ZMQ_PAIR);    assert (self->pipe);    char endpoint [32];    while (true) {        sprintf (endpoint, "inproc://zactor-%04x-%04x/n",                 randof (0x10000), randof (0x10000));        if (zsock_bind (self->pipe, "%s", endpoint) == 0)            break;    }    shim->pipe = zsock_new (ZMQ_PAIR);    assert (shim->pipe);    int rc = zsock_connect (shim->pipe, "%s", endpoint);    assert (rc != -1);    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;}

JNIEXPORT jint JNICALLJava_org_zeromq_czmq_Zsock__1_1connect (JNIEnv *env, jclass c, jlong self, jstring format){    char *format_ = (char *) (*env)->GetStringUTFChars (env, format, NULL);    jint connect_ = (jint) zsock_connect ((zsock_t *) (intptr_t) self, "%s", format_);    (*env)->ReleaseStringUTFChars (env, format, format_);    return connect_;}

static voidconnect_to_server_endpoint (client_t *self){    if (zsock_connect (self->dealer, "%s", self->args->endpoint)) {        engine_set_exception (self, bad_endpoint_event);        zsys_warning ("could not connect to %s", self->args->endpoint);    }}

staticzsock_t* subscriber_sub_socket_new(subscriber_state_t *state){    zsock_t *socket = zsock_new(ZMQ_SUB);    assert(socket);    zsock_set_rcvhwm(socket, state->rcv_hwm);    // set subscription    if (!state->subscriptions || zlist_size(state->subscriptions) == 0) {        if (!state->subscriptions)            state->subscriptions = zlist_new();        zlist_append(state->subscriptions, zconfig_resolve(state->config, "/logjam/subscription", ""));    }    char *subscription = zlist_first(state->subscriptions);    bool subscribed_to_all = false;    while (subscription) {        printf("[I] subscriber: subscribing to '%s'/n", subscription);        if (streq(subscription, ""))            subscribed_to_all = true;        zsock_set_subscribe(socket, subscription);        subscription = zlist_next(state->subscriptions);    }    if (!subscribed_to_all)        zsock_set_subscribe(socket, "heartbeat");    if (!state->devices || zlist_size(state->devices) == 0) {        // convert config file to list of devices        if (!state->devices)            state->devices = zlist_new();        zconfig_t *endpoints = zconfig_locate(state->config, "/logjam/endpoints");        if (!endpoints) {            zlist_append(state->devices, "tcp://localhost:9606");        } else {            zconfig_t *endpoint = zconfig_child(endpoints);            while (endpoint) {                char *spec = zconfig_value(endpoint);                char *new_spec = augment_zmq_connection_spec(spec, 9606);                zlist_append(state->devices, new_spec);                endpoint = zconfig_next(endpoint);            }        }    }    char* device = zlist_first(state->devices);    while (device) {        printf("[I] subscriber: connecting SUB socket to logjam-device via %s/n", device);        int rc = zsock_connect(socket, "%s", device);        log_zmq_error(rc, __FILE__, __LINE__);        assert(rc == 0);        device = zlist_next(state->devices);    }    return socket;}

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

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

int main(int argc, char const * const *argv){  int rc;  zsys_set_sndhwm(1);  zsys_set_linger(100);  void *pusher = zsock_new(ZMQ_PUSH);  assert(pusher);  zsock_set_sndhwm(pusher, 1000);  zsock_set_linger(pusher, 500);  rc = zsock_connect(pusher, "tcp://localhost:12345");  assert(rc==0);  void *puller = zsock_new(ZMQ_PULL);  assert(puller);  zsock_set_rcvhwm(puller, 1000);  zsock_set_linger(puller, 500);  rc = zsock_bind(puller, "tcp://*:12345");  if (rc != 12345){    printf("bind failed: %s/n", zmq_strerror(errno));  }  assert(rc == 12345);  void *publisher = zsock_new(ZMQ_PUB);  assert(publisher);  zsock_set_sndhwm(publisher, 1000);  zsock_set_linger(publisher, 500);  rc = zsock_bind(publisher, "tcp://*:12346");  assert(rc==12346);  // set up event loop  zloop_t *loop = zloop_new();  assert(loop);  zloop_set_verbose(loop, 0);  // push data every 10 ms  rc = zloop_timer(loop, 1, 0, timer_event, pusher);  assert(rc != -1);  zmq_pollitem_t item;  item.socket = puller;  item.events = ZMQ_POLLIN;  rc = zloop_poller(loop, &item, forward, publisher);  assert(rc == 0);  rc = zloop_start(loop);  printf("zloop return: %d", rc);  zloop_destroy(&loop);  assert(loop == NULL);  return 0;}

STATIC mp_obj_t socket_connect(mp_obj_t self_in, mp_obj_t addr_in) {    socket_obj_t *socket = self_in;    socket_check_closed(socket);    struct sockaddr sockaddr;    parse_inet_addr(socket, addr_in, &sockaddr);    int res = zsock_connect(socket->ctx, &sockaddr, sizeof(sockaddr));    RAISE_SOCK_ERRNO(res);    return mp_const_none;}

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

int main(int argc, char** argv) {    zsock_t * sc = zsock_new(ZMQ_PUB);    zsock_connect(sc, "tcp://%s:5560", argv[1]);    while(!zsys_interrupted) {        if(random()%1) {            zstr_send(sc, "ON");        } else {            zstr_send(sc, "OFF");        }        sleep(1);    }    zsock_destroy(&sc);}

static voidconnect_to_server (client_t *self){    if (zsock_connect(self->dealer, "%s", self->args->endpoint)) {        engine_set_exception(self, connect_error_event);        zsys_warning("could not connect to %s", self->args->endpoint);        zsock_send(self->cmdpipe, "si", "FAILURE", 0);    }    else {        zsys_debug("connected to %s", self->args->endpoint);        zsock_send(self->cmdpipe, "si", "SUCCESS", 0);    }}

intzsock_attach (zsock_t *self, const char *endpoints, bool serverish){    assert (self);    if (!endpoints)        return 0;    //  We hold each individual endpoint here    char endpoint [256];    while (*endpoints) {        const char *delimiter = strchr (endpoints, ',');        if (!delimiter)            delimiter = endpoints + strlen (endpoints);        if (delimiter - endpoints > 255)            return -1;        memcpy (endpoint, endpoints, delimiter - endpoints);        endpoint [delimiter - endpoints] = 0;        int rc;        if (endpoint [0] == '@')            rc = zsock_bind (self, "%s", endpoint + 1);        else        if (endpoint [0] == '>')            rc = zsock_connect (self, "%s", endpoint + 1);        else        if (serverish)            rc = zsock_bind (self, "%s", endpoint);        else            rc = zsock_connect (self, "%s", endpoint);                if (rc == -1)            return -1;          //  Bad endpoint syntax                    if (*delimiter == 0)            break;        endpoints = delimiter + 1;    }    return 0;}

// Checks whether client can connect to serverstatic bools_can_connect (zactor_t **proxy, zsock_t **faucet, zsock_t **sink, const char *frontend, const char *backend, bool verbose){    assert (frontend);    assert (*faucet);    int rc = zsock_connect (*faucet, "%s", frontend);    assert (rc == 0);    assert (backend);    assert (*sink);    rc = zsock_connect (*sink, "%s", backend);    assert (rc == 0);    zstr_send (*faucet, "Hello, World");    zpoller_t *poller = zpoller_new (*sink, NULL);    assert (poller);    bool success = (zpoller_wait (poller, 200) == *sink);    zpoller_destroy (&poller);    s_create_test_sockets (proxy, faucet, sink, verbose);    return success;}

static voids_upstream_handle_pipe (upstream_t *self){    char *command = zstr_recv (self->pipe);    if (streq (command, "$TERM")) {        self->terminated = true;    }    else    if (streq (command, "CONNECT")) {        char *endpoint = zstr_recv (self->pipe);        int rc = zsock_connect (self->push, "%s", endpoint);        if (rc != -1)            self->connected = true;        zstr_free (&endpoint);        zsock_bsend (self->pipe, "2", rc);    }    else    if (streq (command, "BIND")) {        char *endpoint = zstr_recv (self->pipe);        int rc = zsock_bind (self->push, "%s", endpoint);        if (rc != -1)            self->connected = true;        zstr_free (&endpoint);        zsock_bsend (self->pipe, "2", rc);    }    else    if (streq (command, "SET_SIZE")) {        int64_t size;        zsock_brecv (self->pipe, "8", &size);        self->size = size;    }    else    if (streq (command, "GET_SIZE")) {        zsock_bsend (self->pipe, "8", self->size);    }    else    if (streq (command, "SET_VARIANCE")) {        int64_t variance;        zsock_brecv (self->pipe, "8", &variance);        self->variance = variance;    }    else    if (streq (command, "GET_VARIANCE")) {        zsock_bsend (self->pipe, "8", self->variance);    }    else {        zsys_error ("upstream: invalid command: %s", command);        assert (false);    }    zstr_free (&command);}

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

server_connect (server_t *self, const char *endpoint)#endif{    zsock_t *remote = zsock_new (ZMQ_DEALER);    assert (remote);          //  No recovery if exhausted#ifdef CZMQ_BUILD_DRAFT_API    //  DRAFT-API: Security    if (public_key){        zcert_t *cert = zcert_new_from_txt (self->public_key, self->secret_key);        zcert_apply(cert, remote);        zsock_set_curve_serverkey (remote, public_key);#ifndef ZMQ_CURVE        // legacy ZMQ support        // inline incase the underlying assert is removed        bool ZMQ_CURVE = false;#endif        assert (zsock_mechanism (remote) == ZMQ_CURVE);        zcert_destroy(&cert);    }#endif    //  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);}

staticzsock_t* parser_push_socket_new(){    zsock_t *socket = zsock_new(ZMQ_PUSH);    assert(socket);    int rc;    // connect socket, taking thread startup time into account    // TODO: this is a hack. better let controller coordinate this    for (int i=0; i<10; i++) {        rc = zsock_connect(socket, "inproc://graylog-forwarder-writer");        if (rc == 0) break;        zclock_sleep(100);    }    return socket;}

voidzyre_peer_connect (zyre_peer_t *self, zuuid_t *from, const char *endpoint){    assert (self);    assert (!self->connected);    //  Create new outgoing socket (drop any messages in transit)    self->mailbox = zsock_new (ZMQ_DEALER);    if (!self->mailbox)        return;             //  Null when we're shutting down        //  Set our own identity on the socket so that receiving node    //  knows who each message came from. Note that we cannot use    //  the UUID directly as the identity since it may contain a    //  zero byte at the start, which libzmq does not like for    //  historical and arguably bogus reasons that it nonetheless    //  enforces.    byte routing_id [ZUUID_LEN + 1] = { 1 };    memcpy (routing_id + 1, zuuid_data (from), ZUUID_LEN);    int rc = zmq_setsockopt (zsock_resolve (self->mailbox),                             ZMQ_IDENTITY, routing_id, ZUUID_LEN + 1);    assert (rc == 0);    //  Set a high-water mark that allows for reasonable activity    zsock_set_sndhwm (self->mailbox, PEER_EXPIRED * 100);    //  Send messages immediately or return EAGAIN    zsock_set_sndtimeo (self->mailbox, 0);    //  Connect through to peer node    rc = zsock_connect (self->mailbox, "%s", endpoint);    if (rc != 0) {        zsys_error ("(%s) cannot connect to endpoint=%s",                    self->origin, endpoint);        //  Don't really have any error handling yet; if connect        //  fails, there's something wrong with connect endpoint?        assert (false);    }    assert (rc == 0);    if (self->verbose)        zsys_info ("(%s) connect to peer: endpoint=%s",                   self->origin, endpoint);    self->endpoint = strdup (endpoint);    self->connected = true;    self->ready = false;}

int main(int argc, char** argv) {    if(argc != 2) {        fprintf(stderr, "Usage: %s ups_name/n", argv[0]);        exit(1);    }    char *addr = NULL;    zyre_t *n = zyre_new(argv[1]);    zyre_start(n);    zyre_join(n, "BIOS");    while(!zsys_interrupted && addr == NULL) {        zyre_event_t *e = zyre_event_new(n);        if(!e)            break;        if(zyre_event_headers(e) && zyre_event_header(e, "HAP_SERVER") != NULL) {            addr = strdup(zyre_event_header(e, "HAP_SERVER"));            printf("Address: %s/n", addr);        }        zyre_event_destroy(&e);    }    zyre_destroy(&n);    if(addr == NULL)        exit(1);    zsock_t * sc = zsock_new(ZMQ_PUB);    zsock_connect(sc, "%s", addr);    bool state = random()%2;    int timeout = 0;    while(!zsys_interrupted) {        if(timeout == 0) {            state = !state;            timeout = 5 + random()%20;        }        timeout--;        if(state) {            zstr_sendx(sc, argv[1], "ON", NULL);            zsys_debug("UPS %s ON", argv[1]);        } else {            zstr_sendx(sc, argv[1], "OFF", NULL);            zsys_debug("UPS %s OFF", argv[1]);        }        sleep(1);    }    zsock_destroy(&sc);}

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

staticzsock_t* parser_pull_socket_new(){    int rc;    zsock_t *socket = zsock_new(ZMQ_PULL);    assert(socket);    // connect socket, taking thread startup time into account    // TODO: this is a hack. better let controller coordinate this    for (int i=0; i<10; i++) {        rc = zsock_connect(socket, "inproc://graylog-forwarder-subscriber");        if (rc == 0) break;        zclock_sleep(100);    }    log_zmq_error(rc, __FILE__, __LINE__);    assert(rc == 0);    return socket;}

static voids_self_start (self_t *self){    assert (!self->sink);    char *endpoint = zsys_sprintf ("inproc://zmonitor-%p", self->monitored);    int rc;#if defined (ZMQ_EVENT_ALL)    rc = zmq_socket_monitor (self->monitored, endpoint, self->events);    assert (rc == 0);#endif    self->sink = zsock_new (ZMQ_PAIR);    assert (self->sink);    rc = zsock_connect (self->sink, "%s", endpoint);    assert (rc == 0);    zpoller_add (self->poller, self->sink);    free (endpoint);}

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