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

static int ntop_zmq_connect(lua_State* vm) {  char *endpoint, *topic;  void *context, *subscriber;  if(ntop_lua_check(vm, __FUNCTION__, 1, LUA_TSTRING)) return(CONST_LUA_PARAM_ERROR);  if((endpoint = (char*)lua_tostring(vm, 1)) == NULL)  return(CONST_LUA_PARAM_ERROR);  if(ntop_lua_check(vm, __FUNCTION__, 2, LUA_TSTRING)) return(CONST_LUA_PARAM_ERROR);  if((topic = (char*)lua_tostring(vm, 2)) == NULL)     return(CONST_LUA_PARAM_ERROR);  context = zmq_ctx_new(), subscriber = zmq_socket(context, ZMQ_SUB);  if(zmq_connect(subscriber, endpoint) != 0) {    zmq_close(subscriber);    zmq_ctx_destroy(context);    return(CONST_LUA_PARAM_ERROR);  }  if(zmq_setsockopt(subscriber, ZMQ_SUBSCRIBE, topic, strlen(topic)) != 0) {    zmq_close(subscriber);    zmq_ctx_destroy(context);    return -1;  }  lua_pushlightuserdata(vm, context);  lua_setglobal(vm, "zmq_context");  lua_pushlightuserdata(vm, subscriber);  lua_setglobal(vm, "zmq_subscriber");  return(CONST_LUA_OK);}

