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

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

int main (void){    setup_test_environment();    void *ctx = zmq_ctx_new ();    assert (ctx);    void *sb = zmq_socket (ctx, ZMQ_REP);    assert (sb);    int rc = zmq_bind (sb, "inproc://a");    assert (rc == 0);    rc = zmq_unbind (sb, "inproc://a");    assert (rc == 0);    rc = zmq_close (sb);    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);    void *sock = zmq_socket (ctx, ZMQ_PUB);    assert (sock);    int rc = zmq_connect (sock, "tcp://localhost:1234");    assert (rc == 0);    rc = zmq_connect (sock, "tcp://[::1]:1234");    assert (rc == 0);    rc = zmq_connect (sock, "tcp://localhost:invalid");    assert (rc == -1);    rc = zmq_connect (sock, "tcp://in val id:1234");    assert (rc == -1);    rc = zmq_connect (sock, "tcp://");    assert (rc == -1);    rc = zmq_connect (sock, "tcp://192.168.0.200:*");    assert (rc == -1);    rc = zmq_connect (sock, "invalid://localhost:1234");    assert (rc == -1);    assert (errno == EPROTONOSUPPORT);    rc = zmq_close (sock);    assert (rc == 0);    rc = zmq_ctx_term (ctx);    assert (rc == 0);    return 0;}

