自学教程：C++ ACE_SOCK_Stream类代码示例

intFT_EventService::report_factory(CORBA::ORB_ptr orb,                   FtRtecEventChannelAdmin::EventChannel_ptr ec){  try{    char* addr = ACE_OS::getenv("EventChannelFactoryAddr");    if (addr != 0) {      // instaniated by object factory, report my ior back to the factory      ACE_INET_Addr factory_addr(addr);      ACE_SOCK_Connector connector;      ACE_SOCK_Stream stream;      ORBSVCS_DEBUG((LM_DEBUG,"connecting to %s/n",addr));      if (connector.connect(stream, factory_addr) == -1)        ORBSVCS_ERROR_RETURN((LM_ERROR, "(%P|%t) Invalid Factory Address/n"), -1);      ORBSVCS_DEBUG((LM_DEBUG,"Factory connected/n"));      CORBA::String_var my_ior_string = orb->object_to_string(ec);      int len = ACE_OS::strlen(my_ior_string.in()) ;      if (stream.send_n(my_ior_string.in(), len) != len)        ORBSVCS_ERROR_RETURN((LM_ERROR, "(%P|%t) IOR Transmission Error/n"), -1);      stream.close();    }  }  catch (...){    return -1;  }  return 0;}

intACE_SOCK_Connector::shared_connect_start (ACE_SOCK_Stream &new_stream,                                          const ACE_Time_Value *timeout,                                          const ACE_Addr &local_sap){  ACE_TRACE ("ACE_SOCK_Connector::shared_connect_start");  if (local_sap != ACE_Addr::sap_any)    {      sockaddr *laddr = reinterpret_cast<sockaddr *> (local_sap.get_addr ());      int size = local_sap.get_size ();      if (ACE_OS::bind (new_stream.get_handle (),                        laddr,                        size) == -1)        {          // Save/restore errno.          ACE_Errno_Guard error (errno);          new_stream.close ();          return -1;        }    }  // Enable non-blocking, if required.  if (timeout != 0      && new_stream.enable (ACE_NONBLOCK) == -1)    return -1;  else    return 0;}

/*fence: 1) send the heartbeat message to repo at regular basis. 1/1 minute, get back the messages,calculate the digest, and send to localhost:99072)listens on localhost:9901, for the incoming raw inputs, calculate the digest, and send it to hub:10007*/intACE_TMAIN(int argc, ACE_TCHAR* argv[]){  ACE_SOCK_Acceptor _9901acceptor;    ACE_INET_Addr _9901addr(9901);  //create the acceptor  if(_9901acceptor.open(_9901addr,1)==-1){    ACE_ERROR_RETURN((LM_ERROR,		      "%p/n","open"),1);  }else if(_9901acceptor.get_local_addr(_9901addr)== -1){    ACE_ERROR_RETURN((LM_ERROR,		      "%p/n","get_local_addr"),1);  }  ACE_DEBUG((LM_INFO,	     "(%P|%t) starting server at port %d/n",	     _9901addr.get_port_number()));  //  ACE_INET_Addr repo_addr(repo_port,repo_host.c_str());  ACE_SOCK_Connector con;  //  ACE_SOCK_Stream cli_stream ;  ACE_Thread_Manager* mgr = ACE_Thread_Manager::instance();//  if(con.connect(cli_stream,repo_addr)==-1){//    ACE_ERROR_RETURN((LM_ERROR,//		      "(%P|%t:%l) %p/n","connection failed"),0);//  }else{//    ACE_DEBUG((LM_DEBUG,//	       "(%P|%t) connected to %s at port %d/n",repo_addr.get_host_name(),repo_addr.get_port_number()));//  }  /*connector side; do in a seperate thread;   */  if(mgr->spawn(fetch_step2,		0,		THR_DETACHED) == -1){    ACE_ERROR ((LM_ERROR,                "(%P|%t) %p/n",                "spawn"));  }  /*    run the accept loop ;   */  do{    ACE_SOCK_Stream stream;    if(_9901acceptor.accept(stream)== -1){      ACE_ERROR_RETURN((LM_ERROR,			"(%P|%t:%l) %p/n","accept failed"),0);    }else{      ACE_DEBUG((LM_DEBUG,		 "(%P|%t:%l) connected to %s at port %d/n",_9901addr.get_host_name(),_9901addr.get_port_number()));    }    if(mgr->spawn(accept_step1,		  reinterpret_cast<void *> (stream.get_handle()),		  THR_DETACHED) == -1){      ACE_ERROR ((LM_ERROR,		  "(%P|%t) %p/n",		  "spawn"));    }  }while(true);    return 0;}

void Collector::PollData(){	ACE_SOCK_Stream peer;	ACE_SOCK_Connector connector;	try	{		if(TryToConnect(connector,peer))		{			std::ostringstream oss;			oss << "poll data form cluster: " << data_source_.name;			LOG4CXX_INFO(log_,oss.str());			char buf[T_SIZE] = "request";			if((peer.send(buf,sizeof(buf)))<=0)				throw "KSN POLLER COLLECTOR SEND REQUEST ERROR!!";			char* ch = new char[T_SIZE]();			if((peer.recv(ch,T_SIZE)<=0))				throw "KSN POLLER COLLECTOR RECV ERROR!!";			msg_ = new ACE_Message_Block(ch,sizeof(*ch));			sender_->putq(msg_);		}		else		{			std::ostringstream oss;			oss << "Can not poll data form " + data_source_.name;			LOG4CXX_WARN(log_,oss.str());		}	}	catch(const char* ch)	{		LOG4CXX_ERROR(log_,ch);	}}

/////////////////////////////////////////////////////////////////////////////  <summary>//  This method writes data from the given buffer to the underlying stream.//  It can block or not, depending on the value of the blocking parameter.//  </summary>////  <param name = "buffer">//  The buffer that contains the data to be written.//  </param>////  <param name = "size">//  The size of the buffer in bytes of the buffer.//  </param>////  <param name = "blocking">//  True if the write request should block;  false otherwise.//  </param>////  <param name = "bytesWritten">//  An out parameter that will contain the number of bytes that have been//  written to the stream.//  </param>////  <returns>//  Returns a MgStreamStatus value indicating the status of the operation.//  </returns>MgStreamHelper::MgStreamStatus MgAceStreamHelper::WriteData(void* buffer,    size_t size, bool blocking, size_t* bytesWritten){    // Do not attempt writing zero byte to the socket as this could be problematic.    if (0 == size)    {        return MgStreamHelper::mssDone;    }    ACE_ASSERT( buffer && size > 0 );    MgStreamHelper::MgStreamStatus stat = MgStreamHelper::mssError;    //  check parameters    if ( buffer && size > 0 )    {        //  init out parameter        if ( bytesWritten != NULL )            *bytesWritten = 0;        ACE_SOCK_Stream stream;        stream.set_handle( m_handle );        ssize_t res = 0;        // On Linux, use MSG_NOSIGNAL to request not to send SIGPIPE on        // errors on stream oriented sockets when the other end breaks        // the connection. The EPIPE error is still returned.        // Note that neither trapping the SIGPIPE signal via an        // ACE_Event_Handler nor calling ACE_OS::signal(SIGPIPE, SIG_IGN)        // seems to work.        if ( blocking )        {            res = stream.send_n(buffer, size, MG_MSG_NOSIGNAL);        }        else        {            res = stream.send(buffer, size, MG_MSG_NOSIGNAL);        }        //  check for failure        if ( res >= 0 )        {            //  update out parameter            if ( bytesWritten != NULL )                *bytesWritten = res;            if ( res == (ssize_t)size )            {                stat = MgStreamHelper::mssDone;            }            else            {                stat = blocking ? MgStreamHelper::mssError : MgStreamHelper::mssNotDone;            }        }    }    return stat;}

intACE_TMAIN(int argc, ACE_TCHAR* argv[]){  ACE_SOCK_Acceptor acceptor;  ACE_INET_Addr addr(10009);  if(acceptor.open(addr,1)){    ACE_ERROR_RETURN((LM_ERROR,		      "%p/n", "open"),1);  }  else if(acceptor.get_local_addr(addr) == -1){    ACE_ERROR_RETURN((LM_ERROR,		      "%p/n", "get_local_addr"),1);  }  ACE_DEBUG((LM_INFO,	     "(%P|%t) starting server at port %d/n",addr.get_port_number()));  ACE_Thread_Manager* mgr = ACE_Thread_Manager::instance();  while(true){    ACE_SOCK_Stream stream;    if(acceptor.accept(stream) == -1){      ACE_ERROR((LM_ERROR,		 "%p/n","accept"));      continue;    }else{      ACE_DEBUG((LM_DEBUG,		 "(%P|%t) spawning one thread/n"));      handle_input(mgr, commu, stream.get_handle());    }  }  return 0;}

intSynch_Thread_Pool_Task::svc (void){    // Creates a factory of HTTP_Handlers binding to synchronous I/O strategy    Synch_HTTP_Handler_Factory factory;    for (;;)    {        ACE_SOCK_Stream stream;        // Lock in this accept.  When it returns, we have a connection.        if (this->acceptor_.accept (stream) == -1)            ACE_ERROR_RETURN ((LM_ERROR, "%p/n", "HTTP_Acceptor::accept"), -1);        ACE_Message_Block *mb;        ACE_NEW_RETURN (mb,                        ACE_Message_Block (HTTP_Handler::MAX_REQUEST_SIZE + 1),                        -1);        // Create an HTTP Handler to handle this request        HTTP_Handler *handler = factory.create_http_handler ();        handler->open (stream.get_handle (), *mb);        // Handler is destroyed when the I/O puts the Handler into the        // done state.        mb->release ();        ACE_DEBUG ((LM_DEBUG,                    " (%t) in Synch_Thread_Pool_Task::svc, recycling/n"));    }    ACE_NOTREACHED(return 0);}

voidJAWS_Synch_IO_No_Cache::send_message (const char *buffer, int length){  ACE_SOCK_Stream stream;  stream.set_handle (this->handle_);  stream.send_n (buffer, length);}

voidJAWS_Synch_IO::receive_file (const char *filename,                             void *initial_data,                             int initial_data_length,                             int entire_length){  ACE_Filecache_Handle handle (ACE_TEXT_CHAR_TO_TCHAR (filename), entire_length);  int result = handle.error ();  if (result == ACE_Filecache_Handle::ACE_SUCCESS)    {      ACE_SOCK_Stream stream;      stream.set_handle (this->handle_);      int bytes_to_memcpy = ACE_MIN (entire_length, initial_data_length);      ACE_OS::memcpy (handle.address (), initial_data, bytes_to_memcpy);      int bytes_to_read = entire_length - bytes_to_memcpy;      int bytes = stream.recv_n ((char *) handle.address () + initial_data_length,                                 bytes_to_read);      if (bytes == bytes_to_read)        this->handler_->receive_file_complete ();      else        result = -1;    }  if (result != ACE_Filecache_Handle::ACE_SUCCESS)    this->handler_->receive_file_error (result);}

voidJAWS_Synch_IO::receive_file (JAWS_IO_Handler *ioh,                             const char *filename,                             void *initial_data,                             unsigned int initial_data_length,                             unsigned int entire_length){  ACE_Filecache_Handle handle (filename,                               (int) entire_length);  int result = handle.error ();  if (result == ACE_Filecache_Handle::ACE_SUCCESS)    {      ACE_SOCK_Stream stream;      stream.set_handle (ioh->handle ());      int bytes_to_memcpy = ACE_MIN (entire_length, initial_data_length);      ACE_OS::memcpy (handle.address (), initial_data, bytes_to_memcpy);      int bytes_to_read = entire_length - bytes_to_memcpy;      int bytes = stream.recv_n ((char *)                                 handle.address () + initial_data_length,                                 bytes_to_read);      if (bytes == bytes_to_read)        ioh->receive_file_complete ();      else        result = -1;    }  if (result != ACE_Filecache_Handle::ACE_SUCCESS)    ioh->receive_file_error (result);}

int	extractCron::do_job(char *jobdesc){	ACE_SOCK_Connector conn;	ACE_SOCK_Stream  peer;	ACE_Time_Value tv(3, 0);	ACE_INET_Addr	addr(harvestPort, harvestIPaddr);	char	urlbuff[1024];	StrStream	httpreq;	int	ret;	printf("JobDesc: %s/n", jobdesc);	CGI::escape(urlbuff, jobdesc);	httpreq.init(2); // 2 kilobytes buff	httpreq.rawadd("GET /harvest?expr=");	httpreq.rawadd(urlbuff);	httpreq.rawadd(" HTTP/1.1/n/n");	httpreq.print();		if ( conn.connect(peer, addr) < 0) {		printf("conn failed!/n");		return 0;	}	ret = peer.send( httpreq.str(), httpreq.len() );	return 0;	}

bool MgAceStreamHelper::IsConnected(){    bool bConnected = true;    ACE_SOCK_Stream stream;    stream.set_handle( m_handle );    UINT8 dummy;    ACE_Time_Value val(0, 0);    ssize_t res = stream.recv_n(&dummy, 1, MSG_PEEK | MG_MSG_NOSIGNAL, &val);    if ( res < 0 )    {        // Error or timeout occured#ifdef _WIN32        int error = ::WSAGetLastError(); // errno doesn't work correctly on Windows        bConnected = ( error == WSAEWOULDBLOCK || error == 0 );#else        bConnected = ( errno == EWOULDBLOCK || errno == 0 || errno == ETIME );#endif    }    else if (res == 0)    {        // No longer connected        bConnected = false;    }    return bConnected;}

int logClient::process(ACE_CString* s){  ACE_SOCK_Stream logger;  ACE_SOCK_Connector connector;  ACE_INET_Addr addr(9876, "127.0.0.1");  if(connector.connect(logger, addr) == -1){    ACE_ERROR_RETURN((LM_ERROR, ACE_TEXT("%p /n"), ACE_TEXT("open")), -1);  }  ACE_Log_Record record(LM_DEBUG,			ACE_OS::time ((time_t *) 0),			ACE_OS::getpid());  record.msg_data(s.c_str());  const size_t max_payload_size =    4    + 8    + 4    + 4    + ACE_Log_Record::MAXLOGMSGLEN    + ACE_CDR::MAX_ALIGNMENT;  ACE_OutputCDR payload(max_payload_size);  payload<< record;  ACE_CDR::ULong length =    ACE_Utils::truncate_cast<ACE_CDR::ULong> (payload.total_length());  ACE_OutputCDR header (ACE_CDR::MAX_ALIGNMENT + 8);  header << ACE_OutputCDR::from_boolean (ACE_CDR_BYTE_ORDER);  header << ACE_CDR::ULong(length);  iovec iov[2];  iov[0].iov_base = header.begin() -> rd_ptr();  iov[0].iov_len = 8;  iov[1].iov_base = payload.begin() -> rd_ptr();  iov[1].iov_len = length;  if (logger.sendv_n(iov, 2) ==  -1)    ACE_ERROR_RETURN((LM_ERROR,"%p/n","send"), -1);  /*    */  ACE_Message_Block* header_p;  auto_ptr<ACE_Message_Block> header(header_p);  ACE_CDR::mb_align(header.get());    ACE_Message_Block* payload_p;  ssize_t count = logger.recv_n(header->wr_ptr(),8);  switch(count){  default:  case -1:  case 0:  case 8:    break;  }  header->wr_ptr(8);  }

int main(int argc, char* argv[]){	if (argc > 1){		return 1;	}		cout << argv[0] << endl;	ACE_INET_Addr addr(1234, ACE_LOCALHOST);	ACE_SOCK_Stream stream;	ACE_SOCK_Acceptor acceptor;	int success = acceptor.open(addr, 1);		ACE_TCHAR addrStr[20];	if (success > 0) {		addr.addr_to_string(addrStr, 20);	}			//*	success = acceptor.accept(stream);	if (success < 0) {		cout << "Cannot accept" << endl;		return 1;	}	//*/	char buf[BUFSIZ];	int n;	char *msg = "You are connected";	stream.send_n(msg, strlen(msg));	stream.close();		/*	while (stream.recv(buf, BUFSIZ)) {		// _write(1, buf, n);		cout << buf << endl;	}		//*/	cout << endl << "Done" << endl;	return 0;}

// thread function that serves the client for the UnMarshalled Octet// teststatic ACE_THR_FUNC_RETURN unmarshalledOctetServer (void *arg){  // unbundle the arguments  ArgStruct * args = reinterpret_cast<ArgStruct *> (arg);  ACE_SOCK_Stream * dataModeStream = args->stream;  ACE_CDR::ULong numIterations = args->numIters;  delete args;  // serve the client for numIterations synchronous invocations  do {    // READ A MESSAGE FROM THE CLIENT    size_t bt;    ACE_CDR::ULong msgBufSize=0;    // read the size of the buffer to follow    if ((dataModeStream->recv_n(&msgBufSize, ACE_CDR::LONG_SIZE, 0, &bt)) == -1)      ACE_ERROR_RETURN((LM_ERROR,                        ACE_TEXT ("%p/n"),                        ACE_TEXT ("recv_n")),                       0);    msgBufSize = ACE_NTOHL(msgBufSize);    // allocate the buffer for the message payload    ACE_CDR::Octet * msgBuf = 0;    ACE_NEW_RETURN(msgBuf,                   ACE_CDR::Octet[msgBufSize],                   0);    // read the buffer    if ((dataModeStream->recv_n(msgBuf, msgBufSize, 0, &bt)) == -1)      ACE_ERROR_RETURN((LM_ERROR,                        ACE_TEXT ("%p/n"),                        ACE_TEXT ("recv_n")),                       0);    // clean up the allocated buffer    delete[] msgBuf;    // SEND A REPLY TO THE CLIENT    // send back a 2 byte reply    ACE_CDR::Short reply;    if ((dataModeStream->send_n(&reply, ACE_CDR::SHORT_SIZE, 0, &bt)) == -1)      ACE_ERROR_RETURN((LM_ERROR,                        ACE_TEXT ("%p/n"),                        ACE_TEXT ("send_n")),                       0);  } while (--numIterations);  // close and destroy the stream  dataModeStream->close();  delete dataModeStream;  return 0;}

voidJAWS_Synch_IO::send_message (JAWS_IO_Handler *ioh,                             const char *buffer,                             unsigned int length){  ACE_SOCK_Stream stream;  stream.set_handle (ioh->handle ());  stream.send_n (buffer, length);}

intACE_TMAIN (int argc, ACE_TCHAR *argv[]){  ACE_LOG_MSG->open (argv[0]);  parse_args (argc, argv);  // Default is to ask the server for ``help.''  static char buf[BUFSIZ] = "help/n";  int n;  ACE_SOCK_Stream   sc;  ACE_SOCK_Connector con;  if (con.connect (sc,                   ACE_INET_Addr (port_number,                                  host_name)) == -1)    ACE_ERROR_RETURN ((LM_ERROR,                       "%p/n%a",                       "connect",                       1),                      -1);  if (remote_reconfigure)    // Remotely instruct the server to reconfigure itself.    ACE_OS::strcpy (buf, "reconfigure/n");  // Send the command.  if (sc.send_n (buf,                 ACE_OS::strlen (buf) + 1) == -1)    ACE_ERROR_RETURN ((LM_ERROR,                       "%p/n%a",                       "send",                       1), -1);  // Next, read the response.  while ((n = sc.recv (buf,                       sizeof buf)) > 0)    if (ACE_OS::write (ACE_STDOUT,                       buf,                       n) != n)      ACE_ERROR_RETURN ((LM_ERROR,                         "%p/n%a",                         "write",                         1),                        -1);  if (sc.close () == -1)      ACE_ERROR_RETURN ((LM_ERROR,                         "%p/n%a",                         "close",                         1),                        -1);  return 0;}

intACE_SOCK_Acceptor::accept (ACE_SOCK_Stream &new_stream,                           ACE_Addr *remote_addr,                           ACE_Time_Value *timeout,                           int restart,                           int reset_new_handle) const{  ACE_TRACE ("ACE_SOCK_Acceptor::accept");  int in_blocking_mode = 0;  if (this->shared_accept_start (timeout,                                 restart,                                 in_blocking_mode) == -1)    return -1;  else    {      // On Win32 the third parameter to <accept> must be a NULL      // pointer if we want to ignore the client's address.      int *len_ptr = 0;      sockaddr *addr = 0;      int len = 0;      if (remote_addr != 0)        {          len = remote_addr->get_size ();          len_ptr = &len;          addr = (sockaddr *) remote_addr->get_addr ();        }      do        new_stream.set_handle (ACE_OS::accept (this->get_handle (),                                               addr,                                               len_ptr));      while (new_stream.get_handle () == ACE_INVALID_HANDLE             && restart != 0             && errno == EINTR             && timeout == 0);      // Reset the size of the addr, so the proper UNIX/IPv4/IPv6 family      // is known.      if (new_stream.get_handle () != ACE_INVALID_HANDLE          && remote_addr != 0)        {          remote_addr->set_size (len);          if (addr)            remote_addr->set_type (addr->sa_family);        }    }  return this->shared_accept_finish (new_stream,                                     in_blocking_mode,                                     reset_new_handle);}

int YARPOutputSocketDgram::Close (const YARPUniqueNameID& name){	ACE_Time_Value timeout (YARP_SOCK_TIMEOUT, 0);	ACE_UNUSED_ARG (name);	OSDataDgram& d = OSDATA(system_resources);	YARP_DBG(THIS_DBG) ((LM_DEBUG, "Pretending to close a connection to port %d on %s/n", 		d._remote_addr.get_port_number(), 		d._remote_addr.get_host_addr()));	/// send the header.	MyMessageHeader hdr;	hdr.SetGood ();	hdr.SetLength (YARP_MAGIC_NUMBER + 1);	ACE_SOCK_Stream stream;	int r = d._service_socket.connect (stream, d._remote_addr, &timeout);	if (r < 0)	{		ACE_DEBUG ((LM_DEBUG, "cannot connect to remote peer %s:%d/n", d._remote_addr.get_host_addr(), d._remote_addr.get_port_number()));		ACE_DEBUG ((LM_DEBUG, "close will complete anyway/n"));		d._connector_socket.close ();		identifier = ACE_INVALID_HANDLE;		return YARP_FAIL;	}	r = stream.send_n (&hdr, sizeof(hdr), 0);	/// wait response.	/// need a timeout here!	hdr.SetBad ();	r = stream.recv (&hdr, sizeof(hdr), 0, &timeout);	if (r < 0)	{		stream.close ();		d._connector_socket.close ();		identifier = ACE_INVALID_HANDLE;		ACE_DEBUG ((LM_DEBUG, "cannot handshake with remote %s:%d/n", d._remote_addr.get_host_addr(), d._remote_addr.get_port_number()));		return YARP_FAIL;	}	hdr.GetLength();	d._remote_addr.set ((u_short)0);	d._remote_acceptor_store.set ((u_short)0);	stream.close ();	d._connector_socket.close ();	identifier = ACE_INVALID_HANDLE;	return YARP_OK;}

intACE_SOCK_Connector::complete (ACE_SOCK_Stream &new_stream,                              ACE_Addr *remote_sap,                              const ACE_Time_Value *tv){  ACE_TRACE ("ACE_SOCK_Connector::complete");  ACE_HANDLE h = ACE::handle_timed_complete (new_stream.get_handle (),                                             tv);  // We failed to get connected.  if (h == ACE_INVALID_HANDLE)    {#if defined (ACE_WIN32)      // Win32 has a timing problem - if you check to see if the      // connection has completed too fast, it will fail - so wait      // <ACE_NON_BLOCKING_BUG_DELAY> microseconds to let it catch up      // then retry to see if it's a real failure.      ACE_Time_Value time (0, ACE_NON_BLOCKING_BUG_DELAY);      ACE_OS::sleep (time);      h = ACE::handle_timed_complete (new_stream.get_handle (),                                      tv);      if (h == ACE_INVALID_HANDLE)        {#endif /* ACE_WIN32 */      // Save/restore errno.      ACE_Errno_Guard error (errno);      new_stream.close ();      return -1;#if defined (ACE_WIN32)        }#endif /* ACE_WIN32 */    }  if (remote_sap != 0)    {      int len = remote_sap->get_size ();      sockaddr *addr = reinterpret_cast<sockaddr *> (remote_sap->get_addr ());      if (ACE_OS::getpeername (h,                               addr,                               &len) == -1)        {          // Save/restore errno.          ACE_Errno_Guard error (errno);          new_stream.close ();          return -1;        }    }  // Start out with non-blocking disabled on the <new_stream>.  new_stream.disable (ACE_NONBLOCK);  return 0;}

intScavenger_Task::svc(void){  this->the_barrier_->wait ();  ACE_DEBUG ((LM_DEBUG, "(%P|%t) Starting scavenger thread/n"));  ACE_SOCK_Stream stream;  {    ACE_INET_Addr remote_sap (this->endpoint_);    ACE_SOCK_Connector connector;    if (connector.connect(stream, remote_sap) == -1)      {        ACE_ERROR((LM_ERROR, "Cannot connect to <%s>/n", endpoint_));        return -1;      }  }  for (;;)    {      ACE_Time_Value period (0, this->period_in_usecs_);      ACE_OS::sleep (period);      {        ACE_GUARD_RETURN (TAO_SYNCH_MUTEX, ace_mon, this->mutex_, -1);        if (this->stopped_)          break;      }      ACE_hrtime_t start = ACE_OS::gethrtime ();      ssize_t n = stream.send_n(&start, sizeof(start));      if (n == 0 || n == -1)        break;      ACE_hrtime_t end;      n = stream.recv(&end, sizeof(end));      if (n == 0 || n == -1)        break;      if (start != end)      {        ACE_ERROR((LM_ERROR,                   "Mismatched response from <%s>/n", endpoint_));        break;      }    }  stream.close();  ACE_DEBUG ((LM_DEBUG, "(%P|%t) Finishing scavenger thread/n"));  return 0;}

voidJAWS_Synch_IO::transmit_file (JAWS_IO_Handler *ioh,                              ACE_HANDLE handle,                              const char *header,                              unsigned int header_size,                              const char *trailer,                              unsigned int trailer_size){  int result = 0;  if (handle != ACE_INVALID_HANDLE)    {      ACE_SOCK_Stream stream;      stream.set_handle (ioh->handle ());      if ((unsigned long) stream.send_n (header, header_size) < header_size)        {          result = -1;        }      else        {          int count;          char buf[BUFSIZ];          do            {              count = ACE_OS::read (handle, buf, sizeof (buf));              if (count <= 0)                break;              if (stream.send_n (buf, count) < count)                {                result = -1;                }            }          while (result == 0);          if ((unsigned long) stream.send_n (trailer, trailer_size)              < trailer_size)            {            result = -1;            }        }    }  if (result == 0)    ioh->transmit_file_complete ();  else    ioh->transmit_file_error (result);}

intHTTP_Server::thread_per_request (HTTP_Handler_Factory &factory){  int grp_id = -1;  // thread per request  // Main thread opens the acceptor  if (this->acceptor_.open (ACE_INET_Addr (this->port_), 1,                            PF_INET, this->backlog_) == -1)    ACE_ERROR_RETURN ((LM_ERROR, ACE_TEXT ("%p/n"),                       ACE_TEXT ("HTTP_Acceptor::open")), -1);  ACE_SOCK_Stream stream;  // When we are throttling, this is the amount of time to wait before  // checking for runnability again.  const ACE_Time_Value wait_time (0, 10);  for (;;)    {      if (this->acceptor_.accept (stream) == -1)        ACE_ERROR_RETURN ((LM_ERROR, ACE_TEXT ("%p/n"),                           ACE_TEXT ("HTTP_Acceptor::accept")), -1);      Thread_Per_Request_Task *t;      // Pass grp_id as a constructor param instead of into open.      ACE_NEW_RETURN (t, Thread_Per_Request_Task (stream.get_handle (),                                                  this->tm_,                                                  grp_id,												  factory),                      -1);      if (t->open () != 0)	ACE_ERROR_RETURN ((LM_ERROR, ACE_TEXT ("%p/n"),                           ACE_TEXT ("Thread_Per_Request_Task::open")),                          -1);      // Throttling is not allowing too many threads to run away.      // Should really use some sort of condition variable here.      if (!this->throttle_)	continue;      // This works because each task has only one thread.      while (this->tm_.num_tasks_in_group (grp_id) > this->threads_)	this->tm_.wait (&wait_time);    }  ACE_NOTREACHED(return 0);}

template <class HANDLER> voidACE_Asynch_Acceptor<HANDLER>::parse_address (const                                             ACE_Asynch_Accept::Result &result,                                             ACE_INET_Addr &remote_address,                                             ACE_INET_Addr &local_address){  ACE_TRACE ("ACE_Asynch_Acceptor<>::parse_address");#if defined (ACE_HAS_AIO_CALLS)  // Use an ACE_SOCK to get the addresses - it knows how to deal with  // ACE_INET_Addr objects and get IPv4/v6 addresses.  ACE_SOCK_Stream str (result.accept_handle ());  str.get_local_addr (local_address);  str.get_remote_addr (remote_address);#elif defined (ACE_HAS_WINSOCK2) && (ACE_HAS_WINSOCK2 != 0)  ACE_Message_Block &message_block = result.message_block ();  sockaddr *local_addr = 0;  sockaddr *remote_addr = 0;  int local_size = 0;  int remote_size = 0;  // This matches setup in accept().  size_t addr_size = sizeof (sockaddr_in) + 16;#if defined (ACE_HAS_IPV6)  if (this->addr_family_ == PF_INET6)    addr_size = sizeof (sockaddr_in6) + 16;#endif /* ACE_HAS_IPV6 */  ::GetAcceptExSockaddrs (message_block.rd_ptr (),                          static_cast<DWORD> (this->bytes_to_read_),                          static_cast<DWORD> (addr_size),                          static_cast<DWORD> (addr_size),                          &local_addr,                          &local_size,                          &remote_addr,                          &remote_size);  local_address.set (reinterpret_cast<sockaddr_in *> (local_addr),                     local_size);  remote_address.set (reinterpret_cast<sockaddr_in *> (remote_addr),                      remote_size);#else  // just in case  errno = ENOTSUP;#endif /* defined (ACE_HAS_WINSOCK2) && (ACE_HAS_WINSOCK2 != 0) */  return;}

static ACE_THR_FUNC_RETURNworker (void *){  ACE_OS::sleep (3);  const ACE_TCHAR *msg = ACE_TEXT ("Message from Connection worker");  ACE_TCHAR buf [BUFSIZ];  buf[0] = static_cast<ACE_TCHAR> ((ACE_OS::strlen (msg) + 1));  ACE_OS::strcpy (&buf[1], msg);  ACE_INET_Addr addr (rendezvous);  ACE_DEBUG((LM_DEBUG,             "(%t) Spawning %d client threads.../n",             cli_thrno));  int grp = ACE_Thread_Manager::instance ()->spawn_n (cli_thrno,                                                      &cli_worker,                                                      buf);  ACE_TEST_ASSERT (grp != -1);  ACE_Thread_Manager::instance ()->wait_grp (grp);  ACE_DEBUG ((LM_DEBUG,              "(%t) Client threads done; shutting down.../n"));  ACE_SOCK_Stream stream;  ACE_SOCK_Connector connect;  if (connect.connect (stream, addr) == -1)    ACE_ERROR ((LM_ERROR,                ACE_TEXT ("(%t) %p Error while connecting/n"),                ACE_TEXT ("connect")));  const ACE_TCHAR *sbuf = ACE_TEXT ("/011shutdown");  ACE_DEBUG ((LM_DEBUG,              "shutdown stream handle = %x/n",              stream.get_handle ()));  if (stream.send_n (sbuf, (ACE_OS::strlen (sbuf) + 1) * sizeof (ACE_TCHAR)) == -1)    ACE_ERROR ((LM_ERROR,                ACE_TEXT ("(%t) %p/n"),                ACE_TEXT ("send_n")));  ACE_DEBUG ((LM_DEBUG,              "Sent message of length  = %d/n",              ACE_OS::strlen (sbuf)));  stream.close ();  return 0;}

////// pretend a connection.int YARPOutputSocketDgram::Connect (const YARPUniqueNameID& name, const YARPString& own_name){	ACE_UNUSED_ARG (name);	OSDataDgram& d = OSDATA(system_resources);	YARP_DBG(THIS_DBG) ((LM_DEBUG, "Pretending a connection to port %d on %s/n", 		d._remote_addr.get_port_number(), 		d._remote_addr.get_host_addr()));	ACE_Time_Value timeout (YARP_SOCK_TIMEOUT, 0);	ACE_SOCK_Stream stream;	int r = d._service_socket.connect (stream, d._remote_addr, &timeout);	if (r < 0)	{		ACE_DEBUG ((LM_ERROR, "cannot connect to remote peer %s:%d/n", d._remote_addr.get_host_addr(), d._remote_addr.get_port_number()));		return YARP_FAIL;	}	/// send the header.	int port_number = 0;	MyMessageHeader hdr;	hdr.SetGood ();	hdr.SetLength (YARP_MAGIC_NUMBER + 128*name.getRequireAck());	stream.send_n (&hdr, sizeof(hdr), 0);	/// fine, now send the name of the connection.	NetInt32 name_len = own_name.length();	stream.send_n (&name_len, sizeof(NetInt32), 0);	stream.send_n (own_name.c_str(), name_len, 0);	/// wait response.	hdr.SetBad ();	r = stream.recv (&hdr, sizeof(hdr), 0, &timeout);	if (r < 0)	{		stream.close ();		ACE_DEBUG ((LM_ERROR, "cannot handshake with remote %s:%d/n", d._remote_addr.get_host_addr(), d._remote_addr.get_port_number()));		return YARP_FAIL;	}	/// stores the remote acceptor address for future use (e.g. closing the connection).	d._remote_acceptor_store = d._remote_addr;	port_number = hdr.GetLength();	if (port_number == -1)	{		/// there might be a real -1 port number -> 65535.		stream.close ();		ACE_DEBUG ((LM_ERROR, "*** error, got garbage back from remote %s:%d/n", d._remote_addr.get_host_addr(), d._remote_addr.get_port_number()));		return YARP_FAIL;	}	/// the connect changes the remote port number to the actual assigned channel.	d._remote_addr.set_port_number (port_number);	stream.close ();	return YARP_OK;}

int Sender::sendMessage(const std::string& str, const ACE_SOCK_Stream& stream, bool quiet) {    ssize_t res = stream.send_n(str.c_str(), (int)str.size(), &timeout);    if(res != (ssize_t)str.size()) {        if(!quiet)            cerr << "Sender::sendMessage(): sending timeout!"<<endl;        return E_SEND;    }    res = stream.recv(response, 1, &timeout);    if(res <= 0) {        if(!quiet)            cerr << "Sender::sendMessage(): response timeout!" << endl;        return E_RESPONSE;    }    return SUCCESS;}