int main (void){    setup_test_environment();        void *ctx1 = zmq_ctx_new ();    assert (ctx1);    void *ctx2 = zmq_ctx_new ();    assert (ctx2);    void *router = zmq_socket (ctx1, ZMQ_ROUTER);        int on = 1;    int rc = zmq_setsockopt (router, ZMQ_ROUTER_MANDATORY, &on, sizeof (on));    assert (rc == 0);    rc = zmq_bind (router, "tcp://127.0.0.1:5555");    assert (rc != -1);     //  Repeat often enough to be sure this works as it should    for (int cycle = 0; cycle < 100; cycle++) {        //  Create dealer with unique explicit identity        //  We assume the router learns this out-of-band        void *dealer = zmq_socket (ctx2, ZMQ_DEALER);        char identity [10];        sprintf (identity, "%09d", cycle);        rc = zmq_setsockopt (dealer, ZMQ_IDENTITY, identity, 10);        assert (rc == 0);        int rcvtimeo = 1000;        rc = zmq_setsockopt (dealer, ZMQ_RCVTIMEO, &rcvtimeo, sizeof (int));        assert (rc == 0);        rc = zmq_connect (dealer, "tcp://127.0.0.1:5555");        assert (rc == 0);        //  Router will try to send to dealer, at short intervals.        //  It typically takes 2-5 msec for the connection to establish        //  on a loopback interface, but we'll allow up to one second        //  before failing the test (e.g. for running on a debugger or        //  a very slow system).        for (int attempt = 0; attempt < 500; attempt++) {            zmq_poll (0, 0, 2);            rc = zmq_send (router, identity, 10, ZMQ_SNDMORE);            if (rc == -1 && errno == EHOSTUNREACH)                continue;            assert (rc == 10);            rc = zmq_send (router, "HELLO", 5, 0);            assert (rc == 5);            break;        }        uint8_t buffer [5];        rc = zmq_recv (dealer, buffer, 5, 0);        assert (rc == 5);        assert (memcmp (buffer, "HELLO", 5) == 0);        close_zero_linger (dealer);    }    zmq_close (router);    zmq_ctx_destroy (ctx1);    zmq_ctx_destroy (ctx2);    return 0;}

int main(int argc, char * argv[]){	void * pCtx = NULL;	void * pSock = NULL;	const char * pAddr = "tcp://*:7766";	//创建context，zmq的socket 需要在context上进行创建 	if ((pCtx = zmq_ctx_new()) == NULL)	{		return 0;	}	//创建zmq socket ，socket目前有6中属性 ，这里使用dealer方式	//具体使用方式请参考zmq官方文档（zmq手册） 	if ((pSock = zmq_socket(pCtx, ZMQ_DEALER)) == NULL)	{		zmq_ctx_destroy(pCtx);		return 0;	}	int iRcvTimeout = 5000;// millsecond	//设置zmq的接收超时时间为5秒 	if (zmq_setsockopt(pSock, ZMQ_RCVTIMEO, &iRcvTimeout, sizeof(iRcvTimeout)) < 0)	{		zmq_close(pSock);		zmq_ctx_destroy(pCtx);		return 0;	}	//绑定地址 tcp://*:7766 	//也就是使用tcp协议进行通信，使用网络端口 7766	if (zmq_bind(pSock, pAddr) < 0)	{		zmq_close(pSock);		zmq_ctx_destroy(pCtx);		return 0;	}	printf("bind at : %s/n", pAddr);	while (1)	{		char szMsg[1024] = { 0 };		printf("waitting.../n");		errno = 0;		//循环等待接收到来的消息，当超过5秒没有接到消息时，		//zmq_recv函数返回错误信息 ，并使用zmq_strerror函数进行错误定位 		if (zmq_recv(pSock, szMsg, sizeof(szMsg), 0) < 0)		{			printf("error = %s/n", zmq_strerror(errno));			continue;		}		printf("received message : %s/n", szMsg);	}	return 0;}

int main(int argc, char * argv[]){	void * pCtx = NULL;	void * pSock = NULL;	//使用tcp协议进行通信，需要连接的目标机器IP地址为192.168.1.2	//通信使用的网络端口 为7766 	const char * pAddr = "tcp://127.0.0.1:7766";//"tcp://192.168.1.198:7766";	//创建context 	if ((pCtx = zmq_ctx_new()) == NULL)	{		return 0;	}	//创建socket 	if ((pSock = zmq_socket(pCtx, ZMQ_DEALER)) == NULL)	{		zmq_ctx_destroy(pCtx);		return 0;	}	int iSndTimeout = 5000;// millsecond	//设置接收超时 	if (zmq_setsockopt(pSock, ZMQ_RCVTIMEO, &iSndTimeout, sizeof(iSndTimeout)) < 0)	{		zmq_close(pSock);		zmq_ctx_destroy(pCtx);		return 0;	}	//连接目标IP192.168.1.2，端口7766 	if (zmq_connect(pSock, pAddr) < 0)	{		zmq_close(pSock);		zmq_ctx_destroy(pCtx);		return 0;	}	//循环发送消息 	while (1)	{		static int i = 0;		char szMsg[1024] = { 0 };		sprintf_s(szMsg, "hello world : %3d", i++);		printf("Enter to send.../n");		if (zmq_send(pSock, szMsg, sizeof(szMsg), 0) < 0)		{			fprintf(stderr, "send message faild/n");			continue;		}		printf("send message : [%s] succeed/n", szMsg);		getchar();	}	return 0;}

bool Service::subscribe( const std::string& brokerAddress,                         const Publisher& publisher ){    void* context = zmq_ctx_new();    void* socket = zmq_socket( context, ZMQ_REQ );    const std::string zmqAddress = std::string("tcp://" ) + brokerAddress;    if( zmq_connect( socket, zmqAddress.c_str( )) == -1 )    {        ZEQINFO << "Can't reach connection broker at " << brokerAddress                << std::endl;        zmq_close( socket );        zmq_ctx_destroy( context );        return false;    }    const std::string& address = publisher.getAddress();    zmq_msg_t request;    zmq_msg_init_size( &request, address.size( ));    memcpy( zmq_msg_data( &request ), address.c_str(), address.size( ));    if( zmq_msg_send( &request, socket, 0 ) == -1 )    {        zmq_msg_close( &request );        ZEQINFO << "Can't send connection request " << address << " to "                << brokerAddress << ": " << zmq_strerror( zmq_errno( ))                << std::endl;        return false;    }    zmq_msg_close( &request );    zmq_msg_t reply;    zmq_msg_init( &reply );    if( zmq_msg_recv( &reply, socket, 0 )  == -1 )    {        zmq_msg_close( &reply );        ZEQINFO << "Can't receive connection reply from " << brokerAddress                << std::endl;        return false;    }    const std::string result( (const char*)zmq_msg_data( &reply ),                              zmq_msg_size( &reply ));    zmq_msg_close( &reply );    zmq_close( socket );    zmq_ctx_destroy( context );    return address == std::string( result );}

intmain (int argc, char **argv){    int i;    char *endpoint = NULL;    void *context, *socket;    if (argc <= 2) {        _usage(argv[0]);        return -1;    }    endpoint = argv[1];    context = zmq_ctx_new();    if (!context) {        _ERR("ZeroMQ context: %s/n", zmq_strerror(errno));        return -1;    }    socket = zmq_socket(context, ZMQ_PUSH);    if (!socket) {        _ERR("ZeroMQ socket: %s/n", zmq_strerror(errno));        zmq_ctx_destroy(context);        return -1;    }    if (zmq_connect(socket, endpoint) == -1) {        _ERR("ZeroMQ connect: %s: %s/n", endpoint, zmq_strerror(errno));        zmq_close(socket);        zmq_ctx_destroy(context);        return -1;    }    for (i = 2; i != (argc - 1); i++) {        if (zmq_send(socket, argv[i], strlen(argv[i]), ZMQ_SNDMORE) == -1) {            _ERR("ZeroMQ send: %s/n", zmq_strerror(errno));        }    }    if (zmq_send(socket, argv[i], strlen(argv[i]), 0) == -1) {        _ERR("ZeroMQ send: %s/n", zmq_strerror(errno));    }    zmq_close(socket);    zmq_ctx_destroy(context);    return 0;}

int main (void){  printf ("Connecting to math server…/n");  void *context = zmq_ctx_new ();  void *requester = zmq_socket (context, ZMQ_REQ);  zmq_connect (requester, "tcp://localhost:5555");  srand(time(NULL));  int request_nbr;  for (request_nbr = 0; request_nbr < 1000; request_nbr++) {    char buffer [10];    int a = randof(30);    int b = randof(30);    printf ("Sending (%d, %d)…/n", a, b);    char to_send[100];    sprintf(to_send, "%d %d", a, b);    s_send(requester, to_send);    char* ret = s_recv(requester);    int sum = atoi(ret);    printf("%d + %d = %d/n", a, b, sum);    free(ret);  }  zmq_close (requester);  zmq_ctx_destroy (context);  return 0;}

ExportInterface::~ExportInterface() {  if(topic)    free(topic);  if(endpoint) free(endpoint);  zmq_close(publisher);  zmq_ctx_destroy(context);}

//-----------------------------------------------------------------------------// TODO: determine level of detail for return value, i.e. void, bool or enum or// if the errors should cause this function to automatically try again.connection_c::error_e connection_c::close( void ) {  // sanity checks  assert( _open );  // close the socket  if( zmq_close( _socket ) == -1 ) {    if( errno == ENOTSOCK ) {      // the provided socket was invalid      return ERROR_SOCKET;    }  }  // if a client or router, destroy the context  if( _role == CLIENT || _role == ROUTER ) {    if( zmq_ctx_destroy( _context ) == -1 ) {      if( errno == EFAULT ) {        // The provided context was invalid        return ERROR_CONTEXT;      } else if( errno == EINTR ) {        // Termination was interrupted by a signal, can be restarted.        return ERROR_INTERRUPT;      } else {        // Note: above should trap specifics, but if fall through then return        // a socket error          return ERROR_SOCKET;      }        }  }  _open = false;  return ERROR_NONE;}

int main (int argc, char *argv []){  // Socket to talk to server  printf ("Collecting updates from weather server…/n");  void *context = zmq_ctx_new ();  void *subscriber = zmq_socket (context, ZMQ_SUB);  int rc = zmq_connect (subscriber, "tcp://localhost:5556");  assert (rc == 0);  // Subscribe to zipcode, default is NYC, 10001  char *filter = (argc > 1)? argv [1]: "10001 ";  rc = zmq_setsockopt (subscriber, ZMQ_SUBSCRIBE,		       filter, strlen (filter));  assert (rc == 0);  // Process 100 updates  int update_nbr;  long total_temp = 0;  for (update_nbr = 0; update_nbr < 100; update_nbr++) {    char *string = s_recv (subscriber);    int zipcode, temperature, relhumidity;    sscanf (string, "%d %d %d",	    &zipcode, &temperature, &relhumidity);    total_temp += temperature;    free (string);  }  printf ("Average temperature for zipcode '%s' was %dF/n",	  filter, (int) (total_temp / update_nbr));  zmq_close (subscriber);  zmq_ctx_destroy (context);  return 0;}

int main(int argc, char* argv[]){  void* ctx = zmq_ctx_new();  void* rep = zmq_socket(ctx, ZMQ_REP);  zmq_connect(rep, "tcp://localhost:6666");  do {    int count = 0, server_id;    char buf[128];    if (argc < 2) {      fprintf(stderr, "arguments error .../n");      break;    }    server_id = atoi(argv[1]);    fprintf(stdout, "reply server[%d] init success .../n", server_id);    while (1) {      memset(buf, 0, sizeof(buf));      zmq_recv(rep, buf, sizeof(buf), 0);      fprintf(stdout, "recevied request {%s}/n", buf);      sprintf(buf, "[%d] reply [%d]/n", server_id, ++count);      zmq_send(rep, buf, strlen(buf), 0);    }  } while (0);  zmq_close(rep);  zmq_ctx_destroy(ctx);  return 0;}

int main(){    void *context = zmq_ctx_new();    void *responder = zmq_socket(context, ZMQ_REP);    zmq_bind(responder, "tcp://*:5555");    while(1) {        zmq_msg_t request;        zmq_msg_init(&request);        zmq_msg_recv(&request, responder, 0);        char *recvmsg ;        int recvsize = zmq_msg_size(&request);        //memset(recvmsg, 0, recvsize);        memcpy(recvmsg, (char *)zmq_msg_data(&request), recvsize);        recvmsg[recvsize] = '/0';        printf("Received request :[%s]/n", recvmsg);        zmq_msg_close(&request);        sleep(1);        zmq_msg_t reply;        //zmq_msg_init_size(&reply, 5);        //memcpy((void*)zmq_msg_data(&reply), "World", 5);        zmq_msg_init_data(&reply, "World", 5, NULL, NULL);        zmq_msg_send(&reply, responder, 0);        zmq_msg_close(&reply);    }    zmq_close(responder);    zmq_ctx_destroy(context);    return 0;}

int main (void){    void *context = zmq_ctx_new ();    void *worker = zmq_socket (context, ZMQ_DEALER);    s_set_id (worker);          //  Set a printable identity    zmq_connect (worker, "tcp://localhost:5671");    int total = 0;    while (1) {        //  Tell the broker we're ready for work        s_sendmore (worker, "");        s_send (worker, "Hi Boss");        //  Get workload from broker, until finished        // free (s_recv (worker));     //  Envelope delimiter        // char *workload = s_recv (worker);        // int finished = (strcmp (workload, "Fired!") == 0);        // free (workload);        // if (finished) {            // printf ("Completed: %d tasks/n", total);            // break;        // }        total++;        //  Do some random work        s_sleep (randof (500) + 1);    }    zmq_close (worker);    zmq_ctx_destroy (context);    return 0;}

int main (int argc, char *argv []){    void *context = zmq_ctx_new ();    void *subscriber = zmq_socket (context, ZMQ_SUB);    void *buffer = (void *) malloc(MAXSZ);    lo_message msg;    int rc = zmq_connect (subscriber, "tcp://localhost:5556");    assert (rc == 0);    //  Subscribe to all    rc = zmq_setsockopt(subscriber, ZMQ_SUBSCRIBE,			NULL, 0);    assert (rc == 0);    osc_init();    while(1) {      int size = zmq_recv(subscriber, buffer, MAXSZ, 0);      msg = (lo_message) buffer;      if (size > 0) {	printf("received a message that's %d big/n", size);	lo_server_dispatch_data(s, buffer, size);      }      else {	printf("oops./n");      }    }    zmq_close (subscriber);    zmq_ctx_destroy (context);    return 0;}

int main(int argc, char** argv) {    if(argc < 3) {        std::cout << "usage: "                   << argv[0]                   << " <front-end URI> <back-end URI>"                  << std::endl;        std::cout << "Example: broker /"tcp://*:5555/" /"tcp://*:6666/"/n";                  return 0;              }    void* ctx = zmq_ctx_new();     void* frontend = zmq_socket(ctx, ZMQ_XPUB);    void* backend = zmq_socket(ctx, ZMQ_XSUB);    const char* brokerURI = argv[1];    const char* backendURI = argv[2];    int rc = zmq_bind(frontend, brokerURI);    assert(rc == 0);    rc = zmq_bind(backend, backendURI);    assert(rc == 0);    while(1) {        zmq_proxy(frontend, backend, 0);    }    rc = zmq_close(frontend);    assert(rc == 0);    rc = zmq_close(backend);    assert(rc == 0);    rc = zmq_ctx_destroy(ctx);    assert(rc == 0);    return 0;}

int main (void){    void *context = zmq_ctx_new ();    //  Socket to talk to clients    void *clients = zmq_socket (context, ZMQ_ROUTER);    zmq_bind (clients, "tcp://*:5555");    //  Socket to talk to workers    void *workers = zmq_socket (context, ZMQ_DEALER);    zmq_bind (workers, "inproc://workers");    //  Launch pool of worker threads    int thread_nbr;    for (thread_nbr = 0; thread_nbr < 5; thread_nbr++) {        pthread_t worker;        pthread_create (&worker, NULL, worker_routine, context);    }    //  Connect work threads to client threads via a queue proxy    zmq_proxy (clients, workers, NULL);    //  We never get here but clean up anyhow    zmq_close (clients);    zmq_close (workers);    zmq_ctx_destroy (context);    return 0;}

int main(){	void *context = zmq_ctx_new();	// Socket to talk to clients	void *clients = zmq_socket(context, ZMQ_ROUTER);	zmq_bind(clients, "tcp://*:5555");	// Socket to talk to workers	void *workers = zmq_socket(context, ZMQ_DEALER);	zmq_bind(workers, "inproc://workers");	// Launch pool of worker threads	const int THREAD_NUM = 5;	int thread_nbr;	WorkerThread *threads[THREAD_NUM];	for (thread_nbr = 0; thread_nbr < THREAD_NUM; ++thread_nbr)	{		threads[thread_nbr] = new WorkerThread(thread_nbr, context);		threads[thread_nbr]->run();	}	zmq_proxy(clients, workers, NULL);	for (thread_nbr = 0; thread_nbr < THREAD_NUM; ++thread_nbr)	{		delete threads[thread_nbr];	}	zmq_close(clients);	zmq_close(workers);	zmq_ctx_destroy(context);	return 0;}

int main (void){    printf ("Connecting to hello world server.../n");    void *context = zmq_ctx_new ();    void *requester = zmq_socket (context, ZMQ_REQ);    zmq_connect (requester, "tcp://localhost:5556");    //zmq_connect (requester, "tcp://localhost:5556");    int t=5000;    zmq_setsockopt(requester,ZMQ_LINGER,&t,sizeof(int));    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);	printf("Sendover/n");        //zmq_recv (requester, buffer, 10, ZMQ_DONTWAIT);        zmq_recv (requester, buffer, 10, 0);        printf ("Received World %d/n", request_nbr);    }    zmq_close (requester);    printf("close/n");    zmq_ctx_destroy (context);    printf("destroy/n");    return 0;}

void s_catch_signals () {    struct sigaction action = {        .sa_handler = s_signalhandler,        .sa_flags = 0    };    sigemptyset(&action.sa_mask);    sigaction(SIGINT, &action, NULL);    sigaction(SIGTERM, &action, NULL);}int main () {    void* context = zmq_ctx_new();    void* socket = zmq_socket(context, ZMQ_REP);    zmq_bind(socket, "tcp://*:5555");    s_catch_signals();    while(1) {        // Blocking read will exit on a signal        zmq_msg_t message;        zmq_msg_init(&message);        zmq_msg_recv(&message, socket, 0);        if (s_interrupted) {            printf("W: interrupt recived, killing server.../n");            break;        }    }    zmq_close(socket);    zmq_ctx_destroy(context);}

int main(){	// Socket to receive messages on	void *context = zmq_ctx_new();	void *receiver = zmq_socket(context, ZMQ_PULL);	zmq_connect(receiver, "tcp://localhost:5557");	// Socket to send messages to	void *sender = zmq_socket(context, ZMQ_PUSH);	zmq_connect(sender, "tcp://localhost:5558");	// Process tasks forever	while (1)	{		char *string = s_recv(receiver);		printf("%s.", string);  // Show progress		fflush(stdout);		s_sleep(atoi(string));  // Do the work		free(string);		s_send(sender, "");     // Send results to sink	}	zmq_close(receiver);	zmq_close(sender);	zmq_ctx_destroy(context);	return 0;}

int main(int argc, char* argv[]){  int i;  void* ctx = zmq_ctx_new();  void* s = zmq_socket(ctx, ZMQ_REQ);  zmq_connect(s, "tcp://localhost:5555");  for (i = 0; i < 10; ++i) {    zmq_msg_t req, rep;    zmq_msg_init_size(&req, 128);    sprintf((char*)zmq_msg_data(&req),       "Hello, world! current count[%d]", i);    zmq_msg_send(&req, s, 0);    zmq_msg_close(&req);    zmq_msg_init(&rep);    zmq_msg_recv(&rep, s, 0);    fprintf(stdout, "reply message : %s/n", (char*)zmq_msg_data(&rep));    zmq_msg_close(&rep);    __sleep(100);  }  zmq_close(s);  zmq_ctx_destroy(ctx);  return 0;}

int main (void){    //  Prepare our context and publisher    void *context = zmq_ctx_new ();    void *publisher = zmq_socket (context, ZMQ_PUSH);    //int rc = zmq_bind (publisher, "tcp://192.168.0.254:5556");    int rc = zmq_connect (publisher, "tcp://192.168.0.254:5559");   assert (rc == 0);    //  Initialize random number generator    srandom ((unsigned) time (NULL));    while (1) {        //  Get values that will fool the boss        int zipcode, temperature, relhumidity;        zipcode     = randof (100000);        temperature = randof (215) - 80;        relhumidity = randof (50) + 10;        //  Send message to all subscribers        char update [20];        //sprintf (update, "%05d %d %d", zipcode, temperature, relhumidity);        sprintf (update, "hello %05d %d %d", zipcode, temperature, relhumidity);		printf(update);	    int size = zmq_send (publisher, update, strlen (update), 0);        //s_send (publisher, update);    }    zmq_close (publisher);    zmq_ctx_destroy (context);    return 0;}

void test_system_max (){    // Keep allocating sockets until we run out of system resources    const int no_of_sockets = 2 * 65536;    void *ctx = zmq_ctx_new ();    zmq_ctx_set (ctx, ZMQ_MAX_SOCKETS, no_of_sockets);    std::vector <void*> sockets;    while (true) {        void *socket = zmq_socket (ctx, ZMQ_PAIR);        if (!socket)            break;        sockets.push_back (socket);    }    assert ((int) sockets.size () < no_of_sockets);    //  System is out of resources, further calls to zmq_socket should return NULL    for (unsigned int i = 0; i < 10; ++i) {        void *socket = zmq_socket (ctx, ZMQ_PAIR);        assert (socket == NULL);    }    // Clean up.    for (unsigned int i = 0; i < sockets.size (); ++i)        zmq_close (sockets [i]);    zmq_ctx_destroy (ctx);}

int main (void) {    void *context = zmq_ctx_new ();    //  Socket to talk to clients    void *responder = zmq_socket (context, ZMQ_REP);    zmq_connect (responder, "tcp://localhost:5560");    while (true) {        //  Wait for next request from client        char *string = s_recv (responder);        printf ("Received request: [%s]/n", string);        free (string);        //  Do some 'work'        sleep (1);        //  Send reply back to client        s_send (responder, "World");    }    //  We never get here but clean up anyhow    zmq_close (responder);    zmq_ctx_destroy (context);    return 0;}

void test_zmq_default_max (){    //  Keep allocating sockets until we hit the default limit    void *ctx = zmq_ctx_new ();    std::vector<void*> sockets;    while (true) {        void *socket = zmq_socket (ctx, ZMQ_PAIR);        if (!socket)            break;        sockets.push_back (socket);    }    //  We may stop sooner if system has fewer available sockets    assert (sockets.size () <= ZMQ_MAX_SOCKETS_DFLT);    //  Further calls to zmq_socket should return NULL    for (unsigned int i = 0; i < 10; ++i) {        void *socket = zmq_socket (ctx, ZMQ_PAIR);        assert (socket == NULL);    }    //  Clean up    for (unsigned int i = 0; i < sockets.size (); ++i)        zmq_close (sockets [i]);    zmq_ctx_destroy (ctx);}

static void *worker_task (void *args){    void *context = zmq_ctx_new ();    void *worker = zmq_socket (context, ZMQ_REQ);    s_set_id (worker);          //  Set a printable identity    zmq_connect (worker, "ipc://backend.ipc");    //  Tell broker we're ready for work    s_send (worker, "READY");    while (1) {        //  Read and save all frames until we get an empty frame        //  In this example there is only 1 but it could be more        char *identity = s_recv (worker);        char *empty = s_recv (worker);        assert (*empty == 0);        free (empty);        //  Get request, send reply        char *request = s_recv (worker);        printf ("Worker: %s/n", request);        free (request);        s_sendmore (worker, identity);        s_sendmore (worker, "");        s_send     (worker, "OK");        free (identity);    }    zmq_close (worker);    zmq_ctx_destroy (context);    return NULL;}

void test_ctx_shutdown(){    int rc;        //  Set up our context and sockets    void *ctx = zmq_ctx_new ();    assert (ctx);        void *socket = zmq_socket (ctx, ZMQ_PULL);    assert (socket);    // Spawn a thread to receive on socket    void *receiver_thread = zmq_threadstart (&receiver, socket);    // Wait for thread to start up and block    msleep (SETTLE_TIME);    // Shutdown context, if we used destroy here we would deadlock.    rc = zmq_ctx_shutdown (ctx);    assert (rc == 0);    // Wait for thread to finish    zmq_threadclose (receiver_thread);    // Close the socket.    rc = zmq_close (socket);    assert (rc == 0);    // Destory the context, will now not hang as we have closed the socket.    rc = zmq_ctx_destroy (ctx);    assert (rc == 0);}

int main (int argc, char ** argv){	printf ("ZEROMQSEND TESTS/n");	printf ("================/n/n");	init (argc, argv);	int major, minor, patch;	zmq_version (&major, &minor, &patch);	printf ("zeromq version is %d.%d.%d/n", major, minor, patch);	context = zmq_ctx_new ();	// Test notification from plugin	test_commit ();	// test timeouts	test_timeoutConnect ();	test_timeoutSubscribe ();	print_result ("testmod_zeromqsend");	zmq_ctx_destroy (context);	return nbError;}

int main (void) {    void *context = zmq_ctx_new ();    void *sink = zmq_socket (context, ZMQ_ROUTER);    zmq_bind (sink, "inproc://example");    //  First allow 0MQ to set the identity    void *anonymous = zmq_socket (context, ZMQ_REQ);    zmq_connect (anonymous, "inproc://example");    s_send (anonymous, "ROUTER uses a generated UUID");    s_dump (sink);    //  Then set the identity ourselves    void *identified = zmq_socket (context, ZMQ_REQ);    zmq_setsockopt (identified, ZMQ_IDENTITY, "PEER2", 5);    zmq_connect (identified, "inproc://example");    s_send (identified, "ROUTER socket uses REQ's socket identity");    s_dump (sink);    zmq_close (sink);    zmq_close (anonymous);    zmq_close (identified);    zmq_ctx_destroy (context);    return 0;}