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

void test_block_on_send_no_peers (void *ctx){    void *sc = zmq_socket (ctx, ZMQ_REQ);    assert (sc);    int timeout = 100;    int rc = zmq_setsockopt (sc, ZMQ_SNDTIMEO, &timeout, sizeof (timeout));    assert (rc == 0);    rc = zmq_send (sc, 0, 0, ZMQ_DONTWAIT);    assert (rc == -1);    assert (errno == EAGAIN);    rc = zmq_send (sc, 0, 0, 0);    assert (rc == -1);    assert (errno == EAGAIN);    rc = zmq_close (sc);    assert (rc == 0);}

int main (int argc, char *argv []){    const char *host;    int message_size;    int message_count;    void *handle;    int eid;    int counter;    void *out_buf;    int rc;    /*  Parse command line arguments.  */    if (argc != 4) {        printf ("usage: c_remote_thr <hostname> <message-size> "            "<message-count>/n");        return 1;    }    host = argv [1];    message_size = atoi (argv [2]);    message_count = atoi (argv [3]);        /*  Create 0MQ transport.  */    handle = zmq_create (host);    /*  Create the wiring.  */    eid = zmq_create_exchange (handle, "E", ZMQ_SCOPE_LOCAL, NULL,        ZMQ_STYLE_LOAD_BALANCING);    assert (eid != -1);    rc = zmq_bind (handle, "E", "Q", NULL, NULL);    assert (rc != -1);    /*  Create message data to send.  */    out_buf = malloc (message_size);    assert (out_buf);    for (counter = 0; counter != message_count + 1; counter ++)        zmq_send (handle, eid, out_buf, message_size, ZMQ_TRUE);    /*  Wait till all messages are sent.  */#ifdef ZMQ_HAVE_WINDOWS    Sleep (5000);#else    sleep (5);#endif    /*  Destroy 0MQ transport.  */    zmq_destroy (handle);    /*  Clean up.  */    free (out_buf);    return 0;}

static voidzmqdrv_send(zmq_drv_t *drv, ErlIOVec *ev){    ErlDrvBinary*  bin   = ev->binv[1];    char*          bytes = bin->orig_bytes;    uint32_t       idx   = ntohl(*(uint32_t*)(bytes+1));    zmq_sock_info* si    = drv->get_socket_info(idx);    uint32_t       flags = ntohl(*(uint32_t*)(bytes+5));    void*          data  = (void *)(bytes + 9);    size_t         size  = bin->orig_size - 9;    if (idx > drv->zmq_socket_count || !si) {        zmqdrv_error_code(drv, ENODEV);        return;    }#ifdef ZMQDRV_DEBUG    uint32_t events;    size_t events_size = sizeof(events);    zmq_getsockopt(si->socket, ZMQ_EVENTS, &events, &events_size);    zmqdrv_fprintf("sending %p [idx=%d] %lu bytes (events=%d)/r/n", si->socket, idx, size, events);#endif    if (si->out_caller != 0) {        // There's still an unwritten message pending        zmqdrv_error_code(drv, EBUSY);        return;    }    // Increment the reference count on binary so that zmq can    // take ownership of it.    driver_binary_inc_refc(bin);    if (zmq_msg_init_data(&si->out_msg, data, size, &zmq_free_binary, bin)) {        zmqdrv_error_code(drv, zmq_errno());        driver_binary_dec_refc(bin);        return;    }    if (zmq_send(si->socket, &si->out_msg, flags | ZMQ_NOBLOCK) == 0) {        zmqdrv_ok(drv);        zmqdrv_ready_input((ErlDrvData)drv, (ErlDrvEvent)si->fd);    } else {        int e = zmq_errno();        if (e == EAGAIN) {            // No msg returned to caller - make him wait until async            // send succeeds            si->out_caller = driver_caller(drv->port);            return;        }        zmqdrv_error_code(drv, e);    }    zmq_msg_close(&si->out_msg);}

    void reply_config()    {      //zmq_msg_t reply;      std::stringstream config_s;      config_s << config;      int reply_size = config_s.str().size();      zmq_send(zmq_responder,	       config_s.str().c_str(),	       reply_size,	       0);    }

/* Send formatted message to url */intlogevent(const char *url, const char *message){	int rc, result;	void *ctx, *socket;	zmq_msg_t query;	zmq_pollitem_t items[1];	/* Send the message */	rc = zmq_msg_init_size(&query, strlen(message));	assert(rc == 0);	memcpy(zmq_msg_data(&query), message, strlen(message));	ctx = zmq_init (1, 1, ZMQ_POLL);	assert(ctx);	socket = zmq_socket(ctx, ZMQ_REQ);	assert(socket);	rc = zmq_connect(socket, url);	assert(rc == 0);	rc = zmq_send(socket, &query, 0);	assert(rc == 0);	zmq_msg_close(&query);	/* Wait for a reply */	rc = zmq_msg_init(&query);	assert(rc == 0);	items[0].socket = socket;	items[0].events = ZMQ_POLLIN;	rc = my_zmqpoll(items, 1, SEND_TIMEOUT);	assert(rc >= 0);	rc = zmq_recv(socket, &query, ZMQ_NOBLOCK);	if (rc == -1 && errno == EAGAIN) {		result = 0;	} else {		assert(rc == 0);		zmq_msg_close(&query);		result = 1;	}	/* Clean up - FIXME: random SEGV on zmq_term	zmq_close(socket);	zmq_term(ctx);	*/	return result;}

void endFrame(){	dataCache_t * temp = dataCacheHead;	rI2CTX_beginFrame();	while(temp != NULL)	{/*		printf("%d ",temp->decParam.index);		switch(temp->decParam.type)		{			case rI2C_INT8: printf("/tINT8 %d/n",*(int8_t*)(temp->decParam.val));break;			case rI2C_UINT8: printf("/tUINT8 %d/n",*(uint8_t*)(temp->decParam.val));break;			case rI2C_INT16: printf("/tINT16 %d/n",*(int16_t*)(temp->decParam.val));break;			case rI2C_UINT16: printf("/tUINT16 %d/n",*(uint16_t*)(temp->decParam.val));break;			case rI2C_INT32: printf("/tINT32 %d/n",*(int32_t*)(temp->decParam.val));break;			case rI2C_UINT32: printf("/tUINT32 %d/n",*(uint32_t*)(temp->decParam.val));break;			case rI2C_INT64: printf("/tINT64 %"PRId64"/n",*(int64_t*)(temp->decParam.val));break;			case rI2C_UINT64: printf("/tUINT64 %"PRId64"/n",*(uint64_t*)(temp->decParam.val));break;			case rI2C_FLOAT: printf("/tFLOAT %f/n",*(float*)(temp->decParam.val));break;			case rI2C_DOUBLE: printf("/tDOUBLE %f/n",*(double*)(temp->decParam.val));break;		}*/		switch(temp->decParam.type)		{			case rI2C_INT8: rI2CTX_addParameter_int8(temp->decParam.index, *(int8_t*)(temp->decParam.val));break;			case rI2C_UINT8: rI2CTX_addParameter_uint8(temp->decParam.index,*(uint8_t*)(temp->decParam.val));break;			case rI2C_INT16: rI2CTX_addParameter_int16(temp->decParam.index,*(int16_t*)(temp->decParam.val));break;			case rI2C_UINT16: rI2CTX_addParameter_uint16(temp->decParam.index,*(uint16_t*)(temp->decParam.val));break;			case rI2C_INT32: rI2CTX_addParameter_int32(temp->decParam.index,*(int32_t*)(temp->decParam.val));break;			case rI2C_UINT32: rI2CTX_addParameter_uint32(temp->decParam.index,*(uint32_t*)(temp->decParam.val));break;			case rI2C_INT64: rI2CTX_addParameter_int64(temp->decParam.index,*(int64_t*)(temp->decParam.val));break;			case rI2C_UINT64: rI2CTX_addParameter_uint64(temp->decParam.index,*(uint64_t*)(temp->decParam.val));break;			case rI2C_FLOAT: rI2CTX_addParameter_float(temp->decParam.index,*(float*)(temp->decParam.val));break;			case rI2C_DOUBLE: rI2CTX_addParameter_double(temp->decParam.index,*(double*)(temp->decParam.val));break;		}		temp = temp->next;	}	rI2CTX_endFrame();	if((rI2CTX_bufferPos + strlen(nodeName) + strlen(nodeName)) < sizeof(buffer)){		char * t = "telemetry ";		memcpy(&buffer,t,strlen(t));		memcpy(buffer + strlen(t),nodeName,strlen(nodeName));		memcpy(buffer + strlen(t) + strlen(nodeName), &rI2CTX_buffer,rI2CTX_bufferPos + strlen(t) + strlen(nodeName));		zmq_send(pubTelemetry, buffer, rI2CTX_bufferPos+strlen(t)+strlen(nodeName), 0);	}else{		printf("ZMQ TX buffer too small./n");	}}

void *run(void *arg){        int ret;        char buf[256];        while (1) {                        zmq_recv(rep, buf, sizeof(buf), 0);                        printf("rep recv: %s/n", buf);                        ret = zmq_send(rep, "rep recv done", 16, 0);                        assert(ret == 16);        }}

int main (int argc, char *argv []){    const char *bind_to;    int roundtrip_count;    int 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, 1, 0);    assert (ctx);    s = zmq_socket (ctx, ZMQ_REP);    assert (s);    rc = zmq_bind (s, bind_to);    assert (rc == 0);    rc = zmq_msg_init (&msg);    assert (rc == 0);    for (i = 0; i != roundtrip_count; i++) {        rc = zmq_recv (s, &msg, 0);        assert (rc == 0);        assert (zmq_msg_size (&msg) == message_size);        rc = zmq_send (s, &msg, 0);        assert (rc == 0);    }    rc = zmq_msg_close (&msg);    assert (rc == 0);    zmq_sleep (1);    rc = zmq_close (s);    assert (rc == 0);    rc = zmq_term (ctx);    assert (rc == 0);    return 0;}

void Chat::run(){	char buff[1024] = { 0 };	zmq_pollitem_t items[3] = { 0 };	items[0].socket = subscriber_;	items[0].events = ZMQ_POLLIN;	items[1].socket = gate_;	items[1].events = ZMQ_POLLIN;	items[2].fd = STDIN_FILENO;	items[2].events = ZMQ_POLLIN;	while(true)	{		int len;		int ret = zmq_poll(items, 3, -1);		if (items[0].revents & ZMQ_POLLIN)		{			len = zmq_recv(items[0].socket, buff, sizeof(buff), 0);			if (len > 0)			{				buff[len] = 0;				std::cout << buff << std::endl;			}			else			{				std::cerr << zmq_strerror(errno);			}		}		if (items[1].revents & ZMQ_POLLIN)		{			len = zmq_recv(items[1].socket, buff, sizeof(buff), 0);			if (len > 0)			{				handle_gate((Action*)buff);			}			else			{				std::cerr << zmq_strerror(errno);			}		}		if (items[2].revents & ZMQ_POLLIN)		{			std::cin.getline(buff, sizeof(buff));			zmq_send(publisher_, buff, std::cin.gcount(), 0);		}	}}

void Statistic::reporter(){  LOG_DEBUG("Initializing inproc://%s statistic reporter", _statname.c_str());  _publisher = stack_data.stats_aggregator->get_internal_publisher(_statname);  std::vector<std::string> new_value;  while (_stat_q.pop(new_value))  {    LOG_DEBUG("Send new value for statistic %s, size %d", _statname.c_str(), new_value.size());    std::string status = "OK";    // If there's no message, just send the envelope and status line.    if (new_value.empty())    {      zmq_send(_publisher, _statname.c_str(), _statname.length(), ZMQ_SNDMORE);      zmq_send(_publisher, status.c_str(), status.length(), 0);    }    else    {      // Otherwise send the envelope, status line, and body, remembering to set SNDMORE on all      // but the last section.      zmq_send(_publisher, _statname.c_str(), _statname.length(), ZMQ_SNDMORE);      zmq_send(_publisher, status.c_str(), status.length(), ZMQ_SNDMORE);      std::vector<std::string>::iterator it;      for (it = new_value.begin(); it + 1 != new_value.end(); ++it)      {        zmq_send(_publisher, it->c_str(), it->length(), ZMQ_SNDMORE);      }      zmq_send(_publisher, it->c_str(), it->length(), 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;}

int main(int argc, char *argv[]){  if (argc < 2) return EXIT_FAILURE;  int N = atoi(argv[1]);  printf("N: %d/n", N);  void *ctx = zmq_init(1);  void *server = zmq_socket(ctx, ZMQ_REQ); // Server  assert(server);  int rc;  rc = zmq_connect(server, "tcp://localhost:8889"); // Actively connect to subscribers  assert(rc == 0);  int *val = (int *)calloc(N, sizeof(int));  zmq_msg_t msg;  long long start_time = sc_time();  // Send  int *buf = (int *)calloc(N, sizeof(int));  memcpy(buf, val, N * sizeof(int));  zmq_msg_init_data(&msg, buf, N * sizeof(int), _dealloc, NULL);  zmq_send(server, &msg, 0);  zmq_msg_close(&msg);  // Receive  zmq_msg_init(&msg);  zmq_recv(server, &msg, 0);  memcpy(val, (int *)zmq_msg_data(&msg), zmq_msg_size(&msg));  zmq_msg_close(&msg);  long long end_time = sc_time();  printf("%s: Time elapsed: %f sec/n", argv[0], sc_time_diff(start_time, end_time));#ifdef __DEBUG__  int i;  printf("%s [ ", argv[0]);  for (i=0; i<N; ++i) {    printf("%d ", val[i]);  }  printf("]/n");#endif  free(val);  zmq_close(server);  zmq_term(ctx);  return EXIT_SUCCESS;}

	void ZmqAppender::append(const log4cplus::spi::InternalLoggingEvent& event)	{		if (m_responder != NULL)		{			log4cplus::tostringstream oss;			layout->formatAndAppend(oss, event);			tstring formattedEvent = oss.str();			zmq_send(m_responder, formattedEvent.c_str(), formattedEvent.length(), 0);			char controlByte;			zmq_recv(m_responder, &controlByte, 1, 0);		}	}

void Heartbeat::report_newcomer(const std::string& peer_id, const std::string& peer_ip){	Action action;	action.type_ = JOIN;	strcpy(action.id_, peer_id.c_str());	strcpy(action.ip_, peer_ip.c_str());	int ret = zmq_send(gate_, (void*)&action, sizeof(action), 0);	if (ret < 0)	{		zmq_strerror(errno);	}}

int main(void){  void *ctx = zmq_ctx_new();  void *sock = zmq_socket(ctx, ZMQ_ROUTER);  int rc = zmq_bind(sock, "tcp://127.0.0.1:5555");  assert(0 == rc);  char buf[128];  int recvBytes = 0;  while (1) {    // recv  id    recvBytes = zmq_recv(sock, buf, sizeof(buf), 0);    if (0 < recvBytes) {      // send it back      printf("recv id bytes=%d/n", recvBytes);      zmq_send(sock, buf, recvBytes, ZMQ_SNDMORE);    }    // delimiter emtpy frame    recvBytes = zmq_recv(sock, buf, sizeof(buf), 0);    assert(0 == recvBytes);    recvBytes = zmq_recv(sock, buf, sizeof(buf), 0);    // actual message    if (0 < recvBytes) {      // send it back      buf[recvBytes] = '/0';      printf("recv msg %s, bytes=%d/n", buf, recvBytes);      zmq_send(sock, "", 0, ZMQ_SNDMORE);      zmq_send(sock, buf, recvBytes, 0);    }  }  zmq_close(sock);  zmq_ctx_destroy(ctx);  return 0;}

int main (void){    void *ctx = zmq_ctx_new ();    assert (ctx);    void *router = zmq_socket (ctx, ZMQ_ROUTER);    assert (router);    int rc = zmq_bind (router, "tcp://127.0.0.1:5560");    assert (rc == 0);    //  Send a message to an unknown peer with the default setting    //  This will not report any error    rc = zmq_send (router, "UNKNOWN", 7, ZMQ_SNDMORE);    assert (rc == 7);    rc = zmq_send (router, "DATA", 4, 0);    assert (rc == 4);    //  Send a message to an unknown peer with mandatory routing    //  This will fail    int mandatory = 1;    rc = zmq_setsockopt (router, ZMQ_ROUTER_MANDATORY, &mandatory, sizeof (mandatory));    assert (rc == 0);    rc = zmq_send (router, "UNKNOWN", 7, ZMQ_SNDMORE);    assert (rc == -1 && errno == EHOSTUNREACH);    //  Create dealer called "X" and connect it to our router    void *dealer = zmq_socket (ctx, ZMQ_DEALER);    assert (dealer);    rc = zmq_setsockopt (dealer, ZMQ_IDENTITY, "X", 1);    assert (rc == 0);    rc = zmq_connect (dealer, "tcp://127.0.0.1:5560");    assert (rc == 0);    //  Get message from dealer to know when connection is ready    char buffer [255];    rc = zmq_send (dealer, "Hello", 5, 0);    assert (rc == 5);    rc = zmq_recv (router, buffer, 255, 0);    assert (rc == 1);    assert (buffer [0] ==  'X');    //  Send a message to connected dealer now    //  It should work    rc = zmq_send (router, "X", 1, ZMQ_SNDMORE);    assert (rc == 1);    rc = zmq_send (router, "Hello", 5, 0);    assert (rc == 5);        rc = zmq_close (router);    assert (rc == 0);    rc = zmq_close (dealer);    assert (rc == 0);    rc = zmq_ctx_term (ctx);    assert (rc == 0);    return 0 ;}

void run_server(){    zctx_t *ctx;    void *socket;    int i, n, nclient;    char buf[256];    char data[256];    char client[256];    ctx = new_context();    socket = new_socket(ctx, ZMQ_ROUTER);//  socket = new_socket(ctx, ZMQ_REP);    assert_result(socket_bind(socket, host), 0);//  assert_result(zmq_bind(socket, host), 0);    log_printf(0, "my identity=%s/n", zsocket_identity(socket));    i = 0;    for (;;) {        log_printf(0, "Waiting %d/n", i);        nclient = zmq_recv(socket, client, sizeof(client), 0);        client[nclient] = 0;        log_printf(0, "From %s [%d]/n", client, nclient);        n = zmq_recv(socket, buf, sizeof(buf), 0);        buf[n] = 0;        if (n != 0) log_printf(0, "Missing EMPTY frame! buf=%s/n", buf);        n = zmq_recv(socket, buf, sizeof(buf), 0);        buf[n] = 0;        log_printf(0, "Got %s/n", buf);        zmq_send(socket, client, nclient, ZMQ_SNDMORE);        zmq_send(socket, NULL, 0, ZMQ_SNDMORE);        snprintf(data, sizeof(buf), "(%s) World %d", buf, i);        zmq_send(socket, data, strlen(data)+1, 0);        i++;    }    destroy_context(ctx);}

void Heartbeat::report_departure(const std::string& peer_id, const std::string& peer_ip){	Action action;	action.type_ = LEAVE;	memcpy(action.id_, peer_id.c_str(), peer_id.length());	memcpy(action.ip_, peer_ip.c_str(), peer_ip.length());	int ret = zmq_send(gate_, (void*)&action, sizeof(action), 0);	if (ret < 0)	{		zmq_strerror(errno);	}}

void EventModule::SendConnStop(void* zmq_sock, int32_t player_idx){    LOG(INFO) << "player_idx[" << player_idx << "] send CONN_STOP to gamesvr ";    if (player_idx == 0)        return;    static ConnData conn_data;    conn_data.conn_cmd = CONN_STOP;    conn_data.conn_fd = 0;    conn_data.player_idx = player_idx;    zmq_send(zmq_sock, (void*)&conn_data, sizeof(conn_data), ZMQ_DONTWAIT);}

void deliver_message(void* sock, int size, int flags){  zmq_msg_t msg;  int retval;  // Prep the message from preset value. ZMQ takes ownership.  void *data = calloc(size,1);  assert(data != NULL);  for(int i=0; i<size; i++){    ((char*) data)[i] = 'A';  }  retval = zmq_msg_init_data(&msg, data, size, free_msg, NULL);  assert(retval == 0);  // printf("sending %s/n",zmq_msg_data(&msg));  retval = zmq_send(sock,&msg,flags);  if(retval != 0){    switch(errno){      case EAGAIN:        puts("zmq_send EAGAIN");        break;      case ENOTSUP:        puts("zmq_send ENOTSUP");        break;      case EFSM:        puts("zmq_send EFSM");        break;      case ETERM:        puts("zmq_send ETERM");        break;      case ENOTSOCK:        puts("zmq_send ENOTSOCK");        break;      case EINTR:        puts("zmq_send EINTR");        break;      case EFAULT:        puts("zmq_send EFAULT");        break;      default:        puts("zmq_send: errno default");    }    exit(EXIT_FAILURE);  } else {    puts("sent the message");  }  // Cleanup  printf("cleaning up msg...");  zmq_msg_close(&msg);}

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

//  Same as bounce, but expect messages to never arrive//  for security or subscriber reasons.voidexpect_bounce_fail (void *server, void *client){    const char *content = "12345678ABCDEFGH12345678abcdefgh";    char buffer [32];    int timeout = 250;    //  Send message from client to server    int rc = zmq_setsockopt (client, ZMQ_SNDTIMEO, &timeout, sizeof (int));    assert (rc == 0);    rc = zmq_send (client, content, 32, ZMQ_SNDMORE);    assert ((rc == 32) || ((rc == -1) && (errno == EAGAIN)));    rc = zmq_send (client, content, 32, 0);    assert ((rc == 32) || ((rc == -1) && (errno == EAGAIN)));    //  Receive message at server side (should not succeed)    rc = zmq_setsockopt (server, ZMQ_RCVTIMEO, &timeout, sizeof (int));    assert (rc == 0);    rc = zmq_recv (server, buffer, 32, 0);    assert (rc == -1);    assert (zmq_errno () == EAGAIN);    //  Send message from server to client to test other direction    //  If connection failed, send may block, without a timeout    rc = zmq_setsockopt (server, ZMQ_SNDTIMEO, &timeout, sizeof (int));    assert (rc == 0);    rc = zmq_send (server, content, 32, ZMQ_SNDMORE);    assert (rc == 32 || (rc == -1 && zmq_errno () == EAGAIN));    rc = zmq_send (server, content, 32, 0);    assert (rc == 32 || (rc == -1 && zmq_errno () == EAGAIN));    //  Receive message at client side (should not succeed)    rc = zmq_setsockopt (client, ZMQ_RCVTIMEO, &timeout, sizeof (int));    assert (rc == 0);    rc = zmq_recv (client, buffer, 32, 0);    assert (rc == -1);    assert (zmq_errno () == EAGAIN);}

static int cli_zmq_send(struct teamdctl *tdc, void *sock, char *buf){    int ret;    int buflen = strlen(buf);    ret = zmq_send(sock, buf, buflen, 0);    if (ret == -1) {        warn(tdc, "zmq: send failed: %s", strerror(errno));        return -errno;    }    free(buf);    return 0;}

voidsend(char *buf, size_t size){    zmq_msg_t msg;    void *sock, *context;    context = zmq_init(1);    sock = zmq_socket(context, ZMQ_PUSH);    zmq_connect(sock, "tcp://localhost:5555");    zmq_msg_init_data(&msg, buf, size, NULL, NULL);    zmq_send(sock, &msg, 0);    zmq_close(sock);    zmq_term(context);}

void check_proxy_stats (void *control_proxy){    zmq_proxy_stats_t total_stats;    int rc;    rc = zmq_send (control_proxy, "STATISTICS", 10, 0);    assert (rc == 10);    // first frame of the reply contains FRONTEND stats:    total_stats.frontend.msg_in = recv_stat (control_proxy, false);    total_stats.frontend.bytes_in = recv_stat (control_proxy, false);    total_stats.frontend.msg_out = recv_stat (control_proxy, false);    total_stats.frontend.bytes_out = recv_stat (control_proxy, false);    // second frame of the reply contains BACKEND stats:    total_stats.backend.msg_in = recv_stat (control_proxy, false);    total_stats.backend.bytes_in = recv_stat (control_proxy, false);    total_stats.backend.msg_out = recv_stat (control_proxy, false);    total_stats.backend.bytes_out = recv_stat (control_proxy, true);    // check stats    if (is_verbose) {        printf (          "frontend: pkts_in=%lu bytes_in=%lu  pkts_out=%lu bytes_out=%lu/n",          (unsigned long int) total_stats.frontend.msg_in,          (unsigned long int) total_stats.frontend.bytes_in,          (unsigned long int) total_stats.frontend.msg_out,          (unsigned long int) total_stats.frontend.bytes_out);        printf (          "backend: pkts_in=%lu bytes_in=%lu  pkts_out=%lu bytes_out=%lu/n",          (unsigned long int) total_stats.backend.msg_in,          (unsigned long int) total_stats.backend.bytes_in,          (unsigned long int) total_stats.backend.msg_out,          (unsigned long int) total_stats.backend.bytes_out);        printf ("clients sent out %d requests/n",                zmq_atomic_counter_value (g_clients_pkts_out));        printf ("workers sent out %d replies/n",                zmq_atomic_counter_value (g_workers_pkts_out));    }    assert (total_stats.frontend.msg_in            == (unsigned) zmq_atomic_counter_value (g_clients_pkts_out));    assert (total_stats.frontend.msg_out            == (unsigned) zmq_atomic_counter_value (g_workers_pkts_out));    assert (total_stats.backend.msg_in            == (unsigned) zmq_atomic_counter_value (g_workers_pkts_out));    assert (total_stats.backend.msg_out            == (unsigned) zmq_atomic_counter_value (g_clients_pkts_out));}

static intrawsend(void *socket, void *data, size_t size){	int rc;	zmq_msg_t msg;	rc = zmq_msg_init_size(&msg, size);	assert(rc == 0);	memcpy(zmq_msg_data(&msg), data, size);	rc = zmq_send(socket, &msg, 0);	assert(rc == 0);	zmq_msg_close(&msg);	return 1;}

/** * send an empty message * which will be received as bang * can be used for signalling/heartbeat*/static void zmq_bang(t_zmq *x) {   if ( ! _can_send(x)) {       return;   }   if ( ! x->zmq_socket) {      error("create and connect socket before sending anything");      return;   }   int r=zmq_send (x->zmq_socket, "", 0, ZMQ_DONTWAIT);   if(r == -1) {      _zmq_error(zmq_errno());      return;   }}

int main(int argc, char **argv) {  logging_init("cli.log");  void *ctx = zmq_init(1);  void *socket = zmq_socket(ctx, ZMQ_DEALER);  int rc = zmq_connect(socket, "tcp://127.0.0.1:9999");  // use below for when this is a router socket  // int rc = zmq_bind(socket, "tcp://127.0.0.1:9999");  assert(rc == 0);  zmq_msg_t msg;  size_t msg_size = sizeof(int);  int x = 99;  boost::posix_time::ptime start_ptime(      boost::posix_time::microsec_clock::local_time());  sleep(3);#ifdef ZC  // with zerocopy  rc = zmq_msg_init_data(&msg, (void *)&x, msg_size, NULL, NULL);  assert(rc == 0);#else  // without zerocopy  pan::log_DEBUG("NOT using zerocopy");  rc = zmq_msg_init_size(&msg, msg_size);  assert(rc == 0);  memcpy(zmq_msg_data(&msg), (void *)&x, msg_size);#endif  // use below line for router socket  // must be the same identity as the receiving socket has set  // s_sendmore(socket, "A");  rc = zmq_send(socket, &msg, 0);  assert(rc == 0);  zmq_msg_close(&msg);  pan::log_DEBUG("Receiving message");  zmq_msg_init(&msg);  rc = zmq_recv(socket, &msg, 0);  pan::log_DEBUG("Received msg: ", pan::integer(*(int *)zmq_msg_data(&msg)));  zmq_close(&msg);  boost::posix_time::ptime stop_ptime(      boost::posix_time::microsec_clock::local_time());  boost::posix_time::time_duration ptime_duration(stop_ptime - start_ptime);  std::cerr << ptime_duration << "/n";  zmq_close(socket);  zmq_term(ctx);}

static intsend(void *socket, char *str){	int rc;	zmq_msg_t msg;		rc = zmq_msg_init_size(&msg, strlen(str) + 1);	assert(rc == 0);	memcpy(zmq_msg_data(&msg), str, strlen(str) + 1);	rc = zmq_send(socket, &msg, 0);	assert(rc == 0);	zmq_msg_close(&msg);	return 1;}