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

size_tzsys_socket_limit (void){    int socket_limit;#if (ZMQ_VERSION >= ZMQ_MAKE_VERSION (4, 1, 0))    if (s_process_ctx)        socket_limit = zmq_ctx_get (s_process_ctx, ZMQ_SOCKET_LIMIT);    else {        void *ctx = zmq_init (1);        socket_limit = zmq_ctx_get (ctx, ZMQ_SOCKET_LIMIT);        zmq_term (ctx);    }    //  ZeroMQ used to report a nonsense value (2^31) which if used would    //  on zmq_ctx_set (ZMQ_MAX_SOCKETS) cause an out-of-memory error. So    //  if we're running on an older library, enforce a sane limit.    if (socket_limit > 65535)        socket_limit = 65535;#else    socket_limit = 1024;#endif    return (size_t) socket_limit;}

voidzctx_destroy (zctx_t **self_p){    assert (self_p);    if (*self_p) {        zctx_t *self = *self_p;        //  Destroy all sockets        if (self->sockets)            while (zlist_size (self->sockets))                zctx__socket_destroy (self, zlist_first (self->sockets));        zlist_destroy (&self->sockets);        zmutex_destroy (&self->mutex);        //  ZMQ context may not yet be instantiated        if (self->context && !self->shadow)            zmq_term (self->context);        free (self);        *self_p = NULL;    }}

bool OTSocket_ZMQ_4::NewContext(){    if (!m_bInitialized) return false;    m_HasContext = false;    if (!this->CloseSocket(true)) return false;    if (NULL != m_pzmq->context_zmq) zmq_term(m_pzmq->context_zmq);    if (NULL != m_pzmq->context_zmq)	delete m_pzmq->context_zmq;	m_pzmq->context_zmq = NULL;    try {        m_pzmq->context_zmq = new zmq::context_t(1,31); // Threads, Max Sockets. (31 is a sane default).    }    catch (std::exception& e) {        OTLog::vError("%s: Exception Caught: %s /n", __FUNCTION__, e.what());        OT_FAIL;    }    m_HasContext = true;    return true;}

int main(int argc, char *argv[]){    if (argc < 3) return EXIT_FAILURE;    int M = atoi(argv[1]);    int N = atoi(argv[2]);    printf("M: %d, N: %d/n", M, N);    void *ctx = zmq_init(1);    void *b = zmq_socket(ctx, ZMQ_PAIR);    zmq_connect(b, "tcp://localhost:4444");    zmq_msg_t msg;    int val[M];    long long start_time = sc_time();    int i;    for (i=0; i<N; i++) {        int *buf = (int *)malloc(M * sizeof(int));        memset(val, i, M * sizeof(int));        memcpy(buf, val, M * sizeof(int));        zmq_msg_init_data(&msg, buf, M * sizeof(int), _dealloc, NULL);        zmq_send(b, &msg, 0);        zmq_msg_close(&msg);        zmq_msg_init(&msg);        zmq_recv(b, &msg, 0);        memcpy(val, (int *)zmq_msg_data(&msg), zmq_msg_size(&msg));        zmq_msg_close(&msg);    }    long long end_time = sc_time();    printf("zmq_a: Time elapsed: %f sec/n", sc_time_diff(start_time, end_time));    zmq_close(b);    zmq_term(ctx);    return EXIT_SUCCESS;}

//  We will do this all in one thread to emphasize the sequence//  of events...int main (void) {    void *context = zmq_init (1);    void *client = zmq_socket (context, ZMQ_ROUTER);    zmq_bind (client, "ipc://routing.ipc");    void *worker = zmq_socket (context, ZMQ_REP);    zmq_setsockopt (worker, ZMQ_IDENTITY, "A", 1);    zmq_connect (worker, "ipc://routing.ipc");    //  Wait for the worker to connect so that when we send a message    //  with routing envelope, it will actually match the worker...    sleep (1);    //  Send papa address, address stack, empty part, and request    s_sendmore (client, "A");    s_sendmore (client, "address 3");    s_sendmore (client, "address 2");    s_sendmore (client, "address 1");    s_sendmore (client, "");    s_send     (client, "This is the workload");    //  Worker should get just the workload	printf("begine dump worker/n");    s_dump (worker);    //  We don't play with envelopes in the worker    s_send (worker, "This is the reply");    //  Now dump what we got off the ROUTER socket...	printf("begine dump client/n");    s_dump (client);    zmq_close (client);    zmq_close (worker);    zmq_term (context);    return 0;}

int main (void){  // wait_for_key("before zmq_init");  void *context = zmq_init (1);  // wait_for_key("before fork");  pid_t pid = fork();    if (pid) {    // wait_for_key("in parent");    int status;    int rpid = waitpid(pid, &status, 0);    printf("[%d] waited for pid %d, got %d status %d/n", getpid(), pid, rpid, status);    printf("[%d] Parent is exiting/n", getpid());  }  else {    zmq_term(context);    // wait_for_key("in child");    exit(0);  }    return 0;}

int main (void){    fprintf (stderr, "test_router_mandatory_tipc running.../n");    void *ctx = zmq_init (1);    assert (ctx);    // Creating the first socket.    void *sa = zmq_socket (ctx, ZMQ_ROUTER);    assert (sa);    int rc = zmq_bind (sa, "tipc://{15560,0,0}");    assert (rc == 0);    // Sending a message to an unknown peer with the default setting    rc = zmq_send (sa, "UNKNOWN", 7, ZMQ_SNDMORE);    assert (rc == 7);    rc = zmq_send (sa, "DATA", 4, 0);    assert (rc == 4);    int mandatory = 1;    // Set mandatory routing on socket    rc = zmq_setsockopt (sa, ZMQ_ROUTER_MANDATORY, &mandatory, sizeof (mandatory));    assert (rc == 0);    // Send a message and check that it fails    rc = zmq_send (sa, "UNKNOWN", 7, ZMQ_SNDMORE | ZMQ_DONTWAIT);    assert (rc == -1 && errno == EHOSTUNREACH);    rc = zmq_close (sa);    assert (rc == 0);    rc = zmq_term (ctx);    assert (rc == 0);    return 0 ;}

int _tmain(int argc, _TCHAR* argv[]){	setlocale(LC_ALL,"Chinese");	setlocale(LC_ALL,"chs");	void *m_context;	void *m_subscriber;	char m_subAddr[64];	m_context = zmq_init(1);	m_subscriber = zmq_socket(m_context,ZMQ_SUB);		char *puberIp = "127.0.0.1";	WORD port = 8585;	memset(m_subAddr,0,sizeof(m_subAddr));	sprintf_s(m_subAddr,"tcp://%s:%d",puberIp,port);	zmq_connect(m_subscriber,m_subAddr);	char *option = "642";	int ret = zmq_setsockopt(m_subscriber,ZMQ_SUBSCRIBE,"642",strlen(option));	while (1)	{		BYTE buffer[1024] = {0};		DWORD bufLen = sizeof(buffer);		DWORD gotLen = zmq_recv(m_subscriber,buffer,bufLen,0);		printf("收到发布信息：%s/n",buffer);	}	zmq_close(m_subscriber);	zmq_term(m_context);	return 0;}

int main (int argc, char *argv []){    fprintf (stderr, "test_router_behavior running.../n");    void *ctx = zmq_init (1);    assert (ctx);    // Creating the first socket.    void *sa = zmq_socket (ctx, ZMQ_ROUTER);    assert (sa);        int rc = zmq_bind (sa, "tcp://127.0.0.1:15560");    assert (rc == 0);    // Sending a message to an unknown peer with the default behavior.    rc = zmq_send (sa, "UNKNOWN", 7, ZMQ_SNDMORE);    assert (rc == 7);    rc = zmq_send (sa, "DATA", 4, 0);    assert (rc == 4);    int behavior = 1;    // Setting the socket behavior to a new mode.    rc = zmq_setsockopt (sa, ZMQ_ROUTER_BEHAVIOR, &behavior, sizeof (behavior));    assert (rc == 0);    // Sending a message to an unknown peer with verbose behavior.    rc = zmq_send (sa, "UNKNOWN", 7, ZMQ_SNDMORE | ZMQ_DONTWAIT);    assert (rc == -1 && errno == EAGAIN);    rc = zmq_close (sa);    assert (rc == 0);    rc = zmq_term (ctx);    assert (rc == 0);    return 0 ;}

int main (void){    void *context = zmq_init (1);    //  This is where the weather server sits    void *frontend = zmq_socket (context, ZMQ_XSUB);    zmq_connect (frontend, "tcp://192.168.55.210:5556");    //  This is our public endpoint for subscribers    void *backend = zmq_socket (context, ZMQ_XPUB);    zmq_bind (backend, "tcp://10.1.1.0:8100");    //  Subscribe on everything    zmq_setsockopt (frontend, ZMQ_SUBSCRIBE, "", 0);    //  Shunt messages out to our own subscribers    while (1) {        while (1) {            zmq_msg_t message;            int64_t more;            //  Process all parts of the message            zmq_msg_init (&message);            zmq_recv (frontend, &message, 0);            size_t more_size = sizeof (more);            zmq_getsockopt (frontend, ZMQ_RCVMORE, &more, &more_size);            zmq_send (backend, &message, more? ZMQ_SNDMORE: 0);            zmq_msg_close (&message);            if (!more)                break;      //  Last message part        }    }    //  We don't actually get here but if we did, we'd shut down neatly    zmq_close (frontend);    zmq_close (backend);    zmq_term (context);    return 0;}

int main () {    void *context = zmq_init (1);    //  Subscriber tells us when it's ready here    void *sync = zmq_socket (context, ZMQ_PULL);    zmq_bind (sync, "tcp://*:5564");    //  We send updates via this socket    void *publisher = zmq_socket (context, ZMQ_PUB);    zmq_bind (publisher, "tcp://*:5565");    //  Prevent publisher overflow from slow subscribers    uint64_t hwm = 1;    zmq_setsockopt (publisher, ZMQ_HWM, &hwm, sizeof (hwm));    //  Specify swap space in bytes, this covers all subscribers    uint64_t swap = 25000000;    zmq_setsockopt (publisher, ZMQ_SWAP, &swap, sizeof (swap));    //  Wait for synchronization request    char *string = s_recv (sync);    free (string);    //  Now broadcast exactly 10 updates with pause    int update_nbr;    for (update_nbr = 0; update_nbr < 10; update_nbr++) {        char string [20];        sprintf (string, "Update %d", update_nbr);        s_send (publisher, string);        sleep (1);    }    s_send (publisher, "END");    sleep (1);              //  Give 0MQ/2.0.x time to flush output    zmq_term (context);    return 0;}

int main (void){    void *context = zmq_init (1);    void *client = zmq_socket (context, ZMQ_ROUTER);    zmq_bind (client, "ipc://routing.ipc");    pthread_t worker;    pthread_create (&worker, NULL, worker_task_a, NULL);    pthread_create (&worker, NULL, worker_task_b, NULL);    //  Wait for threads to connect, since otherwise the messages    //  we send won't be routable.    sleep (1);    //  Send 10 tasks scattered to A twice as often as B    int task_nbr;    srandom ((unsigned) time (NULL));    for (task_nbr = 0; task_nbr < 10; task_nbr++) {        //  Send two message parts, first the address...        if (randof (3) > 0)            s_sendmore (client, "A");        else            s_sendmore (client, "B");        //  And then the workload        s_send (client, "This is the workload");    }    s_sendmore (client, "A");    s_send     (client, "END");    s_sendmore (client, "B");    s_send     (client, "END");    zmq_close (client);    zmq_term (context);    return 0;}

// -- terminatednevn the ZMQ context for the current thread -------------voidciao_zmq_term_(void) {  // .. Execute once per process ...................................  pthread_once(&zmq_ll_once, make_state_key);  // .. Access the state pointer ...................................  ciao_zmq_state *state =    (ciao_zmq_state *)pthread_getspecific(zmq_ll_state_key);  // .. Finish if there is currently no state (nothing to temrinate) ..  if(state!=NULL) {    // .. Void the state for the current thread ....................    pthread_setspecific(zmq_ll_state_key, NULL);    // .. Deallocate all error records .............................    while(state->error_list != NULL) {      ciao_zmq_error_record *rec= state->error_list;      state->error_list= state->error_list->next;      free(rec);    }    // .. Close all sockets and deallocate association tables ......    while(state->socket_list !=NULL) {      ciao_zmq_socket_assoc *assoc= state->socket_list;      state->socket_list= state->socket_list->next;      zmq_close(assoc->zmq_socket);      free(assoc);    }    // .. Finally, terminate the context ...........................    zmq_term(state->zmq_context);    // .. Deallocate the state record ..............................    free(state);  }}

int main(int argc, char **argv){    init_options();    parse_options(argc, argv);    if (GlobalArgs.helpflag || !GlobalArgs.index_name || !GlobalArgs.server_address){	tableserver_usage();	return 0;    }    /* init daemon */     init_process();     if (init_index() < 0){	syslog(LOG_CRIT,"MAIN ERR: unable to init index");	exit(1);    }        void *ctx = zmq_init(1);    if (!ctx){	syslog(LOG_CRIT,"MAIN ERR: unable to init zmq ctx");	exit(1);    }    /* save to global variable to be used in signal handler */    main_ctx = ctx;    if (init_server(ctx) < 0){	syslog(LOG_CRIT,"MAIN ERR: unable to init server");	exit(1);    }    subscriber(ctx);        zmq_term(ctx);    return 0;}

//  We have two workers, here we copy the code, normally these would//  run on different boxes...//static void *worker_task_a (void *args){    void *context = zmq_init (1);    void *worker = zmq_socket (context, ZMQ_DEALER);    zmq_setsockopt (worker, ZMQ_IDENTITY, "A", 1);    zmq_connect (worker, "ipc://routing.ipc");    int total = 0;    while (1) {        //  We receive one part, with the workload        char *request = s_recv (worker);        int finished = (strcmp (request, "END") == 0);        free (request);        if (finished) {            printf ("A received: %d/n", total);            break;        }        total++;    }    zmq_close (worker);    zmq_term (context);    return NULL;}

int main (int argc, char *argv[]){    void *context = zmq_init (1);    //  Socket to send messages on    void *sender = zmq_socket (context, ZMQ_PUSH);    zmq_bind (sender, "tcp://*:5557");    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 (sender, "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 = within (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);    sleep (1);              //  Give 0MQ time to deliver    zmq_term (context);    return 0;}

int WINAPI wine_zmq_term (void* context) {  return zmq_term(context);}

intmain(int argc, char **argv){  int port_no = DEFAULT_PORT, daemon = 0;  const char *max_threads_string = NULL, *port_string = NULL;  const char *address;  const char *send_endpoint = NULL, *recv_endpoint = NULL, *log_base_path = NULL;  int n_processed_args, flags = RUN_MODE_ENABLE_MAX_FD_CHECK;  run_mode mode = run_mode_none;  if (!(default_max_threads = get_core_number())) {    default_max_threads = DEFAULT_MAX_THREADS;  }  /* parse options */  {    static grn_str_getopt_opt opts[] = {      {'c', NULL, NULL, 0, getopt_op_none}, /* deprecated */      {'t', "n-threads", NULL, 0, getopt_op_none},      {'h', "help", NULL, run_mode_usage, getopt_op_update},      {'p', "port", NULL, 0, getopt_op_none},      {'/0', "bind-address", NULL, 0, getopt_op_none}, /* not supported yet */      {'s', "send-endpoint", NULL, 0, getopt_op_none},      {'r', "receive-endpoint", NULL, 0, getopt_op_none},      {'l', "log-base-path", NULL, 0, getopt_op_none},      {'d', "daemon", NULL, run_mode_daemon, getopt_op_update},      {'/0', "disable-max-fd-check", NULL, RUN_MODE_ENABLE_MAX_FD_CHECK,       getopt_op_off},      {'/0', NULL, NULL, 0, 0}    };    opts[0].arg = &max_threads_string;    opts[1].arg = &max_threads_string;    opts[3].arg = &port_string;    opts[4].arg = &address;    opts[5].arg = &send_endpoint;    opts[6].arg = &recv_endpoint;    opts[7].arg = &log_base_path;    n_processed_args = grn_str_getopt(argc, argv, opts, &flags);  }  /* main */  mode = (flags & RUN_MODE_MASK);  if (n_processed_args < 0 ||      (argc - n_processed_args) != 1 ||      mode == run_mode_error) {    usage(stderr);    return EXIT_FAILURE;  } else if (mode == run_mode_usage) {    usage(stdout);    return EXIT_SUCCESS;  } else {    grn_ctx ctx;    void *zmq_ctx;    int max_threads;    if (max_threads_string) {      max_threads = atoi(max_threads_string);      if (max_threads > MAX_THREADS) {        print_error("too many threads. limit to %d.", MAX_THREADS);        max_threads = MAX_THREADS;      }    } else {      max_threads = default_max_threads;    }    if (port_string) {      port_no = atoi(port_string);    }    if (flags & RUN_MODE_ENABLE_MAX_FD_CHECK) {      /* check environment */      struct rlimit rlim;      if (!getrlimit(RLIMIT_NOFILE, &rlim)) {        if (rlim.rlim_max < MIN_MAX_FDS) {          print_error("too small max fds. %d required.", MIN_MAX_FDS);          return -1;        }        rlim.rlim_cur = rlim.rlim_cur;        setrlimit(RLIMIT_NOFILE, &rlim);      }    }    if (mode == run_mode_daemon) {      daemonize();    }    grn_init();    grn_ctx_init(&ctx, 0);    if ((db = grn_db_open(&ctx, argv[n_processed_args]))) {      if ((zmq_ctx = zmq_init(1))) {        signal(SIGTERM, signal_handler);        signal(SIGINT, signal_handler);        signal(SIGQUIT, signal_handler);        serve_threads(max_threads, port_no, argv[n_processed_args], zmq_ctx,                      send_endpoint, recv_endpoint, log_base_path);        zmq_term(zmq_ctx);      } else {        print_error("cannot create zmq context.");//.........这里部分代码省略.........

int main (int argc, char *argv[]){    //  Prepare our context and sockets    void *context = zmq_init (1);    void *frontend = zmq_socket (context, ZMQ_XREP);    void *backend  = zmq_socket (context, ZMQ_XREP);    zmq_bind (frontend, "ipc://frontend.ipc");    zmq_bind (backend,  "ipc://backend.ipc");    int client_nbr;    for (client_nbr = 0; client_nbr < NBR_CLIENTS; client_nbr++) {        pthread_t client;        pthread_create (&client, NULL, client_thread, context);    }    int worker_nbr;    for (worker_nbr = 0; worker_nbr < NBR_WORKERS; worker_nbr++) {        pthread_t worker;        pthread_create (&worker, NULL, worker_thread, context);    }    //  Logic of LRU loop    //  - Poll backend always, frontend only if 1+ worker ready    //  - If worker replies, queue worker as ready and forward reply    //    to client if necessary    //  - If client requests, pop next worker and send request to it    //  Queue of available workers    int available_workers = 0;    char *worker_queue [NBR_WORKERS];    while (1) {        //  Initialize poll set        zmq_pollitem_t items [] = {            //  Always poll for worker activity on backend            { backend,  0, ZMQ_POLLIN, 0 },            //  Poll front-end only if we have available workers            { frontend, 0, ZMQ_POLLIN, 0 }        };        if (available_workers)            zmq_poll (items, 2, -1);        else            zmq_poll (items, 1, -1);        //  Handle worker activity on backend        if (items [0].revents & ZMQ_POLLIN) {            zmsg_t *zmsg = zmsg_recv (backend);            //  Use worker address for LRU routing            assert (available_workers < NBR_WORKERS);            worker_queue [available_workers++] = zmsg_unwrap (zmsg);            //  Forward message to client if it's not a READY            if (strcmp (zmsg_address (zmsg), "READY") == 0)                zmsg_destroy (&zmsg);            else {                zmsg_send (&zmsg, frontend);                if (--client_nbr == 0)                    break;      //  Exit after N messages            }        }        if (items [1].revents & ZMQ_POLLIN) {            //  Now get next client request, route to next worker            zmsg_t *zmsg = zmsg_recv (frontend);            zmsg_wrap (zmsg, worker_queue [0], "");            zmsg_send (&zmsg, backend);            //  Dequeue and drop the next worker address            free (worker_queue [0]);            DEQUEUE (worker_queue);            available_workers--;        }    }    sleep (1);    zmq_term (context);    return 0;}

void cleanup() {    if (q_open) zmq_msg_close(&query);    if (r_open) zmq_msg_close(&result);    zmq_close(s);    zmq_term(ctx);}

int main (void){    s_version_assert (2, 1);    srandom ((unsigned) time (NULL));    void *context = zmq_init (1);    void *worker = s_worker_socket (context);    //  If liveness hits zero, queue is considered disconnected    size_t liveness = HEARTBEAT_LIVENESS;    size_t interval = INTERVAL_INIT;    //  Send out heartbeats at regular intervals    uint64_t heartbeat_at = s_clock () + HEARTBEAT_INTERVAL;    int cycles = 0;    while (1) {        zmq_pollitem_t items [] = { { worker,  0, ZMQ_POLLIN, 0 } };        zmq_poll (items, 1, HEARTBEAT_INTERVAL * 1000);        if (items [0].revents & ZMQ_POLLIN) {            //  Get message            //  - 3-part envelope + content -> request            //  - 1-part "HEARTBEAT" -> heartbeat            zmsg_t *msg = zmsg_recv (worker);            if (msg_parts (msg) == 3) {                //  Simulate various problems, after a few cycles                cycles++;                if (cycles > 3 && randof (5) == 0) {                    printf ("I: (%s) simulating a crash/n", identity);                    zmsg_destroy (&msg);                    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 (msg));                zmsg_send (&msg, worker);                liveness = HEARTBEAT_LIVENESS;                sleep (1);              //  Do some heavy work            }            else            if (msg_parts (msg) == 1            && strcmp (msg_body (msg), "HEARTBEAT") == 0)                liveness = HEARTBEAT_LIVENESS;            else {                printf ("E: (%s) invalid message/n", identity);                zmsg_dump (msg);            }            interval = INTERVAL_INIT;        }        else        if (--liveness == 0) {            printf ("W: (%s) heartbeat failure, can't reach queue/n",                identity);            printf ("W: (%s) reconnecting in %zd msec.../n",                identity, interval);            s_sleep (interval);            if (interval < INTERVAL_MAX)                interval *= 2;            zmq_close (worker);            worker = s_worker_socket (context);            liveness = HEARTBEAT_LIVENESS;        }        //  Send heartbeat to queue if it's time        if (s_clock () > heartbeat_at) {            heartbeat_at = s_clock () + HEARTBEAT_INTERVAL;            printf ("I: (%s) worker heartbeat/n", identity);            s_send (worker, "HEARTBEAT");        }    }    zmq_close (worker);    zmq_term (context);    return 0;}

ZMQ_EXPORT int WINAPI mql4zmq_term (void *context){	return zmq_term(context);}

int main(int argc, char **argv){    int ret, opt, exit_code = EXIT_LOCAL_FAILURE;    void *ctx = NULL, *socket = NULL;    const char *endpoint = default_ep;    struct pcma_req req;    zmq_pollitem_t pollitem;    zmq_msg_t msg;    if (argc < 1)        help(NULL);    if (argc < 2)        help(argv[0]);    if (argc < 2)        help(argv[0]);    while ((opt = getopt(argc, argv, "ve:t:")) != -1) {        switch (opt) {        case 'v':            log_level++;            break;        case 'e':            endpoint = optarg;            break;        case 't':            timeout = atol(optarg);            if (timeout >= LONG_MAX / 1000L) {                LOG_ERROR("timeout too high/n");                goto err;            }            break;        default:            help(argv[0]);        }    }    LOG_INFO("using endpoint %s/n", endpoint);    if (timeout >= 0) {        LOG_INFO("using a %li ms timeout/n", timeout);    }    if (!(ctx = zmq_init(1)))        MAIN_ERR_FAIL("zmq_init");    if (!(socket = zmq_socket(ctx, ZMQ_REQ)))        MAIN_ERR_FAIL("zmq_socket");    if (zmq_connect(socket, endpoint) < 0)        MAIN_ERR_FAIL("zmq_connect");    if (optind >= argc) {        LOG_ERROR("command expected/n");        goto err;    }    req.argc = argc - optind;    req.argv = argv + optind;    ret = pcma_send(socket, pcma_req_packfn, &req);    if (ret < 0) {        LOG_ERROR("pcma_send failed with %i/n", ret);        goto err;    }    if (timeout >= 0) {        pollitem.socket = socket;        pollitem.events = ZMQ_POLLIN;        ret = zmq_poll(&pollitem, 1, timeout * 1000L);        if (ret < 0)            MAIN_ERR_FAIL("zmq_poll");        if (ret == 0) {            int zero = 0;            LOG_ERROR("timeout after %li ms/n", timeout);            zmq_setsockopt(socket, ZMQ_LINGER, &zero, sizeof(zero));            goto err;        }    }    if (zmq_msg_init(&msg) < 0)        MAIN_ERR_FAIL("zmq_msg_init");    if (zmq_recv(socket, &msg, 0) < 0)        MAIN_ERR_FAIL("zmq_recv");    ret = handle_rep(&msg);    if (ret > 0)        exit_code = EXIT_REMOTE_FAILURE;    if (ret != 0) {        LOG_ERROR("handle_rep failed with %i/n", ret);        goto err;    }    if (zmq_msg_close(&msg) < 0) {        perror("zmq_msg_close");        goto err;    }    if (zmq_close(socket) < 0)        MAIN_ERR_FAIL("zmq_close");    if (zmq_term(ctx) < 0)//.........这里部分代码省略.........

EXPORT int WINAPI _zmq_term (void* context) {  return zmq_term(context);  }

int main (int argc, char *argv []){    const char *connect_to;    int message_count;    int message_size;    void *ctx;    void *s;    int rc;    int i;    zmq_msg_t msg;    if (argc != 4) {        printf ("usage: remote_thr <connect-to> <message-size> "            "<message-count>/n");        return 1;    }    connect_to = argv [1];    message_size = atoi (argv [2]);    message_count = atoi (argv [3]);    ctx = zmq_init (1);    if (!ctx) {        printf ("error in zmq_init: %s/n", zmq_strerror (errno));        return -1;    }    s = zmq_socket (ctx, ZMQ_PUB);    if (!s) {        printf ("error in zmq_socket: %s/n", zmq_strerror (errno));        return -1;    }    //  Add your socket options here.    //  For example ZMQ_RATE, ZMQ_RECOVERY_IVL and ZMQ_MCAST_LOOP for PGM.    rc = zmq_connect (s, connect_to);    if (rc != 0) {        printf ("error in zmq_connect: %s/n", zmq_strerror (errno));        return -1;    }    for (i = 0; i != message_count; i++) {        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;        }#if defined ZMQ_MAKE_VALGRIND_HAPPY        memset (zmq_msg_data (&msg), 0, message_size);#endif        rc = zmq_sendmsg (s, &msg, 0);        if (rc < 0) {            printf ("error in zmq_sendmsg: %s/n", zmq_strerror (errno));            return -1;        }        rc = zmq_msg_close (&msg);        if (rc != 0) {            printf ("error in zmq_msg_close: %s/n", zmq_strerror (errno));            return -1;        }    }    rc = zmq_close (s);    if (rc != 0) {        printf ("error in zmq_close: %s/n", zmq_strerror (errno));        return -1;    }    rc = zmq_term (ctx);    if (rc != 0) {        printf ("error in zmq_term: %s/n", zmq_strerror (errno));        return -1;    }    return 0;}

int main (int argc, char *argv []){    const char *bind_to;    int message_count;    size_t message_size;    void *ctx;    void *s;    int rc;    int i;    zmq_msg_t msg;    void *watch;    unsigned long elapsed;    unsigned long throughput;    double megabits;    if (argc != 4) {        printf ("usage: local_thr <bind-to> <message-size> <message-count>/n");        return 1;    }    bind_to = argv [1];    message_size = atoi (argv [2]);    message_count = atoi (argv [3]);    ctx = zmq_init (1);    if (!ctx) {        printf ("error in zmq_init: %s/n", zmq_strerror (errno));        return -1;    }    s = zmq_socket (ctx, ZMQ_PULL);    if (!s) {        printf ("error in zmq_socket: %s/n", zmq_strerror (errno));        return -1;    }    //  Add your socket options here.    //  For example ZMQ_RATE, ZMQ_RECOVERY_IVL and ZMQ_MCAST_LOOP for PGM.    rc = zmq_bind (s, bind_to);    if (rc != 0) {        printf ("error in zmq_bind: %s/n", zmq_strerror (errno));        return -1;    }    rc = zmq_msg_init (&msg);    if (rc != 0) {        printf ("error in zmq_msg_init: %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;    }    watch = zmq_stopwatch_start ();    for (i = 0; i != message_count - 1; i++) {        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);    if (elapsed == 0)        elapsed = 1;    rc = zmq_msg_close (&msg);    if (rc != 0) {        printf ("error in zmq_msg_close: %s/n", zmq_strerror (errno));        return -1;    }    throughput = (unsigned long)        ((double) message_count / (double) elapsed * 1000000);    megabits = (double) (throughput * message_size * 8) / 1000000;    printf ("message size: %d [B]/n", (int) message_size);    printf ("message count: %d/n", (int) message_count);    printf ("mean throughput: %d [msg/s]/n", (int) throughput);    printf ("mean throughput: %.3f [Mb/s]/n", (double) megabits);    rc = zmq_close (s);    if (rc != 0) {        printf ("error in zmq_close: %s/n", zmq_strerror (errno));        return -1;    }    rc = zmq_term (ctx);//.........这里部分代码省略.........

int main (int argc, char *argv []){    const char *bind_to;    int roundtrip_count;    size_t message_size;    void *ctx;    void *s;    int rc;    int i;    zmq_msg_t msg;    if (argc != 4) {        printf ("usage: local_lat <bind-to> <message-size> "            "<roundtrip-count>/n");        return 1;    }    bind_to = argv [1];    message_size = atoi (argv [2]);    roundtrip_count = atoi (argv [3]);    ctx = zmq_init (1);    if (!ctx) {        printf ("error in zmq_init: %s/n", zmq_strerror (errno));        return -1;    }    s = zmq_socket (ctx, ZMQ_REP);    if (!s) {        printf ("error in zmq_socket: %s/n", zmq_strerror (errno));        return -1;    }    rc = zmq_bind (s, bind_to);    if (rc != 0) {        printf ("error in zmq_bind: %s/n", zmq_strerror (errno));        return -1;    }    rc = zmq_msg_init (&msg);    if (rc != 0) {        printf ("error in zmq_msg_init: %s/n", zmq_strerror (errno));        return -1;    }    for (i = 0; i != roundtrip_count; i++) {        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;        }        rc = zmq_sendmsg (s, &msg, 0);        if (rc < 0) {            printf ("error in zmq_sendmsg: %s/n", zmq_strerror (errno));            return -1;        }    }    rc = zmq_msg_close (&msg);    if (rc != 0) {        printf ("error in zmq_msg_close: %s/n", zmq_strerror (errno));        return -1;    }    zmq_sleep (1);    rc = zmq_close (s);    if (rc != 0) {        printf ("error in zmq_close: %s/n", zmq_strerror (errno));        return -1;    }    rc = zmq_term (ctx);    if (rc != 0) {        printf ("error in zmq_term: %s/n", zmq_strerror (errno));        return -1;    }    return 0;}

int main (){    fprintf (stderr, "pubsub_udp test running.../n");    void *ctx = zmq_init (1);    assert (ctx);    void *pub = zmq_socket (ctx, ZMQ_PUB);    assert (pub);    int rc = zmq_bind (pub, "udp://127.0.0.1:5555");    assert (rc != -1);    void *sub = zmq_socket (ctx, ZMQ_SUB);    assert (sub);    rc = zmq_connect (sub, "udp://127.0.0.1:5555");    assert (rc != -1);    rc = zmq_setsockopt (sub, ZMQ_SUBSCRIBE, "", 0);    assert (rc == 0);    //  Just in case there's an delay in lower parts of the network stack.    sleep (1);        const char *content = "12345678ABCDEFGH12345678abcdefgh";    //  Send a message with two identical parts.    rc = zmq_send (pub, content, 32, ZMQ_SNDMORE);    assert (rc == 32);    rc = zmq_send (pub, content, 32, 0);    assert (rc == 32);        //  Receive the first part.    char rcvbuf [32];    int rcvmore = 0;    size_t rcvmore_sz = sizeof rcvmore;    rc = zmq_recv (sub, rcvbuf, 32, 0);    assert (rc == 32);    rc = zmq_getsockopt (sub, ZMQ_RCVMORE, &rcvmore, &rcvmore_sz);    assert (rc == 0);    //  There must be one more part to receive.    assert (rcvmore);    //  And the content must match what was sent.    assert (memcmp (rcvbuf, content, 32) == 0);    //  Receive the second part.    rc = zmq_recv (sub, rcvbuf, 32, 0);    assert (rc == 32);    rcvmore_sz = sizeof rcvmore;    rc = zmq_getsockopt (sub, ZMQ_RCVMORE, &rcvmore, &rcvmore_sz);    assert (rc == 0);    //  There must not be another part.    assert (!rcvmore);    //  And the content must match what was sent.    assert (memcmp (rcvbuf, content, 32) == 0);    rc = zmq_close (pub);    assert (rc == 0);    rc = zmq_close (sub);    assert (rc == 0);    rc = zmq_term (ctx);    assert (rc == 0);    return 0 ;}

int main (void){    fprintf (stderr, "test_router_mandatory2 running.../n");    void *ctx = zmq_init (1);    assert (ctx);    // Creating the first socket.    void *sa = zmq_socket (ctx, ZMQ_ROUTER);    assert (sa);        int rc = zmq_bind (sa, "tcp://127.0.0.1:15560");    assert (rc == 0);    // Sending a message to an unknown peer with the default setting    rc = zmq_send (sa, "SOCKET", 6, ZMQ_SNDMORE);    assert (rc == 6);    rc = zmq_send (sa, "DATA", 4, 0);    assert (rc == 4);    int mandatory = 1;    // Set mandatory routing on socket    rc = zmq_setsockopt (sa, ZMQ_ROUTER_MANDATORY, &mandatory, sizeof (mandatory));    assert (rc == 0);    // Send a message and check that it fails    rc = zmq_send (sa, "SOCKET", 6, ZMQ_SNDMORE | ZMQ_DONTWAIT);    assert (rc == -1 && errno == EHOSTUNREACH);		//create a receiver socket		void *receiver = zmq_socket (ctx, ZMQ_ROUTER);    assert (receiver);		//give it a name		rc = zmq_setsockopt (receiver, ZMQ_IDENTITY, "SOCKET", 6);  	assert (rc == 0);		//connect to sender		rc = zmq_connect (receiver, "tcp://127.0.0.1:15560");  	assert (rc == 0);		//sleep to connect		zmq_sleep(1);			 // Send a message and check that it NOT fails    rc = zmq_send (sa, "SOCKET", 6, ZMQ_SNDMORE);    assert (rc == 6);		rc = zmq_send (sa, "DATA", 4, 0);    assert (rc == 4);//sleep to receive		zmq_sleep(1);		//receive it on socket		//TODO:		char buffer[16];		memset (&buffer, 0, sizeof(buffer));		rc =  zmq_recv (receiver, &buffer, sizeof(buffer), ZMQ_DONTWAIT);		assert(rc > 0);		rc =  zmq_recv (receiver, &buffer, sizeof(buffer), ZMQ_DONTWAIT);		printf ("rc is %d, in buffer: %s /n",rc, buffer);		assert(rc == 4);		//close receiver socket		rc = zmq_close(receiver);		assert (rc == 0);		//sleep to disconnect		zmq_sleep(5);		//send again to check if fails  	// Send a message and check that it fails    rc = zmq_send (sa, "SOCKET", 6, ZMQ_SNDMORE | ZMQ_DONTWAIT);		printf ("rc is %d/n", rc);    assert (rc == -1 && errno == EHOSTUNREACH);		//closing stuff    rc = zmq_close (sa);    assert (rc == 0);	    rc = zmq_term (ctx);    assert (rc == 0);    return 0 ;}

 zcontext::~zcontext() {     int rc = zmq_term(_Mp_ctx);     assert(rc == 0); }