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

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

boolGlobalServer_init (    GlobalServer *self,    GlobalServerStartupInfo *info) {    memcpy (&self->info, info, sizeof (self->info));    // ==========================    //   Allocate ZMQ objects    // ==========================    if (!(self->cliConnection = zsock_new (ZMQ_RAW_ROUTER))) {        error ("Cannot allocate a new CLI zsock.");        return false;    }    if (!(self->zonesConnection = zsock_new (ZMQ_REQ))) {        error ("Cannot allocate a new zones zsock.");        return false;    }    // Connect to Redis Server    if (!(self->redis = Redis_new (&self->info.redisInfo))) {        error ("Cannot initialize Redis connection.");        return false;    }    if (!(Redis_connect (self->redis))) {        error ("Cannot connect to Redis.");        return false;    }    return true;}

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

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

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

voidzyre_node_test (bool verbose){    printf (" * zyre_node: ");    zsock_t *pipe = zsock_new (ZMQ_PAIR);    zsock_t *outbox = zsock_new (ZMQ_PAIR);    zyre_node_t *node = zyre_node_new (pipe, outbox);    zyre_node_destroy (&node);    zsock_destroy (&pipe);    //  Node takes ownership of outbox and destroys it    printf ("OK/n");}

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

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 (void){    printf("Starting dealer.../n");    //  Socket to talk to clients    zsock_t *server = zsock_new (ZMQ_DEALER);    int rc = zsock_bind (server, "tcp://*:5556");    checkRetCode(rc);    printf("Hiding from others.../n");    char* msg = "LucTeo (blue shirt + red)";    int caught = 0;    // while ( 1 )    {        rc = zstr_send (server, msg);        checkRetCode(rc);        if ( !caught )        {            printf("I'm caught. :(/n");            msg = "Too late, secret was secret but it's taken";            caught = 1;        }    }    // zsock_destroy (&server);    return 0;}

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

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

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

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

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

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

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

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

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

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

JNIEXPORT jlong JNICALLJava_org_zeromq_czmq_Zsock__1_1new (JNIEnv *env, jclass c, jint type){    //  Disable CZMQ signal handling; allow Java to deal with it    zsys_handler_set (NULL);    jlong new_ = (jlong) (intptr_t) zsock_new ((int) type);    return new_;}

bool eventServerInit(EventServer *self, EventServerStartupInfo *info, ServerType serverType) {    memcpy(&self->info, info, sizeof(self->info));    // create a unique publisher for the EventServer    if (!(self->eventsInput = zsock_new(ZMQ_SUB))) {        error("Cannot allocate a new Server SUBSCRIBER");        return false;    }    // create a connection to the router    if (!(self->router = zsock_new(ZMQ_PUB))) {        error("Cannot create zsock to the router.");        return false;    }    // initialize Redis connection    if (!(self->redis = redisNew(&info->redisInfo))) {        error("Cannot initialize a new Redis connection.");        return false;    }    // initialize hashtable of clients around    if (!(self->clientsGraph = graphNew())) {        error("Cannot allocate a new clients Graph.");        return false;    }    switch (serverType)    {        case SERVER_TYPE_BARRACK:            self->eventServerProcess = barrackEventServerProcess;            break;        case SERVER_TYPE_SOCIAL:            // self->eventServerProcess = socialEventServerProcess;            break;        case SERVER_TYPE_ZONE:            self->eventServerProcess = zoneEventServerProcess;            break;        default:            // No EventServer registred for this server        break;    }    return true;}

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

staticzsock_t* subscriber_push_socket_new(){    zsock_t *socket = zsock_new(ZMQ_PUSH);    assert(socket);    int rc = zsock_bind(socket, "inproc://graylog-forwarder-subscriber");    assert(rc == 0);    return socket;}

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

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

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

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

int main (void) {    //  Create and bind server socket    zsock_t *server = zsock_new (ZMQ_PUSH);    zsock_bind (server, "tcp://*:9000");    //  Create and connect client socket    zsock_t *client = zsock_new (ZMQ_PULL);    zsock_connect (client, "tcp://127.0.0.1:9000");        //  Send a single message from server to client    zstr_send (server, "Hello");    char *message = zstr_recv (client);    assert (streq (message, "Hello"));    free (message);    puts ("Grasslands test OK");        zsock_destroy (&client);    zsock_destroy (&server);    return 0;}