int main(void){    void *context = zmq_ctx_new();        //Socket to connect to Server    void *requester = zmq_socket(context, ZMQ_REQ);    zmq_connect(requester, "tcp://localhost:8888");        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 create_sub_socket(void* ctx) {	std::stringstream conn;	conn << "tcp://" << g_sip << ":" << g_port_sub;	void * sub = zmq_socket(ctx, ZMQ_SUB);	int rc = zmq_connect(sub, conn.str().c_str());	assert(rc == 0);	if (rc != 0) {		std::cerr << "Error Connect to " << conn.str() << " errno " << errno << std::endl;		zmq_close(sub);		exit(EXIT_FAILURE);	}	char temp[256] = { 0 };	sprintf(temp, "%s ", g_sn);	printf("Subscribe id: %s/n", temp);	rc = zmq_setsockopt(sub, ZMQ_SUBSCRIBE, temp, strlen(temp));	assert(rc == 0);	rc = zmq_setsockopt(sub, ZMQ_SUBSCRIBE, "ALL ", strlen("ALL "));	assert(rc == 0);	g_sub_socket = sub;	return 0;}

//********************************************************************************// Function: redirect_count//// Description: Recieve Control Plane redirect count.//********************************************************************************static int32_t redirect_count(void){    void *context = zmq_ctx_new ();    void *requester = zmq_socket(context, ZMQ_REQ);    int32_t timeo{};    zmq_setsockopt(requester, ZMQ_LINGER, (void*) &timeo, sizeof(timeo));    char buffer[64];    sprintf(buffer, "tcp://localhost:%d", TCPLANE_SERVICE);    zmq_connect(requester, buffer);    sprintf(buffer, "%d", redirectCount);    zmq_send(requester, buffer, strlen(buffer), 0);    size_t size = timed_read(requester, buffer, sizeof(buffer), READ_TIMEOUT);    if(size)        size = atoi(buffer);        zmq_close(requester);    zmq_ctx_destroy(context);    return size;}

int main (int argc, char const *argv[]) {  void* context = zmq_ctx_new();  //create a SUB socket  void* subscriber = zmq_socket(context, ZMQ_SUB);  const char* filter;    if(argc > 1) {    filter = argv[1];  } else {    filter = "Company1|";  }  printf("Collecting stock information from the server./n");  int conn = zmq_connect(subscriber, "tcp://localhost:4040");  // must set a subscription for SUB socket  conn = zmq_setsockopt(subscriber, ZMQ_SUBSCRIBE, filter, strlen(filter));  int i = 0;  for(i = 0; i < 10; i++) {    zmq_msg_t reply;    zmq_msg_init(&reply);    // receive the message, previous message is deallocated    zmq_msg_recv(&reply, subscriber, 0);        int length = zmq_msg_size(&reply);    char* value = malloc(length + 1);    memcpy(value, zmq_msg_data(&reply), length);    zmq_msg_close(&reply);    printf("%s/n", value);        free(value);  }  zmq_close(subscriber);  zmq_ctx_destroy(context);  return 0;}

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

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

bool CZMQAbstractPublishNotifier::Initialize(void *pcontext){    assert(!psocket);    // check if address is being used by other publish notifier    std::multimap<std::string, CZMQAbstractPublishNotifier*>::iterator i = mapPublishNotifiers.find(address);    if (i==mapPublishNotifiers.end())    {        psocket = zmq_socket(pcontext, ZMQ_PUB);        if (!psocket)        {            zmqError("Failed to create socket");            return false;        }        int rc = zmq_bind(psocket, address.c_str());        if (rc!=0)        {            zmqError("Failed to bind address");            zmq_close(psocket);            return false;        }        // register this notifier for the address, so it can be reused for other publish notifier        mapPublishNotifiers.insert(std::make_pair(address, this));        return true;    }    else    {        LogPrint(BCLog::ZMQ, "zmq: Reusing socket for address %s/n", address);        psocket = i->second->psocket;        mapPublishNotifiers.insert(std::make_pair(address, this));        return true;    }}

void Executor::work(){    void *socket = zmq_socket(vcontext, ZMQ_REP);    zmq_msg_t zMsg;    char *cstring = NULL;    size_t msgSize = 0;        zmq_connect(socket, "inproc://storageWorkers");        while(vstop==false)    {        zmq_msg_init(&zMsg);        zmq_msg_recv(&zMsg, socket, 0);        msgSize = zmq_msg_size(&zMsg);        cstring = new char[msgSize+1];        memcpy(cstring, zmq_msg_data(&zMsg), msgSize);        zmq_msg_close(&zMsg);        cstring[msgSize] = '/0';        std::string str(cstring);        delete cstring;        Message* msg = vmsgFactory->parse(str);        Action *action = vactionFactory->parse(*msg);        delete msg;        msg = action->execute();        delete action;        msgSize = msg->toString().size()+1;        zmq_msg_init_size(&zMsg, msgSize);        memcpy(zmq_msg_data(&zMsg), msg->toString().c_str(), msgSize);        delete msg;        zmq_msg_send(&zMsg, socket, 0);        zmq_msg_close(&zMsg);    }    zmq_close(socket);}    

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

static void subscriber_thread_main (void *pvoid){    const proxy_hwm_cfg_t *cfg = (proxy_hwm_cfg_t *) pvoid;    const int idx = cfg->thread_idx;    void *subsocket = zmq_socket (cfg->context, ZMQ_SUB);    assert (subsocket);    set_hwm (subsocket);    TEST_ASSERT_SUCCESS_ERRNO (zmq_setsockopt (subsocket, ZMQ_SUBSCRIBE, 0, 0));    TEST_ASSERT_SUCCESS_ERRNO (      zmq_connect (subsocket, cfg->backend_endpoint[idx]));    //  Receive message_count messages    uint64_t rxsuccess = 0;    bool success = true;    while (success) {        zmq_msg_t msg;        int rc = zmq_msg_init (&msg);        assert (rc == 0);        rc = zmq_msg_recv (&msg, subsocket, 0);        if (rc != -1) {            TEST_ASSERT_SUCCESS_ERRNO (zmq_msg_close (&msg));            rxsuccess++;        }        if (rxsuccess == message_count)            break;    }    //  Cleanup    zmq_close (subsocket);    printf ("subscriber thread ended/n");}

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

// worker thread functionstatic void* worker_routine (void *context){	// socket to talk to dispatcher	void* receiver = zmq_socket(context, ZMQ_REP); 	if (!receiver) 	{		dzlog_error("create socket to talk to dispatcher failed, err: %s", zmq_strerror(errno));		return NULL;	}	if (zmq_connect(receiver, "inproc://workers") != 0)	{		dzlog_error("worker thread connect worker_socket failed, err: %s", zmq_strerror(errno));		return NULL;	}	dzlog_info("worker thread run ...");	while (1) 	{		zmq_msg_t request;		zmq_msg_init(&request);		zmq_recvmsg(receiver, &request, 0); 		printf("Recv Hello/n"); 		zmq_msg_close(&request);		sleep(1);				zmq_msg_t reply;		zmq_msg_init_size(&reply, 5);		memcpy(zmq_msg_data (&reply), "World", 5); 		zmq_sendmsg(receiver, &reply, 0); 		zmq_msg_close(&reply);	}	zmq_close (receiver);	return NULL;}

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 kill_server(){    char addr[32];    void *ctx = main_ctx;    snprintf(addr, 32, "tcp://%s:%d", GlobalArgs.server_address, GlobalArgs.port + 1);    void *skt = socket_connect(ctx, ZMQ_REQ, addr);    if (!skt){	syslog(LOG_CRIT,"KILLSERVER: unable to get req skt");	return -1;    }    uint8_t tn = table_number;    table_number = 0;    char *tmpstr = strdup(kill_str);    sendmore_msg_data(skt, tmpstr, strlen(tmpstr)+1, free_fn, NULL);    send_msg_vsm(skt, &tn, sizeof(uint8_t));    time_t curr_time;    time(&curr_time);    zmq_msg_t msg;    zmq_msg_init(&msg);    int err = 0;    do {	err = zmq_recv(skt, &msg, ZMQ_NOBLOCK);	sleep(1);    } while (err && time(NULL) < curr_time + 10 );    if (err){	syslog(LOG_ERR, "KILLSERVER: no ack recieved - ignore");	err = -1;    }    zmq_close(skt);    syslog(LOG_DEBUG,"KILLSERVER: server killed");    return err;}

int main (void){    //  Create the infrastructure    void *ctx = zmq_ctx_new ();    assert (ctx);    void *sb = zmq_socket (ctx, ZMQ_ROUTER);    assert (sb);    int val = 0;    int rc = zmq_setsockopt (sb, ZMQ_LINGER, &val, sizeof (val));    assert (rc == 0);    do_bind_and_verify (sb, "tcp://127.0.0.1:5560");    do_bind_and_verify (sb, "tcp://127.0.0.1:5561");    rc = zmq_close (sb);    assert (rc == 0);        rc = zmq_ctx_term (ctx);    assert (rc == 0);    return 0 ;}

static void *worker_routine (void *context) {    int thread_counter = 0;    //  Socket to talk to dispatcher    void *receiver = zmq_socket (context, ZMQ_REP);    zmq_connect (receiver, "inproc://workers");    char *reply = malloc(50);    while (1) {        char *string = s_recv (receiver);        printf ("Received request: [%s]/n", string);        free (string);        //  Do some 'work'        //sleep (1);        //  Send reply back to client        sprintf(reply, "%d World", thread_counter++);        s_send (receiver, reply);    }    zmq_close (receiver);    return NULL;}

int main (int argc, char *argv []){  // Socket to talk to server  printf ("Collecting updates from chat 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);  char *filter = (argc > 1)? argv [1]: ">";  rc = zmq_setsockopt (subscriber, ZMQ_SUBSCRIBE,		       filter, strlen (filter));  assert (rc == 0);  for (;;) {    char *string = s_recv (subscriber);    printf("Received: %s/n", string);    free(string);  }  zmq_close (subscriber);  zmq_ctx_destroy (context);  return 0;}

int main(void){    // Prepare our context and subscriber    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 messages contents        char *contents = s_recv(subscriber);        printf("[%s] %s/n", address, contents);        free(address);        free(contents);    }    // We never get here, but clean up anyhow    zmq_close(subscriber);    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);                                                       /*通过tcp协议，5555端口，连接本机服务端*/                                                              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;                                                                                          }

void *status_thread_start(void *zcontext){    WINDOW *status_window;    status_window = status_window_init();    void *receiver = zmq_socket(zcontext, ZMQ_PULL);    zmq_bind (receiver, "tcp://*:5558");    char *status;        while(1) {        status = s_recv(receiver);        mvwprintw(status_window, 0, 0, blank);        wrefresh(status_window);        mvwprintw(status_window, 0, 0, status);        wrefresh(status_window);        free(status);    }    zmq_close(receiver);    delwin(status_window);    return NULL;}

int main(int argc, char **argv) {  (void)argc;  (void)argv;  auto *context = zmq_ctx_new();  rpc_socket = zmq_socket(context, ZMQ_PAIR);  int rc = zmq_connect(rpc_socket, "tcp://127.0.0.1:9999");  if (rc == 0) {    std::cout << "Connected to 127.0.0.1:9999" << std::endl;  }  while (true) {    rpc_say("Hello rpc world");    auto res = rpc_add(5, 8);    std::cout << "for 5 + 8 server said: " << res << std::endl;    std::this_thread::sleep_for(std::chrono::seconds(1));  }  zmq_close(rpc_socket);  zmq_ctx_destroy(context);  return 0;}

int main (void){  // Prepare our context and publisher  void *context = zmq_ctx_new ();  void *publisher = zmq_socket (context, ZMQ_PUB);  void *receiver = zmq_socket (context, ZMQ_REP);  int rc = zmq_bind (publisher, "tcp://*:5556");  assert (rc == 0);  rc = zmq_bind (publisher, "ipc://chat.ipc");  assert (rc == 0);  rc = zmq_bind (receiver, "tcp://*:5558");  assert (rc == 0);  // Initialize random number generator  while (1) {    char* message = s_recv (receiver);    s_send (publisher, message);    free (message);  }  zmq_close (publisher);  zmq_ctx_destroy (context);  return 0;}

static void* routine(void* ctx){  int index = 0;  char buf[128];  void* rep = zmq_socket(ctx, ZMQ_REP);  zmq_connect(rep, "inproc://workers");  while (1) {    memset(buf, 0, sizeof(buf));    zmq_recv(rep, buf, sizeof(buf), 0);    fprintf(stdout, "recevied request: %s/n", buf);    usleep(100 * 1000);    sprintf(buf, "reply[%ud] result index : %d",       (unsigned int)pthread_self(), ++index);    zmq_send(rep, buf, strlen(buf), 0);  }  zmq_close(rep);  return NULL;}

int main(){    printf("Connection to hello world server..../n");    void *context = zmq_ctx_new();    void *socket = zmq_socket(context, ZMQ_REQ);    zmq_connect(socket, "tcp://localhost:8888");    int request_no;    for(request_no = 0; request_no != 10; ++request_no) {        char buffer[10];        printf("Sending Hello %d.../n", request_no);        if(-1 == zmq_send(socket, "Hello", 5, 0)){            printf("send message error, %d, %s/n", errno, zmq_strerror(errno));            break;        }        // zmq_recv(socket, buffer, 10, 0);        // printf("Received world %d/n", request_no);    }    zmq_close(socket);    zmq_ctx_destroy(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;}

void Van::monitor() {  VLOG(1) << "starting monitoring...";  void *s = CHECK_NOTNULL(zmq_socket(context_, ZMQ_PAIR));  CHECK(!zmq_connect(s, "inproc://monitor"));  while (true) {    zmq_msg_t msg;    zmq_msg_init(&msg);    if (zmq_msg_recv(&msg, s, 0) == -1) {      if (errno == EINTR)        continue;      break;    }    uint8_t *data = (uint8_t *)zmq_msg_data(&msg);    int event = *(reinterpret_cast<uint16_t *>(data));    int value = *(reinterpret_cast<uint32_t *>(data + 2));    if (event == ZMQ_EVENT_DISCONNECTED) {      auto &manager = PostOffice::getInstance().manager();      if (isScheduler()) {        Lock l(fdToNodeIdMu_);        if (fdToNodeId_.find(value) == fdToNodeId_.end()) {          LOG(WARNING) << "cannot find the node id for Fd = " << value;          continue;        }        manager.nodeDisconnected(fdToNodeId_[value]);      } else {        manager.nodeDisconnected(scheduler_.id());      }    }    if (event == ZMQ_EVENT_MONITOR_STOPPED) {      break;    }  }  zmq_close(s);  VLOG(1) << "monitor stopped";}

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);    // Subscriber 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);        printf("%03d   ", update_nbr);        printf(string);        printf(" %d/n");        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;}