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

int main (void){    setup_test_environment();    void *ctx = zmq_ctx_new ();    assert (ctx);    //  Spawn ZAP handler    void *zap_thread = zmq_threadstart (&zap_handler, ctx);    //  Server socket will accept connections    void *server = zmq_socket (ctx, ZMQ_DEALER);    assert (server);    int rc = zmq_setsockopt (server, ZMQ_IDENTITY, "IDENT", 6);    assert (rc == 0);    int as_server = 1;    rc = zmq_setsockopt (server, ZMQ_PLAIN_SERVER, &as_server, sizeof (int));    assert (rc == 0);    rc = zmq_bind (server, "tcp://127.0.0.1:9998");    assert (rc == 0);    char username [256];    char password [256];    //  Check PLAIN security with correct username/password    void *client = zmq_socket (ctx, ZMQ_DEALER);    assert (client);    strcpy (username, "admin");    rc = zmq_setsockopt (client, ZMQ_PLAIN_USERNAME, username, strlen (username));    assert (rc == 0);    strcpy (password, "password");    rc = zmq_setsockopt (client, ZMQ_PLAIN_PASSWORD, password, strlen (password));    assert (rc == 0);    rc = zmq_connect (client, "tcp://localhost:9998");    assert (rc == 0);    bounce (server, client);    rc = zmq_close (client);    assert (rc == 0);    //  Check PLAIN security with badly configured client (as_server)    //  This will be caught by the plain_server class, not passed to ZAP    client = zmq_socket (ctx, ZMQ_DEALER);    assert (client);    as_server = 1;    rc = zmq_setsockopt (client, ZMQ_PLAIN_SERVER, &as_server, sizeof (int));    assert (rc == 0);    rc = zmq_connect (client, "tcp://localhost:9998");    assert (rc == 0);    expect_bounce_fail (server, client);    close_zero_linger (client);    //  Check PLAIN security -- failed authentication    client = zmq_socket (ctx, ZMQ_DEALER);    assert (client);    strcpy (username, "wronguser");    strcpy (password, "wrongpass");    rc = zmq_setsockopt (client, ZMQ_PLAIN_USERNAME, username, strlen (username));    assert (rc == 0);    rc = zmq_setsockopt (client, ZMQ_PLAIN_PASSWORD, password, strlen (password));    assert (rc == 0);    rc = zmq_connect (client, "tcp://localhost:9998");    assert (rc == 0);    expect_bounce_fail (server, client);    close_zero_linger (client);    //  Shutdown    rc = zmq_close (server);    assert (rc == 0);    rc = zmq_ctx_term (ctx);    assert (rc == 0);    //  Wait until ZAP handler terminates    zmq_threadclose (zap_thread);    return 0;}

//.........这里部分代码省略.........    bounce (server, client);    rc = zmq_close (client);    assert (rc == 0);    //  Check CURVE security with a garbage server key    //  This will be caught by the curve_server class, not passed to ZAP    char garbage_key [] = "0000111122223333444455556666777788889999";    client = zmq_socket (ctx, ZMQ_DEALER);    assert (client);    rc = zmq_setsockopt (client, ZMQ_CURVE_SERVERKEY, garbage_key, 41);    assert (rc == 0);    rc = zmq_setsockopt (client, ZMQ_CURVE_PUBLICKEY, client_public, 41);    assert (rc == 0);    rc = zmq_setsockopt (client, ZMQ_CURVE_SECRETKEY, client_secret, 41);    assert (rc == 0);    rc = zmq_connect (client, "tcp://localhost:9998");    assert (rc == 0);    expect_bounce_fail (server, client);    close_zero_linger (client);    //  Check CURVE security with a garbage client public key    //  This will be caught by the curve_server class, not passed to ZAP    client = zmq_socket (ctx, ZMQ_DEALER);    assert (client);    rc = zmq_setsockopt (client, ZMQ_CURVE_SERVERKEY, server_public, 41);    assert (rc == 0);    rc = zmq_setsockopt (client, ZMQ_CURVE_PUBLICKEY, garbage_key, 41);    assert (rc == 0);    rc = zmq_setsockopt (client, ZMQ_CURVE_SECRETKEY, client_secret, 41);    assert (rc == 0);    rc = zmq_connect (client, "tcp://localhost:9998");    assert (rc == 0);    expect_bounce_fail (server, client);    close_zero_linger (client);    //  Check CURVE security with a garbage client secret key    //  This will be caught by the curve_server class, not passed to ZAP    client = zmq_socket (ctx, ZMQ_DEALER);    assert (client);    rc = zmq_setsockopt (client, ZMQ_CURVE_SERVERKEY, server_public, 41);    assert (rc == 0);    rc = zmq_setsockopt (client, ZMQ_CURVE_PUBLICKEY, client_public, 41);    assert (rc == 0);    rc = zmq_setsockopt (client, ZMQ_CURVE_SECRETKEY, garbage_key, 41);    assert (rc == 0);    rc = zmq_connect (client, "tcp://localhost:9998");    assert (rc == 0);    expect_bounce_fail (server, client);    close_zero_linger (client);    //  Check CURVE security with bogus client credentials    //  This must be caught by the ZAP handler    char bogus_public [41];    char bogus_secret [41];    zmq_curve_keypair (bogus_public, bogus_secret);    client = zmq_socket (ctx, ZMQ_DEALER);    assert (client);    rc = zmq_setsockopt (client, ZMQ_CURVE_SERVERKEY, server_public, 41);    assert (rc == 0);    rc = zmq_setsockopt (client, ZMQ_CURVE_PUBLICKEY, bogus_public, 41);    assert (rc == 0);    rc = zmq_setsockopt (client, ZMQ_CURVE_SECRETKEY, bogus_secret, 41);    assert (rc == 0);    rc = zmq_connect (client, "tcp://localhost:9998");    assert (rc == 0);    expect_bounce_fail (server, client);    close_zero_linger (client);    //  Check CURVE security with NULL client credentials    //  This must be caught by the curve_server class, not passed to ZAP    client = zmq_socket (ctx, ZMQ_DEALER);    assert (client);    rc = zmq_connect (client, "tcp://localhost:9998");    assert (rc == 0);    expect_bounce_fail (server, client);    close_zero_linger (client);    //  Check CURVE security with PLAIN client credentials    //  This must be caught by the curve_server class, not passed to ZAP    client = zmq_socket (ctx, ZMQ_DEALER);    assert (client);    rc = zmq_setsockopt (client, ZMQ_PLAIN_USERNAME, "admin", 5);    assert (rc == 0);    rc = zmq_setsockopt (client, ZMQ_PLAIN_PASSWORD, "password", 8);    assert (rc == 0);    expect_bounce_fail (server, client);    close_zero_linger (client);        //  Shutdown    rc = zmq_close (server);    assert (rc == 0);    rc = zmq_ctx_term (ctx);    assert (rc == 0);    //  Wait until ZAP handler terminates    zmq_threadclose (zap_thread);    return 0;}

int main (void){    setup_test_environment ();    void *ctx = zmq_ctx_new ();    assert (ctx);    g_clients_pkts_out = zmq_atomic_counter_new ();    g_workers_pkts_out = zmq_atomic_counter_new ();    // Control socket receives terminate command from main over inproc    void *control = zmq_socket (ctx, ZMQ_PUB);    assert (control);    int linger = 0;    int rc = zmq_setsockopt (control, ZMQ_LINGER, &linger, sizeof (linger));    assert (rc == 0);    rc = zmq_bind (control, "inproc://control");    assert (rc == 0);    // Control socket receives terminate command from main over inproc    void *control_proxy = zmq_socket (ctx, ZMQ_REQ);    assert (control_proxy);    rc = zmq_setsockopt (control_proxy, ZMQ_LINGER, &linger, sizeof (linger));    assert (rc == 0);    rc = zmq_bind (control_proxy, "inproc://control_proxy");    assert (rc == 0);    void *threads[QT_CLIENTS + 1];    struct thread_data databags[QT_CLIENTS + 1];    for (int i = 0; i < QT_CLIENTS; i++) {        databags[i].ctx = ctx;        databags[i].id = i;        threads[i] = zmq_threadstart (&client_task, &databags[i]);    }    threads[QT_CLIENTS] = zmq_threadstart (&server_task, ctx);    msleep (500); // Run for 500 ms then quit    if (is_verbose)        printf ("stopping all clients and server workers/n");    rc = zmq_send (control, "STOP", 4, 0);    assert (rc == 4);    msleep (500); // Wait for all clients and workers to STOP#ifdef ZMQ_BUILD_DRAFT_API    if (is_verbose)        printf ("retrieving stats from the proxy/n");    check_proxy_stats (control_proxy);#endif    if (is_verbose)        printf ("shutting down all clients and server workers/n");    rc = zmq_send (control, "TERMINATE", 9, 0);    assert (rc == 9);    if (is_verbose)        printf ("shutting down the proxy/n");    rc = zmq_send (control_proxy, "TERMINATE", 9, 0);    assert (rc == 9);    rc = zmq_close (control);    assert (rc == 0);    rc = zmq_close (control_proxy);    assert (rc == 0);    for (int i = 0; i < QT_CLIENTS + 1; i++)        zmq_threadclose (threads[i]);    rc = zmq_ctx_term (ctx);    assert (rc == 0);    return 0;}

//.........这里部分代码省略.........    rc = zmq_poller_add (poller, sink, sink, ZMQ_POLLIN);    assert (rc == 0);    //  Send a message    char data[1] = {'H'};    rc = zmq_send_const (vent, data, 1, 0);    assert (rc == 1);    //  We expect a message only on the sink    rc = zmq_poller_wait (poller, &event, -1);    assert (rc == 0);    assert (event.socket == sink);    assert (event.user_data == sink);    rc = zmq_recv (sink, data, 1, 0);    assert (rc == 1);    //  We expect timed out    rc = zmq_poller_wait (poller, &event, 0);    assert (rc == -1);    assert (errno == EAGAIN);    //  Stop polling sink    rc = zmq_poller_remove (poller, sink);    assert (rc == 0);    //  Check we can poll an FD    rc = zmq_connect (bowl, my_endpoint_0);    assert (rc == 0);    fd_t fd;    size_t fd_size = sizeof (fd);    rc = zmq_getsockopt (bowl, ZMQ_FD, &fd, &fd_size);    assert (rc == 0);    rc = zmq_poller_add_fd (poller, fd, bowl, ZMQ_POLLIN);    assert (rc == 0);    rc = zmq_poller_wait (poller, &event, 500);    assert (rc == 0);    assert (event.socket == NULL);    assert (event.fd == fd);    assert (event.user_data == bowl);    zmq_poller_remove_fd (poller, fd);#if defined(ZMQ_SERVER) && defined(ZMQ_CLIENT)    //  Polling on thread safe sockets    rc = zmq_poller_add (poller, server, NULL, ZMQ_POLLIN);    assert (rc == 0);    rc = zmq_connect (client, my_endpoint_1);    assert (rc == 0);    rc = zmq_send_const (client, data, 1, 0);    assert (rc == 1);    rc = zmq_poller_wait (poller, &event, 500);    assert (rc == 0);    assert (event.socket == server);    assert (event.user_data == NULL);    rc = zmq_recv (server, data, 1, 0);    assert (rc == 1);    //  Polling on pollout    rc = zmq_poller_modify (poller, server, ZMQ_POLLOUT | ZMQ_POLLIN);    assert (rc == 0);    rc = zmq_poller_wait (poller, &event, 0);    assert (rc == 0);    assert (event.socket == server);    assert (event.user_data == NULL);    assert (event.events == ZMQ_POLLOUT);    //  Stop polling server    rc = zmq_poller_remove (poller, server);    assert (rc == 0);#endif    //  Destroy sockets, poller and ctx    rc = zmq_close (sink);    assert (rc == 0);    rc = zmq_close (vent);    assert (rc == 0);    rc = zmq_close (bowl);    assert (rc == 0);#if defined(ZMQ_SERVER) && defined(ZMQ_CLIENT)    rc = zmq_close (server);    assert (rc == 0);    rc = zmq_close (client);    assert (rc == 0);#endif    test_null_poller_pointers (ctx);    test_null_socket_pointers ();    test_null_event_pointers (ctx);    test_add_modify_remove_corner_cases (ctx);    test_wait_corner_cases ();    rc = zmq_poller_destroy (&poller);    assert (rc == 0);    rc = zmq_ctx_term (ctx);    assert (rc == 0);    return 0;}

int main (void){    setup_test_environment();    void *ctx = zmq_ctx_new ();    assert (ctx);    //  Spawn ZAP handler    //  We create and bind ZAP socket in main thread to avoid case    //  where child thread does not start up fast enough.    void *handler = zmq_socket (ctx, ZMQ_REP);    assert (handler);    int rc = zmq_bind (handler, "inproc://zeromq.zap.01");    assert (rc == 0);    void *zap_thread = zmq_threadstart (&zap_handler, handler);    //  We bounce between a binding server and a connecting client    //     //  We first test client/server with no ZAP domain//     //  Libzmq does not call our ZAP handler, the connect must succeed//     void *server = zmq_socket (ctx, ZMQ_DEALER);//     assert (server);//     void *client = zmq_socket (ctx, ZMQ_DEALER);//     assert (client);//     rc = zmq_bind (server, "tcp://127.0.0.1:9000");//     assert (rc == 0);//     rc = zmq_connect (client, "tcp://127.0.0.1:9000");//     assert (rc == 0);//     bounce (server, client);//     close_zero_linger (client);//     close_zero_linger (server);    //  Now define a ZAP domain for the server; this enables     //  authentication. We're using the wrong domain so this test    //  must fail.    void *server = zmq_socket (ctx, ZMQ_DEALER);    assert (server);    void *client = zmq_socket (ctx, ZMQ_DEALER);    assert (client);    rc = zmq_setsockopt (server, ZMQ_ZAP_DOMAIN, "WRONG", 5);    assert (rc == 0);    rc = zmq_bind (server, "tcp://127.0.0.1:9001");    assert (rc == 0);    rc = zmq_connect (client, "tcp://127.0.0.1:9001");    assert (rc == 0);    expect_bounce_fail (server, client);    close_zero_linger (client);    close_zero_linger (server);//     //  Now use the right domain, the test must pass//     server = zmq_socket (ctx, ZMQ_DEALER);//     assert (server);//     client = zmq_socket (ctx, ZMQ_DEALER);//     assert (client);//     rc = zmq_setsockopt (server, ZMQ_ZAP_DOMAIN, "TEST", 4);//     assert (rc == 0);//     rc = zmq_bind (server, "tcp://127.0.0.1:9002");//     assert (rc == 0);//     rc = zmq_connect (client, "tcp://127.0.0.1:9002");//     assert (rc == 0);//     bounce (server, client);//     close_zero_linger (client);//     close_zero_linger (server);    //  Shutdown    rc = zmq_ctx_term (ctx);    assert (rc == 0);    //  Wait until ZAP handler terminates    zmq_threadclose (zap_thread);    return 0;}

int zmq_ctx_destroy (void *ctx_){    return zmq_ctx_term (ctx_);}

int main (void){    setup_test_environment();    size_t len = MAX_SOCKET_STRING;    char endpoint1[MAX_SOCKET_STRING];    char endpoint2[MAX_SOCKET_STRING];    void *ctx = zmq_ctx_new ();    assert (ctx);    //  First, create an intermediate device    void *xpub = zmq_socket (ctx, ZMQ_XPUB);    assert (xpub);    int rc = zmq_bind (xpub, "tcp://127.0.0.1:*");    assert (rc == 0);    rc = zmq_getsockopt (xpub, ZMQ_LAST_ENDPOINT, endpoint1, &len);    assert (rc == 0);    void *xsub = zmq_socket (ctx, ZMQ_XSUB);    assert (xsub);    rc = zmq_bind (xsub, "tcp://127.0.0.1:*");    assert (rc == 0);    len = MAX_SOCKET_STRING;    rc = zmq_getsockopt (xsub, ZMQ_LAST_ENDPOINT, endpoint2, &len);    assert (rc == 0);    //  Create a publisher    void *pub = zmq_socket (ctx, ZMQ_PUB);    assert (pub);    rc = zmq_connect (pub, endpoint2);    assert (rc == 0);    //  Create a subscriber    void *sub = zmq_socket (ctx, ZMQ_SUB);    assert (sub);    rc = zmq_connect (sub, endpoint1);    assert (rc == 0);    //  Subscribe for all messages.    rc = zmq_setsockopt (sub, ZMQ_SUBSCRIBE, "", 0);    assert (rc == 0);    //  Pass the subscription upstream through the device    char buff [32];    rc = zmq_recv (xpub, buff, sizeof (buff), 0);    assert (rc >= 0);    rc = zmq_send (xsub, buff, rc, 0);    assert (rc >= 0);    //  Wait a bit till the subscription gets to the publisher    msleep (SETTLE_TIME);    //  Send an empty message    rc = zmq_send (pub, NULL, 0, 0);    assert (rc == 0);    //  Pass the message downstream through the device    rc = zmq_recv (xsub, buff, sizeof (buff), 0);    assert (rc >= 0);    rc = zmq_send (xpub, buff, rc, 0);    assert (rc >= 0);    //  Receive the message in the subscriber    rc = zmq_recv (sub, buff, sizeof (buff), 0);    assert (rc == 0);    //  Clean up.    rc = zmq_close (xpub);    assert (rc == 0);    rc = zmq_close (xsub);    assert (rc == 0);    rc = zmq_close (pub);    assert (rc == 0);    rc = zmq_close (sub);    assert (rc == 0);    rc = zmq_ctx_term (ctx);    assert (rc == 0);    return 0 ;}

int main (void){    void *ctx = zmq_ctx_new ();    assert (ctx);    //  First, create an intermediate device    void *xpub = zmq_socket (ctx, ZMQ_XPUB);    assert (xpub);    int rc = zmq_bind (xpub, "tcp://127.0.0.1:5560");    assert (rc == 0);    void *xsub = zmq_socket (ctx, ZMQ_XSUB);    assert (xsub);    rc = zmq_bind (xsub, "tcp://127.0.0.1:5561");    assert (rc == 0);    //  Create a publisher    void *pub = zmq_socket (ctx, ZMQ_PUB);    assert (pub);    rc = zmq_connect (pub, "tcp://127.0.0.1:5561");    assert (rc == 0);    //  Create a subscriber    void *sub = zmq_socket (ctx, ZMQ_SUB);    assert (sub);    rc = zmq_connect (sub, "tcp://127.0.0.1:5560");    assert (rc == 0);    //  Subscribe for all messages.    rc = zmq_setsockopt (sub, ZMQ_SUBSCRIBE, "", 0);    assert (rc == 0);    //  Pass the subscription upstream through the device    char buff [32];    rc = zmq_recv (xpub, buff, sizeof (buff), 0);    assert (rc >= 0);    rc = zmq_send (xsub, buff, rc, 0);    assert (rc >= 0);    //  Wait a bit till the subscription gets to the publisher    struct timespec t = { 0, 250 * 1000000 };    nanosleep (&t, NULL);    //  Send an empty message    rc = zmq_send (pub, NULL, 0, 0);    assert (rc == 0);    //  Pass the message downstream through the device    rc = zmq_recv (xsub, buff, sizeof (buff), 0);    assert (rc >= 0);    rc = zmq_send (xpub, buff, rc, 0);    assert (rc >= 0);    //  Receive the message in the subscriber    rc = zmq_recv (sub, buff, sizeof (buff), 0);    assert (rc == 0);    //  Clean up.    rc = zmq_close (xpub);    assert (rc == 0);    rc = zmq_close (xsub);    assert (rc == 0);    rc = zmq_close (pub);    assert (rc == 0);    rc = zmq_close (sub);    assert (rc == 0);    rc = zmq_ctx_term (ctx);    assert (rc == 0);    return 0 ;}

int main (int argc, char *argv[]){    if (argc != 3) {        printf ("usage: inproc_thr <message-size> <message-count>/n");        return 1;    }    message_size = atoi (argv[1]);    message_count = atoi (argv[2]);    printf ("message size: %d [B]/n", (int) message_size);    printf ("message count: %d/n", (int) message_count);    void *context = zmq_ctx_new ();    assert (context);    int rv = zmq_ctx_set (context, ZMQ_IO_THREADS, 4);    assert (rv == 0);    //  START ALL SECONDARY THREADS    const char *pub1 = "inproc://perf_pub1";    const char *pub2 = "inproc://perf_pub2";    const char *sub1 = "inproc://perf_backend";    proxy_hwm_cfg_t cfg_global = {};    cfg_global.context = context;    cfg_global.frontend_endpoint[0] = pub1;    cfg_global.frontend_endpoint[1] = pub2;    cfg_global.backend_endpoint[0] = sub1;    cfg_global.control_endpoint = "inproc://ctrl";    //  Proxy    proxy_hwm_cfg_t cfg_proxy = cfg_global;    void *proxy = zmq_threadstart (&proxy_thread_main, (void *) &cfg_proxy);    assert (proxy != 0);    //  Subscriber 1    proxy_hwm_cfg_t cfg_sub1 = cfg_global;    cfg_sub1.thread_idx = 0;    void *subscriber =      zmq_threadstart (&subscriber_thread_main, (void *) &cfg_sub1);    assert (subscriber != 0);    //  Start measuring    void *watch = zmq_stopwatch_start ();    //  Publisher 1    proxy_hwm_cfg_t cfg_pub1 = cfg_global;    cfg_pub1.thread_idx = 0;    void *publisher1 =      zmq_threadstart (&publisher_thread_main, (void *) &cfg_pub1);    assert (publisher1 != 0);    //  Publisher 2    proxy_hwm_cfg_t cfg_pub2 = cfg_global;    cfg_pub2.thread_idx = 1;    void *publisher2 =      zmq_threadstart (&publisher_thread_main, (void *) &cfg_pub2);    assert (publisher2 != 0);    //  Wait for all packets to be received    zmq_threadclose (subscriber);    //  Stop measuring    unsigned long elapsed = zmq_stopwatch_stop (watch);    if (elapsed == 0)        elapsed = 1;    unsigned long throughput =      (unsigned long) ((double) message_count / (double) elapsed * 1000000);    double megabits = (double) (throughput * message_size * 8) / 1000000;    printf ("mean throughput: %d [msg/s]/n", (int) throughput);    printf ("mean throughput: %.3f [Mb/s]/n", (double) megabits);    //  Wait for the end of publishers...    zmq_threadclose (publisher1);    zmq_threadclose (publisher2);    //  ... then close the proxy    terminate_proxy (&cfg_proxy);    zmq_threadclose (proxy);    int rc = zmq_ctx_term (context);    assert (rc == 0);    return 0;}

int main (void){    setup_test_environment();    void *ctx = zmq_ctx_new ();    assert (ctx);    //  Spawn ZAP handler    void *zap_thread = zmq_threadstart (&zap_handler, ctx);    //  Server socket will accept connections    void *server = zmq_socket (ctx, ZMQ_DEALER);    assert (server);    int rc = zmq_setsockopt (server, ZMQ_IDENTITY, "IDENT", 6);    assert (rc == 0);    int as_server = 1;    rc = zmq_setsockopt (server, ZMQ_PLAIN_SERVER, &as_server, sizeof (int));    assert (rc == 0);    rc = zmq_bind (server, "tcp://127.0.0.1:9998");    assert (rc == 0);    char username [256];    char password [256];    //  Check PLAIN security with correct username/password    void *client = zmq_socket (ctx, ZMQ_DEALER);    assert (client);    strcpy (username, "admin");    rc = zmq_setsockopt (client, ZMQ_PLAIN_USERNAME, username, strlen (username));    assert (rc == 0);    strcpy (password, "password");    rc = zmq_setsockopt (client, ZMQ_PLAIN_PASSWORD, password, strlen (password));    assert (rc == 0);    rc = zmq_connect (client, "tcp://localhost:9998");    assert (rc == 0);    bounce (server, client);    rc = zmq_close (client);    assert (rc == 0);    //  Check PLAIN security with badly configured client (as_server)    //  This will be caught by the plain_server class, not passed to ZAP    client = zmq_socket (ctx, ZMQ_DEALER);    assert (client);    as_server = 1;    rc = zmq_setsockopt (client, ZMQ_PLAIN_SERVER, &as_server, sizeof (int));    assert (rc == 0);    rc = zmq_connect (client, "tcp://localhost:9998");    assert (rc == 0);    expect_bounce_fail (server, client);    close_zero_linger (client);    //  Check PLAIN security -- failed authentication    client = zmq_socket (ctx, ZMQ_DEALER);    assert (client);    strcpy (username, "wronguser");    strcpy (password, "wrongpass");    rc = zmq_setsockopt (client, ZMQ_PLAIN_USERNAME, username, strlen (username));    assert (rc == 0);    rc = zmq_setsockopt (client, ZMQ_PLAIN_PASSWORD, password, strlen (password));    assert (rc == 0);    rc = zmq_connect (client, "tcp://localhost:9998");    assert (rc == 0);    expect_bounce_fail (server, client);    close_zero_linger (client);    // Unauthenticated messages from a vanilla socket shouldn't be received    struct sockaddr_in ip4addr;    int s;    ip4addr.sin_family = AF_INET;    ip4addr.sin_port = htons (9998);    inet_pton (AF_INET, "127.0.0.1", &ip4addr.sin_addr);    s = socket (AF_INET, SOCK_STREAM, IPPROTO_TCP);    rc = connect (s, (struct sockaddr*) &ip4addr, sizeof (ip4addr));    assert (rc > -1);    // send anonymous ZMTP/1.0 greeting    send (s, "/x01/x00", 2, 0);    // send sneaky message that shouldn't be received    send (s, "/x08/x00sneaky/0", 9, 0);    int timeout = 250;    zmq_setsockopt (server, ZMQ_RCVTIMEO, &timeout, sizeof (timeout));    char *buf = s_recv (server);    if (buf != NULL) {        printf ("Received unauthenticated message: %s/n", buf);        assert (buf == NULL);    }    close (s);    //  Shutdown    rc = zmq_close (server);    assert (rc == 0);    rc = zmq_ctx_term (ctx);    assert (rc == 0);    //  Wait until ZAP handler terminates    zmq_threadclose (zap_thread);    return 0;}

//.........这里部分代码省略.........    void *dealer = zmq_socket (ctx, ZMQ_DEALER);    assert (dealer);    rc = zmq_setsockopt (dealer, ZMQ_LINGER, &zero, sizeof (zero));    assert (rc == 0);    rc = zmq_connect (dealer, "tcp://localhost:5555");    //  Send a message on the dealer socket    rc = zmq_send (dealer, "Hello", 5, 0);    assert (rc == 5);    //  First frame is identity    zmq_msg_t identity;    rc = zmq_msg_init (&identity);    assert (rc == 0);    rc = zmq_msg_recv (&identity, router, 0);    assert (rc > 0);    assert (zmq_msg_more (&identity));    //  Second frame is greeting signature    byte buffer [255];    rc = zmq_recv (router, buffer, 255, 0);    assert (rc == 10);    assert (memcmp (buffer, greeting.signature, 10) == 0);    //  Send our own protocol greeting    rc = zmq_msg_send (&identity, router, ZMQ_SNDMORE);    assert (rc > 0);    rc = zmq_send (router, &greeting, sizeof (greeting), 0);    assert (rc == sizeof (greeting));    //  Now we expect the data from the DEALER socket    //  First frame is, again, the identity of the connection    rc = zmq_msg_recv (&identity, router, 0);    assert (rc > 0);    assert (zmq_msg_more (&identity));    //  Second frame contains the rest of greeting along with    //  the Ready command    rc = zmq_recv (router, buffer, 255, 0);    assert (rc == 97);    //  First two bytes are major and minor version numbers.    assert (buffer [0] == 3);       //  ZMTP/3.0    assert (buffer [1] == 0);    //  Mechanism is "NULL"    assert (memcmp (buffer + 2, "NULL/0/0/0/0/0/0/0/0/0/0/0/0/0/0/0/0", 22) == 0);    assert (memcmp (buffer + 54, "/0/51READY/0", 8) == 0);    assert (memcmp (buffer + 62, "/13Socket-Type/0/0/0/6DEALER", 22) == 0);    assert (memcmp (buffer + 84, "/10Identity/0/0/0/0", 13) == 0);    //  Announce we are ready    memcpy (buffer, "/0/51READY/0", 8);    memcpy (buffer + 8, "/13Socket-Type/0/0/0/6STREAM", 22);    memcpy (buffer + 30, "/10Identity/0/0/0/0", 13);    //  Send Ready command    rc = zmq_msg_send (&identity, router, ZMQ_SNDMORE);    assert (rc > 0);    rc = zmq_send (router, buffer, 43, 0);    assert (rc == 43);    //  Now we expect the data from the DEALER socket    //  First frame is, again, the identity of the connection    rc = zmq_msg_recv (&identity, router, 0);    assert (rc > 0);    assert (zmq_msg_more (&identity));    //  Third frame contains Hello message from DEALER    rc = zmq_recv (router, buffer, sizeof buffer, 0);    assert (rc == 7);    //  Then we have a 5-byte message "Hello"    assert (buffer [0] == 0);       //  Flags = 0    assert (buffer [1] == 5);       //  Size = 5    assert (memcmp (buffer + 2, "Hello", 5) == 0);    //  Send "World" back to DEALER    rc = zmq_msg_send (&identity, router, ZMQ_SNDMORE);    assert (rc > 0);    byte world [] = { 0, 5, 'W', 'o', 'r', 'l', 'd' };    rc = zmq_send (router, world, sizeof (world), 0);    assert (rc == sizeof (world));    //  Expect response on DEALER socket    rc = zmq_recv (dealer, buffer, 255, 0);    assert (rc == 5);    assert (memcmp (buffer, "World", 5) == 0);    rc = zmq_close (dealer);    assert (rc == 0);    rc = zmq_close (router);    assert (rc == 0);    rc = zmq_ctx_term (ctx);    assert (rc == 0);    return 0;}

static voidtest_stream_handshake_timeout_accept (void){    int rc;    //  Set up our context and sockets    void *ctx = zmq_ctx_new ();    assert (ctx);    //  We use this socket in raw mode, to make a connection and send nothing    void *stream = zmq_socket (ctx, ZMQ_STREAM);    assert (stream);    int zero = 0;    rc = zmq_setsockopt (stream, ZMQ_LINGER, &zero, sizeof (zero));    assert (rc == 0);    rc = zmq_connect (stream, "tcp://localhost:5557");    assert (rc == 0);    //  We'll be using this socket to test TCP stream handshake timeout    void *dealer = zmq_socket (ctx, ZMQ_DEALER);    assert (dealer);    rc = zmq_setsockopt (dealer, ZMQ_LINGER, &zero, sizeof (zero));    assert (rc == 0);    int val, tenth = 100;    size_t vsize = sizeof(val);    // check for the expected default handshake timeout value - 30 sec    rc = zmq_getsockopt (dealer, ZMQ_HANDSHAKE_IVL, &val, &vsize);    assert (rc == 0);    assert (vsize == sizeof(val));    assert (val == 30000);    // make handshake timeout faster - 1/10 sec    rc = zmq_setsockopt (dealer, ZMQ_HANDSHAKE_IVL, &tenth, sizeof (tenth));    assert (rc == 0);    vsize = sizeof(val);    // make sure zmq_setsockopt changed the value    rc = zmq_getsockopt (dealer, ZMQ_HANDSHAKE_IVL, &val, &vsize);    assert (rc == 0);    assert (vsize == sizeof(val));    assert (val == tenth);    //  Create and connect a socket for collecting monitor events on dealer    void *dealer_mon = zmq_socket (ctx, ZMQ_PAIR);    assert (dealer_mon);    rc = zmq_socket_monitor (dealer, "inproc://monitor-dealer",          ZMQ_EVENT_CONNECTED | ZMQ_EVENT_DISCONNECTED | ZMQ_EVENT_ACCEPTED);    assert (rc == 0);    //  Connect to the inproc endpoint so we'll get events    rc = zmq_connect (dealer_mon, "inproc://monitor-dealer");    assert (rc == 0);    // bind dealer socket to accept connection from non-sending stream socket    rc = zmq_bind (dealer, "tcp://127.0.0.1:5557");    assert (rc == 0);    // we should get ZMQ_EVENT_ACCEPTED and then ZMQ_EVENT_DISCONNECTED    int event = get_monitor_event (dealer_mon, NULL, NULL);    assert (event == ZMQ_EVENT_ACCEPTED);    event = get_monitor_event (dealer_mon, NULL, NULL);    assert (event == ZMQ_EVENT_DISCONNECTED);    rc = zmq_close (dealer);    assert (rc == 0);    rc = zmq_close (dealer_mon);    assert (rc == 0);    rc = zmq_close (stream);    assert (rc == 0);    rc = zmq_ctx_term (ctx);    assert (rc == 0);}

int main (void){    setup_test_environment();    int val;    int rc;    char buffer[16];    // TEST 1.     // First we're going to attempt to send messages to two    // pipes, one connected, the other not. We should see    // the PUSH load balancing to both pipes, and hence half    // of the messages getting queued, as connect() creates a    // pipe immediately.         void *context = zmq_ctx_new();    assert (context);    void *to = zmq_socket(context, ZMQ_PULL);    assert (to);    // Bind the one valid receiver    val = 0;    rc = zmq_setsockopt(to, ZMQ_LINGER, &val, sizeof(val));    assert (rc == 0);    rc = zmq_bind (to, "tcp://127.0.0.1:6555");    assert (rc == 0);    // Create a socket pushing to two endpoints - only 1 message should arrive.    void *from = zmq_socket (context, ZMQ_PUSH);    assert(from);    val = 0;    zmq_setsockopt (from, ZMQ_LINGER, &val, sizeof (val));    // This pipe will not connect    rc = zmq_connect (from, "tcp://localhost:5556");    assert (rc == 0);    // This pipe will     rc = zmq_connect (from, "tcp://localhost:6555");    assert (rc == 0);    // We send 10 messages, 5 should just get stuck in the queue    // for the not-yet-connected pipe    for (int i = 0; i < 10; ++i) {        rc = zmq_send (from, "Hello", 5, 0);        assert (rc == 5);    }    // We now consume from the connected pipe    // - we should see just 5    int timeout = 250;    rc = zmq_setsockopt (to, ZMQ_RCVTIMEO, &timeout, sizeof (int));    assert (rc == 0);    int seen = 0;    while (true) {        rc = zmq_recv (to, &buffer, sizeof (buffer), 0);        if (rc == -1)            break;          //  Break when we didn't get a message        seen++;    }    assert (seen == 5);    rc = zmq_close (from);    assert (rc == 0);    rc = zmq_close (to);    assert (rc == 0);    rc = zmq_ctx_term (context);    assert (rc == 0);    // TEST 2    // This time we will do the same thing, connect two pipes,     // one of which will succeed in connecting to a bound     // receiver, the other of which will fail. However, we will     // also set the delay attach on connect flag, which should     // cause the pipe attachment to be delayed until the connection    // succeeds.     context = zmq_ctx_new();    // Bind the valid socket    to = zmq_socket (context, ZMQ_PULL);    assert (to);    rc = zmq_bind (to, "tcp://127.0.0.1:5560");    assert (rc == 0);    val = 0;    rc = zmq_setsockopt (to, ZMQ_LINGER, &val, sizeof(val));    assert (rc == 0);    // Create a socket pushing to two endpoints - all messages should arrive.    from = zmq_socket (context, ZMQ_PUSH);    assert (from);    val = 0;    rc = zmq_setsockopt (from, ZMQ_LINGER, &val, sizeof(val));    assert (rc == 0);    // Set the key flag    val = 1;    rc = zmq_setsockopt (from, ZMQ_IMMEDIATE, &val, sizeof(val));    assert (rc == 0);//.........这里部分代码省略.........

//.........这里部分代码省略.........    if (!s) {        printf ("error in zmq_socket: %s/n", zmq_strerror (errno));        return -1;    }    rc = zmq_bind (s, "inproc://lat_test");    if (rc != 0) {        printf ("error in zmq_bind: %s/n", zmq_strerror (errno));        return -1;    }#if defined ZMQ_HAVE_WINDOWS    local_thread = (HANDLE) _beginthreadex (NULL, 0,        worker, ctx, 0 , NULL);    if (local_thread == 0) {        printf ("error in _beginthreadex/n");        return -1;    }#else    rc = pthread_create (&local_thread, NULL, worker, ctx);    if (rc != 0) {        printf ("error in pthread_create: %s/n", zmq_strerror (rc));        return -1;    }#endif    rc = zmq_msg_init_size (&msg, message_size);    if (rc != 0) {        printf ("error in zmq_msg_init_size: %s/n", zmq_strerror (errno));        return -1;    }    memset (zmq_msg_data (&msg), 0, message_size);    printf ("message size: %d [B]/n", (int) message_size);    printf ("roundtrip count: %d/n", (int) roundtrip_count);    watch = zmq_stopwatch_start ();    for (i = 0; i != roundtrip_count; i++) {        rc = zmq_sendmsg (s, &msg, 0);        if (rc < 0) {            printf ("error in zmq_sendmsg: %s/n", zmq_strerror (errno));            return -1;        }        rc = zmq_recvmsg (s, &msg, 0);        if (rc < 0) {            printf ("error in zmq_recvmsg: %s/n", zmq_strerror (errno));            return -1;        }        if (zmq_msg_size (&msg) != message_size) {            printf ("message of incorrect size received/n");            return -1;        }    }    elapsed = zmq_stopwatch_stop (watch);    rc = zmq_msg_close (&msg);    if (rc != 0) {        printf ("error in zmq_msg_close: %s/n", zmq_strerror (errno));        return -1;    }    latency = (double) elapsed / (roundtrip_count * 2);#if defined ZMQ_HAVE_WINDOWS    DWORD rc2 = WaitForSingleObject (local_thread, INFINITE);    if (rc2 == WAIT_FAILED) {        printf ("error in WaitForSingleObject/n");        return -1;    }    BOOL rc3 = CloseHandle (local_thread);    if (rc3 == 0) {        printf ("error in CloseHandle/n");        return -1;    }#else    rc = pthread_join (local_thread, NULL);    if (rc != 0) {        printf ("error in pthread_join: %s/n", zmq_strerror (rc));        return -1;    }#endif    printf ("average latency: %.3f [us]/n", (double) latency);    rc = zmq_close (s);    if (rc != 0) {        printf ("error in zmq_close: %s/n", zmq_strerror (errno));        return -1;    }    rc = zmq_ctx_term (ctx);    if (rc != 0) {        printf ("error in zmq_ctx_term: %s/n", zmq_strerror (errno));        return -1;    }    return 0;}

int main(int, char**) {    setup_test_environment();    void* context = zmq_ctx_new();    void* pubSocket;    void* subSocket;    (pubSocket = zmq_socket(context, ZMQ_XPUB))         || printf("zmq_socket: %s/n", zmq_strerror(errno));    (subSocket = zmq_socket(context, ZMQ_SUB))          || printf("zmq_socket: %s/n", zmq_strerror(errno));    zmq_setsockopt(subSocket, ZMQ_SUBSCRIBE, "foo", 3)  && printf("zmq_setsockopt: %s/n",zmq_strerror(errno));      zmq_bind(pubSocket, "inproc://someInProcDescriptor") && printf("zmq_bind: %s/n", zmq_strerror(errno));    //zmq_bind(pubSocket, "tcp://127.0.0.1:30010") && printf("zmq_bind: %s/n", zmq_strerror(errno));      int more;    size_t more_size = sizeof(more);    int iteration = 0;      while (1) {        zmq_pollitem_t items [] = {            { subSocket, 0, ZMQ_POLLIN, 0 }, // read publications            { pubSocket, 0, ZMQ_POLLIN, 0 }, // read subscriptions        };        int rc = zmq_poll (items, 2, 100);            if (items [1].revents & ZMQ_POLLIN) {            while (1) {                zmq_msg_t msg;                zmq_msg_init (&msg);                zmq_msg_recv (&msg, pubSocket, 0);                char* buffer = (char*)zmq_msg_data(&msg);                if (buffer[0] == 0) {                    assert(isSubscribed);                    isSubscribed = false;                }                 else {                    assert(!isSubscribed);                    isSubscribed = true;                }                zmq_getsockopt (pubSocket, ZMQ_RCVMORE, &more, &more_size);                zmq_msg_close (&msg);                if (!more)                    break;      //  Last message part            }        }        if (items[0].revents & ZMQ_POLLIN) {            while (1) {                zmq_msg_t msg;                zmq_msg_init (&msg);                zmq_msg_recv (&msg, subSocket, 0);                zmq_getsockopt (subSocket, ZMQ_RCVMORE, &more, &more_size);                zmq_msg_close (&msg);                        if (!more) {                    publicationsReceived++;                    break;      //  Last message part                }            }        }        if (iteration == 1) {            zmq_connect(subSocket, "inproc://someInProcDescriptor") && printf("zmq_connect: %s/n", zmq_strerror(errno));            msleep (SETTLE_TIME);        }        if (iteration == 4) {            zmq_disconnect(subSocket, "inproc://someInProcDescriptor") && printf("zmq_disconnect(%d): %s/n", errno, zmq_strerror(errno));        }        if (iteration > 4 && rc == 0)            break;                zmq_msg_t channelEnvlp;        ZMQ_PREPARE_STRING(channelEnvlp, "foo", 3);        zmq_msg_send (&channelEnvlp, pubSocket, ZMQ_SNDMORE) >= 0 || printf("zmq_msg_send: %s/n",zmq_strerror(errno));        zmq_msg_close(&channelEnvlp) && printf("zmq_msg_close: %s/n",zmq_strerror(errno));        zmq_msg_t message;        ZMQ_PREPARE_STRING(message, "this is foo!", 12);        zmq_msg_send (&message, pubSocket, 0) >= 0 || printf("zmq_msg_send: %s/n",zmq_strerror(errno));        zmq_msg_close(&message) && printf("zmq_msg_close: %s/n",zmq_strerror(errno));        iteration++;    }    assert(publicationsReceived == 3);    assert(!isSubscribed);    zmq_close(pubSocket) && printf("zmq_close: %s", zmq_strerror(errno));    zmq_close(subSocket) && printf("zmq_close: %s", zmq_strerror(errno));      zmq_ctx_term(context);    return 0;}

int main (void){#ifndef HAVE_LIBSODIUM    printf ("libsodium not installed, skipping CURVE test/n");    return 0;#endif    int rc;    size_t optsize;    int mechanism;    int as_server;        void *ctx = zmq_ctx_new ();    assert (ctx);    //  Server socket will accept connections    void *server = zmq_socket (ctx, ZMQ_DEALER);    assert (server);    //  Client socket that will try to connect to server    void *client = zmq_socket (ctx, ZMQ_DEALER);    assert (client);    //  Test keys from the zmq_curve man page    char client_public [] = "[email
C++ zmq_errno函数代码示例
C++ zmq_ctx_new函数代码示例