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

int main(void) {  // Socked to talk to clients  printf("Connecting to the server/n");  void *context = zmq_ctx_new();  void *requester = zmq_socket(context, ZMQ_REQ);  zmq_connect(requester, "tcp://localhost:5555");  int request_nbr;  for (request_nbr = 0; request_nbr < 10; request_nbr++) {    char buffer[10];    printf("Sending Hello %d.../n", request_nbr);    zmq_send(requester, "Hello", 5, 0);    zmq_recv(requester, buffer, 10, 0);    printf("Received World %d/n", request_nbr);  }  zmq_close(requester);  zmq_ctx_destroy(context);  return 0;}

int main(){	void *context = zmq_ctx_new();	// Socket to talk to server	void *requester = zmq_socket(context, ZMQ_REQ);	zmq_connect(requester, "tcp://localhost:5559");	int request_nbr;	for (request_nbr = 0; request_nbr != 10; ++request_nbr)	{		s_send(requester, "Hello");		char *string = s_recv(requester);		printf("Received reply %d [%s]/n", request_nbr, string);		free(string);	}	zmq_close(requester);	zmq_ctx_destroy(context);	return 0;}

int main (void){  void *context = zmq_ctx_new ();  // Socket to send messages on  void *sender = zmq_socket (context, ZMQ_PUSH);  zmq_bind (sender, "tcp://*:5557");  // Socket to send start of batch message on  void *sink = zmq_socket (context, ZMQ_PUSH);  zmq_connect (sink, "tcp://localhost:5558");  printf ("Press Enter when the workers are ready: ");  getchar ();  printf ("Sending tasks to workers…/n");  // The first message is "0" and signals start of batch  s_send (sink, "0");  // Initialize random number generator  srandom ((unsigned) time (NULL));  // Send 100 tasks  int task_nbr;  int total_msec = 0; // Total expected cost in msecs  for (task_nbr = 0; task_nbr < 100; task_nbr++) {    int workload;    // Random workload from 1 to 100msecs    workload = randof (100) + 1;    total_msec += workload;    char string [10];    sprintf (string, "%d", workload);    s_send (sender, string);  }  printf ("Total expected cost: %d msec/n", total_msec);  zmq_close (sink);  zmq_close (sender);  zmq_ctx_destroy (context);  return 0;}

int test_blocking (int send_hwm_, int msg_cnt_){    // Set up bind socket    void *pub_socket = test_context_socket (ZMQ_PUB);    TEST_ASSERT_SUCCESS_ERRNO (zmq_bind (pub_socket, "inproc://a"));    // Set up connect socket    void *sub_socket = test_context_socket (ZMQ_SUB);    TEST_ASSERT_SUCCESS_ERRNO (zmq_connect (sub_socket, "inproc://a"));    //set a hwm on publisher    TEST_ASSERT_SUCCESS_ERRNO (      zmq_setsockopt (pub_socket, ZMQ_SNDHWM, &send_hwm_, sizeof (send_hwm_)));    int wait = 1;    TEST_ASSERT_SUCCESS_ERRNO (      zmq_setsockopt (pub_socket, ZMQ_XPUB_NODROP, &wait, sizeof (wait)));    TEST_ASSERT_SUCCESS_ERRNO (      zmq_setsockopt (sub_socket, ZMQ_SUBSCRIBE, 0, 0));    // Send until we block    int send_count = 0;    int recv_count = 0;    while (send_count < msg_cnt_) {        const int rc = zmq_send (pub_socket, NULL, 0, ZMQ_DONTWAIT);        if (rc == 0) {            ++send_count;        } else if (-1 == rc) {            TEST_ASSERT_EQUAL_INT (EAGAIN, errno);            recv_count += receive (sub_socket);            TEST_ASSERT_EQUAL_INT (send_count, recv_count);        }    }    recv_count += receive (sub_socket);    // Clean up    test_context_socket_close (sub_socket);    test_context_socket_close (pub_socket);    return recv_count;}

void test_basic (){    size_t len = MAX_SOCKET_STRING;    char my_endpoint[MAX_SOCKET_STRING];    void *router = test_context_socket (ZMQ_ROUTER);    TEST_ASSERT_SUCCESS_ERRNO (zmq_bind (router, "tcp://127.0.0.1:*"));    TEST_ASSERT_SUCCESS_ERRNO (      zmq_getsockopt (router, ZMQ_LAST_ENDPOINT, my_endpoint, &len));    //  Send a message to an unknown peer with the default setting    //  This will not report any error    send_string_expect_success (router, "UNKNOWN", ZMQ_SNDMORE);    send_string_expect_success (router, "DATA", 0);    //  Send a message to an unknown peer with mandatory routing    //  This will fail    int mandatory = 1;    TEST_ASSERT_SUCCESS_ERRNO (zmq_setsockopt (router, ZMQ_ROUTER_MANDATORY,                                               &mandatory, sizeof (mandatory)));    int rc = zmq_send (router, "UNKNOWN", 7, ZMQ_SNDMORE);    TEST_ASSERT_EQUAL_INT (-1, rc);    TEST_ASSERT_EQUAL_INT (EHOSTUNREACH, errno);    //  Create dealer called "X" and connect it to our router    void *dealer = test_context_socket (ZMQ_DEALER);    TEST_ASSERT_SUCCESS_ERRNO (zmq_setsockopt (dealer, ZMQ_ROUTING_ID, "X", 1));    TEST_ASSERT_SUCCESS_ERRNO (zmq_connect (dealer, my_endpoint));    //  Get message from dealer to know when connection is ready    send_string_expect_success (dealer, "Hello", 0);    recv_string_expect_success (router, "X", 0);    //  Send a message to connected dealer now    //  It should work    send_string_expect_success (router, "X", ZMQ_SNDMORE);    send_string_expect_success (router, "Hello", 0);    test_context_socket_close (router);    test_context_socket_close (dealer);}

int main (void){    srandom ((unsigned) time (NULL));    void *context = zmq_init (1);    void *worker = zmq_socket (context, ZMQ_REQ);    //  Set random identity to make tracing easier    char identity [10];    sprintf (identity, "%04X-%04X", randof (0x10000), randof (0x10000));    zmq_setsockopt (worker, ZMQ_IDENTITY, identity, strlen (identity));    zmq_connect (worker, "tcp://localhost:5556");    //  Tell queue we're ready for work    printf ("I: (%s) worker ready/n", identity);    s_send (worker, "READY");    int cycles = 0;    while (1) {        zmsg_t *zmsg = zmsg_recv (worker);        //  Simulate various problems, after a few cycles        cycles++;        if (cycles > 3 && randof (5) == 0) {            printf ("I: (%s) simulating a crash/n", identity);            zmsg_destroy (&zmsg);            break;        }        else        if (cycles > 3 && randof (5) == 0) {            printf ("I: (%s) simulating CPU overload/n", identity);            sleep (5);        }        printf ("I: (%s) normal reply - %s/n", identity, zmsg_body (zmsg));        sleep (1);              //  Do some heavy work        zmsg_send (&zmsg, worker);    }    zmq_close (worker);    zmq_term (context);    return 0;}

int main (){    void *ctx = zmq_init (1);    assert (ctx);        void *pub = zmq_socket (ctx, ZMQ_PUB);    assert (pub);    int rc = zmq_bind (pub, "tcp://*:2211");    assert (rc == 0);        void *sub = zmq_socket (ctx, ZMQ_SUB);    assert (sub);    uint64_t size = 1;    rc = zmq_setsockopt (sub, ZMQ_HWM, &size, sizeof (size));    assert (rc == 0);        int64_t swap = 10000000;    rc = zmq_setsockopt (sub, ZMQ_SWAP, &swap, sizeof (swap));    assert (rc == 0);        rc = zmq_setsockopt (sub, ZMQ_SUBSCRIBE, "", 0);    assert (rc == 0);    rc = zmq_connect (sub, "tcp://127.0.0.1:2211");    assert (rc == 0);    int count;    for (count = 0; count < 100000; count++) {        zmq_msg_t msg;        zmq_msg_init_size (&msg, 0);        zmq_send (pub, &msg, 0);        zmq_msg_close (&msg);    }    int linger = 0;    zmq_setsockopt (pub, ZMQ_LINGER, &linger, sizeof (int));    zmq_setsockopt (sub, ZMQ_LINGER, &linger, sizeof (int));    zmq_close (pub);    zmq_close (sub);    zmq_term (ctx);    return 0;}

void test_round_robin_out (void *ctx){    void *req = zmq_socket (ctx, ZMQ_REQ);    assert (req);    int rc = zmq_bind (req, bind_address);    assert (rc == 0);    const size_t services = 5;    void *rep [services];    for (size_t peer = 0; peer < services; peer++) {        rep [peer] = zmq_socket (ctx, ZMQ_REP);        assert (rep [peer]);        int timeout = 250;        rc = zmq_setsockopt (rep [peer], ZMQ_RCVTIMEO, &timeout, sizeof (int));        assert (rc == 0);        rc = zmq_connect (rep [peer], connect_address);        assert (rc == 0);    }    //  We have to give the connects time to finish otherwise the requests     //  will not properly round-robin. We could alternatively connect the    //  REQ sockets to the REP sockets.    msleep (SETTLE_TIME);        // Send our peer-replies, and expect every REP it used once in order    for (size_t peer = 0; peer < services; peer++) {        s_send_seq (req, "ABC", SEQ_END);        s_recv_seq (rep [peer], "ABC", SEQ_END);        s_send_seq (rep [peer], "DEF", SEQ_END);        s_recv_seq (req, "DEF", SEQ_END);    }    close_zero_linger (req);    for (size_t peer = 0; peer < services; peer++)        close_zero_linger (rep [peer]);    // Wait for disconnects.    msleep (SETTLE_TIME);}

void test_pair_tcp (extra_func_t extra_func_ = NULL){    void *sb = test_context_socket (ZMQ_PAIR);    if (extra_func_)        extra_func_ (sb);    char my_endpoint[MAX_SOCKET_STRING];    bind_loopback_ipv4 (sb, my_endpoint, sizeof my_endpoint);    void *sc = test_context_socket (ZMQ_PAIR);    if (extra_func_)        extra_func_ (sc);    TEST_ASSERT_SUCCESS_ERRNO (zmq_connect (sc, my_endpoint));    bounce (sb, sc);    test_context_socket_close (sc);    test_context_socket_close (sb);}

int main (void){void *context = zmq_init (1);// Socket to talk to serverprintf ("Connecting to test_rep server... /n");void *requester = zmq_socket (context, ZMQ_SUB);zmq_connect (requester, "tcp://127.0.0.1:5555");zmq_setsockopt(requester, ZMQ_SUBSCRIBE, "", 0);while(1) {    zmq_msg_t reply;    zmq_msg_init (&reply);    zmq_recv (requester, &reply, 0);    printf ("Received %s /n", zmq_msg_data(&reply) );    zmq_msg_close (&reply);}zmq_close (requester);zmq_term (context);return 0;}

static void *step2 (void *context){    // Bind inproc socket before starting step1    void *receiver = zmq_socket (context, ZMQ_PAIR);    zmq_bind (receiver, "inproc://step2");    pthread_t thread;    pthread_create (&thread, NULL, step1, context);    // Wait for signal and pass it on    char *string = s_recv (receiver);    free (string);    zmq_close (receiver);    // Connect to step3 and tel it we're ready    void *xmitter = zmq_socket (context, ZMQ_PAIR);    zmq_connect (xmitter, "inproc://step3");    s_send (xmitter, "READY");    zmq_close (xmitter);    return NULL;}

int main(){    void *context = zmq_ctx_new();    void *subscriber = zmq_socket(context, ZMQ_SUB);    zmq_connect(subscriber, "tcp://localhost:5563");    zmq_setsockopt(subscriber, ZMQ_SUBSCRIBE, "B", 1);    while (1) {        // Read envelope with address        char *address = s_recv(subscriber);        // Read message contents        char *contents = s_recv(subscriber);        printf("[%s] %s/n", address, contents);        free(address);        free(contents);    }    zmq_close(subscriber);    zmq_ctx_destroy(context);    return 0;}

int main(){	printf("Connecting to hello world server.../n");	void *context = zmq_ctx_new();	void *requester = zmq_socket(context, ZMQ_REQ);	zmq_connect(requester, "tcp://localhost:5555");	int request_nbr;	for (request_nbr = 0; request_nbr != 10; ++request_nbr)	{		char buffer[10];		printf("Sending Hello %d.../n", request_nbr);		zmq_send(requester, "Hello", 5, 0);		zmq_recv(requester, buffer, 10, 0);		printf("Received World %d/n", request_nbr);	}	zmq_close(requester);	zmq_ctx_destroy(context);	return 0;}

int main(int argc, char *argv []) {    void *context, *socket;    char *topic, *string;    int64_t more;        if (argc < 2) {        fprintf(stderr, "USAGE: zmq_echo <address> [topic]/n");        return 1;    }    context = zmq_init(1);    printf("connecting to %s.../n", argv[1]);    socket = zmq_socket(context, ZMQ_SUB);    zmq_connect(socket, argv[1]);        if (argc > 2) {        topic = argv[2];        printf("subscribing to /"%s/" topic.../n", topic);    } else {        topic = "";        printf("subscribing to all topics.../n");    }    zmq_setsockopt(socket, ZMQ_SUBSCRIBE, topic, strlen(topic));        printf(">> ");    fflush(stdout);        while (1) {        string = zmq_recv_string(socket, &more);        printf(more ? "%s" : "%s/n>> ", string);        fflush(stdout);        free(string);    }    zmq_close(socket);    zmq_term(context);    return 0;}

// This checks for normal operation - that is pings and pongs being// exchanged normally. There should be an accepted event on the server,// and then no event afterwards.static voidtest_heartbeat_notimeout (int is_curve){    int rc;    //  Set up our context and sockets    void *ctx = zmq_ctx_new ();    assert (ctx);    void * server, * server_mon;    prep_server_socket(ctx, 1, is_curve, &server, &server_mon);    void * client = zmq_socket(ctx, ZMQ_DEALER);    if(is_curve)        setup_curve(client, 0);    rc = zmq_connect(client, "tcp://127.0.0.1:5556");    // Give it a sec to connect and handshake    msleep(100);    // By now everything should report as connected    rc = get_monitor_event(server_mon);    assert(rc == ZMQ_EVENT_ACCEPTED);    // We should still be connected because pings and pongs are happenin'    rc = get_monitor_event(server_mon);    assert(rc == -1);    rc = zmq_close (client);    assert (rc == 0);    rc = zmq_close (server);    assert (rc == 0);    rc = zmq_close (server_mon);    assert (rc == 0);    rc = zmq_ctx_term (ctx);    assert (rc == 0);}

static void prep_server_socket (int set_heartbeats_,                                int is_curve_,                                void **server_out_,                                void **mon_out_,                                char *endpoint_,                                size_t ep_length_,                                int socket_type_){    //  We'll be using this socket in raw mode    void *server = test_context_socket (socket_type_);    int value = 0;    TEST_ASSERT_SUCCESS_ERRNO (      zmq_setsockopt (server, ZMQ_LINGER, &value, sizeof (value)));    if (set_heartbeats_) {        value = 50;        TEST_ASSERT_SUCCESS_ERRNO (          zmq_setsockopt (server, ZMQ_HEARTBEAT_IVL, &value, sizeof (value)));    }    if (is_curve_)        setup_curve (server, 1);    bind_loopback_ipv4 (server, endpoint_, ep_length_);    //  Create and connect a socket for collecting monitor events on dealer    void *server_mon = test_context_socket (ZMQ_PAIR);    TEST_ASSERT_SUCCESS_ERRNO (zmq_socket_monitor (      server, "inproc://monitor-dealer",      ZMQ_EVENT_CONNECTED | ZMQ_EVENT_DISCONNECTED | ZMQ_EVENT_ACCEPTED));    //  Connect to the inproc endpoint so we'll get events    TEST_ASSERT_SUCCESS_ERRNO (      zmq_connect (server_mon, "inproc://monitor-dealer"));    *server_out_ = server;    *mon_out_ = server_mon;}

void test_bind_before_connect (){    void *ctx = zmq_ctx_new ();    assert (ctx);    // Bind first    void *bindSocket = zmq_socket (ctx, ZMQ_PAIR);    assert (bindSocket);    int rc = zmq_bind (bindSocket, "inproc://bbc");    assert (rc == 0);    // Now connect    void *connectSocket = zmq_socket (ctx, ZMQ_PAIR);    assert (connectSocket);    rc = zmq_connect (connectSocket, "inproc://bbc");    assert (rc == 0);    // Queue up some data    rc = zmq_send_const (connectSocket, "foobar", 6, 0);    assert (rc == 6);    // Read pending message    zmq_msg_t msg;    rc = zmq_msg_init (&msg);    assert (rc == 0);    rc = zmq_msg_recv (&msg, bindSocket, 0);    assert (rc == 6);    void *data = zmq_msg_data (&msg);    assert (memcmp ("foobar", data, 6) == 0);    // Cleanup    rc = zmq_close (connectSocket);    assert (rc == 0);    rc = zmq_close (bindSocket);    assert (rc == 0);    rc = zmq_ctx_term (ctx);    assert (rc == 0);}

/* send 100 messages over PAIR sockets * sender in one event handler, receiver in another */void test_ev_zmq (void){    struct ev_loop *loop;    void *zctx;    void *zin, *zout;    int i;    ev_zmq win, wout;    ok ((loop = ev_loop_new (EVFLAG_AUTO)) != NULL,        "ev_loop_new works");    ok ((zctx = zmq_init (1)) != NULL,        "initialized zmq context");    ok ((zout = zmq_socket (zctx, ZMQ_PAIR)) != NULL        && zmq_bind (zout, "inproc://eventloop_test") == 0,        "PAIR socket bind ok");    ok ((zin = zmq_socket (zctx, ZMQ_PAIR)) != NULL        && zmq_connect (zin, "inproc://eventloop_test") == 0,        "PAIR socket connect ok");    i = 0;    ev_zmq_init (&win, zsock_rx_cb, zin, EV_READ);    win.data = &i;    ev_zmq_init (&wout, zsock_tx_cb, zout, EV_WRITE);    ev_zmq_start (loop, &win);    ev_zmq_start (loop, &wout);    ok (ev_run (loop, 0) == 0,        "both watchers removed themselves and ev_run exited");    ev_zmq_stop (loop, &win);    ev_zmq_stop (loop, &wout);    cmp_ok (i, "==", 100,        "ev_zmq handler ran 100 times");    ev_loop_destroy (loop);    zmq_close (zin);    zmq_close (zout);    zmq_ctx_destroy (zctx);}

void test_leak (void){    char my_endpoint[256];    void *ctx = zmq_ctx_new ();    assert (ctx);    void *sb = zmq_socket (ctx, ZMQ_REP);    assert (sb);    int rc = zmq_bind (sb, "ipc://*");    assert (rc == 0);    size_t len = sizeof(my_endpoint);    rc = zmq_getsockopt (sb, ZMQ_LAST_ENDPOINT, my_endpoint, &len);    assert (rc == 0);    void *sc = zmq_socket (ctx, ZMQ_REQ);    assert (sc);    rc = zmq_connect (sc, my_endpoint);    assert (rc == 0);    rc = s_send (sc, "leakymsg");    assert (rc == strlen ("leakymsg"));    char *buf = s_recv (sb);    free (buf);    rc = zmq_close (sc);    assert (rc == 0);    msleep (SETTLE_TIME);    rc = s_send (sb, "leakymsg");    assert (rc == strlen ("leakymsg"));    rc = zmq_close (sb);    assert (rc == 0);    rc = zmq_ctx_term (ctx);    assert (rc == 0);}

int main(int argc, char *argv[]) {    q_open = r_open = 0;    if (argc != 2) {        printf("usage: %s <message>/n", argv[0]);        exit(-1);    }    ctx = zmq_init(1);    if (ctx < 0) {        printf("could not create context/n");        exit(-1);    }    s = zmq_socket(ctx, ZMQ_REQ);    check((long)s,"could not initialize socket");    printf("connecting.../n");    check(zmq_connect(s, "tcp://127.0.0.1:5554"),        "could not connect to server");    check(zmq_msg_init_data(&query, argv[1], strlen(argv[1]), NULL, NULL),        "could not initialize message");    q_open = 1;    printf("sending.../n");    check(zmq_send(s, &query, 0), "could not send message/n");    printf("waiting for result.../n");    check(zmq_msg_init(&result), "could not initialize message");    r_open = 1;    check(zmq_recv(s, &result, 0),        "could not receive reply");    printf("got response from server: %s/n", (char *)zmq_msg_data(&result));    cleanup();}

int main (void) {    setup_test_environment ();    void *ctx = zmq_ctx_new ();    assert (ctx);        void *stream = zmq_socket (ctx, ZMQ_STREAM);    assert (stream);    void *dealer = zmq_socket (ctx, ZMQ_DEALER);    assert (dealer);        int rc = zmq_bind (stream, "tcp://127.0.0.1:5555");    assert (rc >= 0);    rc = zmq_connect (dealer, "tcp://127.0.0.1:5555");    assert (rc >= 0);    zmq_send (dealer, "", 0, 0);        zmq_msg_t ident, empty;    zmq_msg_init (&ident);    rc = zmq_msg_recv (&ident, stream, 0);    assert (rc >= 0);    rc = zmq_msg_init_data (&empty, (void *) "", 0, NULL, NULL);    assert (rc >= 0);        rc = zmq_msg_send (&ident, stream, ZMQ_SNDMORE);    assert (rc >= 0);    rc = zmq_msg_close (&ident);    assert (rc >= 0);        rc = zmq_msg_send (&empty, stream, 0);    assert (rc >= 0);        //  This close used to fail with Bad Address    rc = zmq_msg_close (&empty);    assert (rc >= 0);        close_zero_linger (dealer);    close_zero_linger (stream);    zmq_ctx_term (ctx);}

int main (){    void *ctx = zmq_init (1);    assert (ctx);    void *dealer = zmq_socket (ctx, ZMQ_DEALER);    assert (dealer);    int rc = zmq_connect (dealer, "tcp://localhost:2211");    assert (rc == 0);    while (1) {        //  Send signal to parent        zmq_msg_t msg;        zmq_msg_init (&msg);        zmq_sendmsg (dealer, &msg, 0);        zmq_msg_close (&msg);        //  Do some work        sleep (1);    }    return 0;}

static ngx_int_tngx_http_upstream_zmq_connect(ngx_http_request_t *r, ngx_connection_t *c, ngx_http_upstream_zmq_peer_data_t *dp, void *dc){    ngx_http_upstream_t         * u;    int                           zmq_rc;    char                          ep[1024];    u = r->upstream;    assert(dp->name->data != NULL);    assert(dp->name->len > 0);    assert(dp->zsock != NULL);    if (dp->state == state_zmq_disconnected)    {        ngx_memcpy(ep, dp->name->data, dp->name->len);        ep[dp->name->len] = '/0';        zmq_debug(r->connection->log, "starting connection to %s", ep);        zmq_rc = zmq_connect (dp->zsock, ep);        if (zmq_rc  != 0)        {           ngx_log_error(NGX_LOG_ERR, c->log, 0, "failed to connect: %d", (int) zmq_rc);           return NGX_ERROR;        }        zmq_debug(r->connection->log, "changing state to connecting");        dp->state = state_zmq_connecting;        c->log->action = "connecting to endpoint";    }    else    {        zmq_debug(r->connection->log, "WARNING: trying to connect when state is %d", dp->state);    }    return NGX_AGAIN;}

// REQ socket monitor threadstatic void req_socket_monitor (void *ctx){    zmq_event_t event;    std::string ep ;    int rc;    void *s = zmq_socket (ctx, ZMQ_PAIR);    assert (s);    rc = zmq_connect (s, "inproc://monitor.req");    assert (rc == 0);    while (!read_msg(s, event, ep)) {        assert (ep == addr);        switch (event.event) {            case ZMQ_EVENT_CONNECTED:                assert (event.value > 0);                req_socket_events |= ZMQ_EVENT_CONNECTED;                req2_socket_events |= ZMQ_EVENT_CONNECTED;                break;            case ZMQ_EVENT_CONNECT_DELAYED:                assert (event.value != 0);                req_socket_events |= ZMQ_EVENT_CONNECT_DELAYED;                break;            case ZMQ_EVENT_CLOSE_FAILED:                assert (event.value != 0);                req_socket_events |= ZMQ_EVENT_CLOSE_FAILED;                break;            case ZMQ_EVENT_CLOSED:                assert (event.value != 0);                req_socket_events |= ZMQ_EVENT_CLOSED;                break;            case ZMQ_EVENT_DISCONNECTED:                assert (event.value != 0);                req_socket_events |= ZMQ_EVENT_DISCONNECTED;                break;        }    }    zmq_close (s);}

void main(){    fprintf(stdout,"#####  Client-1 is running - C Broker Test  #####/n");    void *context = zmq_ctx_new ();    void *requester = zmq_socket (context, ZMQ_PUSH);    zmq_connect (requester, "tcp://127.0.0.1:5559");    char *str_message="i am client-1";    int request_number;    //sending 1000000 requests from client    for (request_number = 1; request_number <= 1000000; request_number++) {        zstr_send (requester, str_message);    }    zmq_close (requester);    zmq_ctx_destroy (context);}

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);    rc = zmq_bind (server, "tcp://*:9999");    assert (rc == 0);            //  Client socket that will try to connect to server    void *client = zmq_socket (ctx, ZMQ_DEALER);    assert (client);    rc = zmq_connect (client, "tcp://localhost:9999");    assert (rc == 0);        bounce (server, client);        rc = zmq_close (client);    assert (rc == 0);    rc = zmq_close (server);    assert (rc == 0);        //  Shutdown    rc = zmq_ctx_term (ctx);    assert (rc == 0);        //  Wait until ZAP handler terminates.    zmq_threadclose (zap_thread);    return 0;}

        bool SubscriberZMQ::_onInit(TUInt port)        {            m_context = zmq_ctx_new();            if (m_context == 0)            {                zmqlog("Error occurred during zmq_ctx_new()");                return false;            }                         m_socket = zmq_socket(m_context, ZMQ_SUB);            if (m_socket == 0)            {                zmqlog("Error occurred during zmq_socket()");                clear();                return false;            }                         TChar address[100] = {'/0'};            sprintf(address, "tcp://localhost:%u", port);            int result = zmq_connect(m_socket, address);            if (result != 0)            {                zmqlog("Error occurred during zmq_init()");                clear();                return false;            }                        const TString& subscribeId = getSubscribeId();            result = zmq_setsockopt (m_socket, ZMQ_SUBSCRIBE, subscribeId.c_str(), subscribeId.length() - 1);            if (result != 0)            {                zmqlog("Error occurred during zmq_setsockopt");                clear();                return false;            }                        return true;        }

static voidprep_server_socket(void * ctx, int set_heartbeats, int is_curve, void ** server_out, void ** mon_out){    int rc;    //  We'll be using this socket in raw mode    void *server = zmq_socket (ctx, ZMQ_ROUTER);    assert (server);    int value = 0;    rc = zmq_setsockopt (server, ZMQ_LINGER, &value, sizeof (value));    assert (rc == 0);    if(set_heartbeats) {        value = 50;        rc = zmq_setsockopt (server, ZMQ_HEARTBEAT_IVL, &value, sizeof(value));        assert (rc == 0);    }    if(is_curve)        setup_curve(server, 1);    rc = zmq_bind (server, "tcp://127.0.0.1:5556");    assert (rc == 0);    //  Create and connect a socket for collecting monitor events on dealer    void *server_mon = zmq_socket (ctx, ZMQ_PAIR);    assert (server_mon);    rc = zmq_socket_monitor (server, "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 (server_mon, "inproc://monitor-dealer");    assert (rc == 0);    *server_out = server;    *mon_out = server_mon;}

int register_table(void *ctx, const char *address){    void *skt = zmq_socket(ctx, ZMQ_REQ);    assert(skt);    assert(zmq_connect(skt, address) == 0);    char *str = strdup(init_str);    send_msg_data(skt, str, strlen(str)+1, free_fn, NULL);    int64_t more;    size_t msg_size, more_size = sizeof(int64_t);    void *data;    recieve_msg(skt, &msg_size, &more, &more_size, &data);    if (msg_size != sizeof(uint8_t)){	return -1;    }    memcpy(&table_number, data, sizeof(uint8_t));    free(data);    if (more) flushall_msg_parts(skt);    assert(zmq_close(skt) == 0);    return 0;}