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

CmResult COFP13MPAggregateFlowStatsRequestMsg::StreamTo(ACE_OutputCDR &os){    if (COFP13MultipartMsg::StreamTo(os) != CM_OK)    {        return CM_ERROR_FAILURE;    }    os<<m_stats_request.table_id;    os.write_octet_array(m_stats_request.pad, sizeof(m_stats_request.pad));    os<<m_stats_request.out_port;    os<<m_stats_request.out_group;    os.write_octet_array(m_stats_request.pad2, sizeof(m_stats_request.pad2));        os<<m_stats_request.cookie;    os<<m_stats_request.cookie_mask;    bool bGood = os.good_bit();    CM_ASSERT_RETURN(bGood, CM_ERROR_FAILURE);    return m_match.StreamTo(os);}

void PubDriver::add_subscription (    CORBA::Long       reader_id,    const char *      sub_addr    ){  ::OpenDDS::DCPS::ReaderAssociationSeq associations;  associations.length (1);  associations[0].readerTransInfo.transport_id = 1; // TBD - not right  OpenDDS::DCPS::NetworkAddress network_order_address(sub_addr);  ACE_OutputCDR cdr;  cdr << network_order_address;  size_t len = cdr.total_length ();  associations[0].readerTransInfo.data    = OpenDDS::DCPS::TransportInterfaceBLOB    (len,    len,    (CORBA::Octet*)(cdr.buffer ()));  associations[0].readerId = reader_id;  associations[0].subQos = TheServiceParticipant->initial_SubscriberQos ();  associations[0].readerQos = TheServiceParticipant->initial_DataReaderQos ();  OpenDDS::DCPS::RepoId pub_id = foo_datawriter_servant_->get_publication_id();  datawriter_servant_->add_associations (pub_id, associations);}

	    static void serialize( ACE_OutputCDR& cdr, const LifeCycleFrame& frame ) {	       cdr.write_ushort( frame.endian_mark_ );		   cdr.write_ushort( frame.proto_version_ );		   cdr.write_ushort( frame.ctrl_);		   cdr.write_ushort( frame.hoffset_ );		   cdr.write_ulong( frame.command_ );	    }

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

voidtofSession_i::tof_debug( const CORBA::WChar * text, const CORBA::WChar * key ){    ACE_OutputCDR cdr;    cdr.write_wstring( text );    cdr.write_wstring( key );    ACE_Message_Block * mb = cdr.begin()->duplicate();    mb->msg_type( constants::MB_DEBUG );    pTask_->putq( mb );}

boolMulticastTransport::connection_info_i(TransportLocator& info) const{  NetworkAddress network_address(this->config_i_->group_address_);  ACE_OutputCDR cdr;  cdr << network_address;  const CORBA::ULong len = static_cast<CORBA::ULong>(cdr.total_length());  char* buffer = const_cast<char*>(cdr.buffer()); // safe  info.transport_type = "multicast";  info.data = TransportBLOB(len, len, reinterpret_cast<CORBA::Octet*>(buffer));  return true;}

// send the test header (only contains number of iterations)int sendHeader(ACE_SOCK_Stream & stream) {  // create an ACE CDR output stream and place the header information  // into it  ACE_OutputCDR hdrCDR;    hdrCDR << ACE_OutputCDR::from_boolean (ACE_CDR_BYTE_ORDER);    hdrCDR << ACE_CDR::ULong(Options_Manager::test_iterations+primerIterations);  if (!hdrCDR.good_bit())    return (0);  // send the header to the server (HEADER IS 8 BYTES LONG)  size_t bt;  if (stream.send_n(hdrCDR.begin(), 0,  &bt) == -1)    return 0;  return 1;}

ACE_CDR::BooleanACE_IBM1047_ISO8859::write_string (ACE_OutputCDR& out,                                   ACE_CDR::ULong len,                                   const ACE_CDR::Char* x){  if (out.write_ulong (len + 1))    return this->write_char_array (out, x, len + 1);  return 0;}

voidSubDriver::run(){  // Set up the publications.  OpenDDS::DCPS::AssociationData publications[1];  publications[0].remote_id_                = this->pub_id_;  publications[0].remote_data_.transport_id = 2; // TBD later - wrong  OpenDDS::DCPS::NetworkAddress network_order_address(this->pub_addr_str_);  ACE_OutputCDR cdr;  cdr << network_order_address;  size_t len = cdr.total_length ();  publications[0].remote_data_.data    = OpenDDS::DCPS::TransportInterfaceBLOB    (len,    len,    (CORBA::Octet*)(cdr.buffer ()));  // Write a file so that test script knows we're ready  FILE * file = ACE_OS::fopen ("subready.txt", "w");  ACE_OS::fprintf (file, "Ready/n");  ACE_OS::fclose (file);  this->subscriber_.init(ALL_TRAFFIC,                         this->sub_id_,                         1,               /* size of publications array */                         publications,                         this->num_msgs_);  // Wait until we receive our expected message from the remote  // publisher.  For this test, we should wait until we receive the  // "Hello World!" message that we expect.  Then this program  // can just shutdown.  while (this->subscriber_.received_test_message() == 0)    {      ACE_OS::sleep(1);    }  // Tear-down the entire Transport Framework.  TheTransportFactory->release();  TheServiceParticipant->shutdown();}

size_tUdpInst::populate_locator(OpenDDS::DCPS::TransportLocator& info) const{  if (this->local_address_ != ACE_INET_Addr()) {    NetworkAddress network_address(this->local_address_,                                   this->local_address_.is_any());    ACE_OutputCDR cdr;    cdr << network_address;    const CORBA::ULong len = static_cast<CORBA::ULong>(cdr.total_length());    char* buffer = const_cast<char*>(cdr.buffer()); // safe    info.transport_type = "udp";    info.data = TransportBLOB(len, len, reinterpret_cast<CORBA::Octet*>(buffer));    return 1;  } else {    return 0;  }}

/** * UDPGenerator::gloveDgramWrite * * @param remotehost * @param remoteport * @param glovedata * * @return */int UDPGenerator::gloveDgramWrite(    const ACE_TCHAR * remotehost,    u_short remoteport,    const DataGloveData &glovedata){    ACE_DEBUG ((LM_DEBUG, "Sender::initiate_write called/n"));    const size_t max_payload_size =        4 //boolean alignment flag        + 4 //payload length        + glovedata.length // Data Glove data length        + ACE_CDR::MAX_ALIGNMENT; //pading    // Rescuer header    u_short myid = htons(5);    u_short mysize = htons(max_payload_size);    ACE_Message_Block* rescuerheader= 0;    ACE_NEW_RETURN(rescuerheader, ACE_Message_Block(4), -1);    rescuerheader->copy((const char *)&myid, 2);    rescuerheader->copy((const char *)&mysize, 2);    // My DGS stuff (header and payload)    ACE_OutputCDR header (ACE_CDR::MAX_ALIGNMENT + 8);    header << ACE_OutputCDR::from_boolean (ACE_CDR_BYTE_ORDER);    header << ACE_CDR::ULong(length);    // DGS Payload    ACE_OutputCDR payload (max_payload_size);    payload << glovedata;    ACE_CDR::ULong length = payload.total_length();    iovec iov[3];    iov[0].iov_base = rescuerheader->rd_ptr();    iov[0].iov_len  = 4;    iov[1].iov_base = header.begin()->rd_ptr();    iov[1].iov_len  = HEADER_SIZE;    iov[2].iov_base = payload.begin()->rd_ptr();    iov[2].iov_len  = length;    ACE_INET_Addr serverAddr(remoteport, remotehost);    return sockDgram_.send(iov,3,  serverAddr );}

CmResult COFP10PacketOutMsg::StreamTo(ACE_OutputCDR &os){    CmResult lRet = CM_ERROR_FAILURE;        lRet = COFPMessage::StreamTo(os);    if (CM_FAILED(lRet))    {        ACE_ERROR((LM_ERROR, ACE_TEXT("COFP10PacketOutMsg::StreamTo, COFPMessage::StreamTo fail/n")));        return lRet;    }    os<<m_tMember.buffer_id;    os<<m_tMember.in_port;    os<<m_tMember.actions_len;        bool bGood = os.good_bit();    CM_ASSERT_RETURN(bGood, CM_ERROR_FAILURE);        std::list<COFP10Action *>::const_iterator it = m_action_list.begin();    while (it != m_action_list.end())    {        lRet = (*it)->StreamTo(os);        CM_ASSERT_RETURN(CM_SUCCEEDED(lRet), CM_ERROR_FAILURE);        it++;    }        if (m_PacketData)    {        ACE_DEBUG((LM_DEBUG,                    ACE_TEXT("COFP10PacketOutMsg::StreamTo, m_PacketData->total_length() = %u/n"),                   m_PacketData->total_length()));        std::string str = m_PacketData->flatten_chained();        os.write_char_array(str.c_str(), str.length());    }        bGood = os.good_bit();    CM_ASSERT_RETURN(bGood, CM_ERROR_FAILURE);    return CM_OK;}

boolTcpTransport::connection_info_i(TransportLocator& local_info) const{  DBG_ENTRY_LVL("TcpTransport", "connection_info_i", 6);  VDBG_LVL((LM_DEBUG, "(%P|%t) TcpTransport public address str %C/n",            this->tcp_config_->get_public_address().c_str()), 2);  // Get the public address string from the inst (usually the local address)  NetworkAddress network_order_address(this->tcp_config_->get_public_address());  ACE_OutputCDR cdr;  cdr << network_order_address;  const CORBA::ULong len = static_cast<CORBA::ULong>(cdr.total_length());  char* buffer = const_cast<char*>(cdr.buffer()); // safe  local_info.transport_type = "tcp";  local_info.data = TransportBLOB(len, len,                                  reinterpret_cast<CORBA::Octet*>(buffer));  return true;}

int DataWrapper::write(const ACE_OutputCDR& cdr){	//This method writes to an ace socket stream. First, it will collect	//the appropriate header which includes the message length. It will then	//combine ACE blocks that hold the header and message, and send it.	//If preparing the block or sending fails, it returns an appropriate error.	//If the actual socket send fails, the connection will be closed (not sure what	//else to do, since it shouldn't fail under a good connection)	if(!cdr.good_bit())		ACE_ERROR_RETURN((LM_ERROR, "%s, %p/n",toString(theAddress).c_str(),"Failed while sending CDR"), -1);	ACE_OutputCDR aceHead(headerLength);	//Put header info into a CDR	if(!(aceHead << (ACE_CDR::ULong)bitsForward) || !(aceHead << (ACE_CDR::ULong)cdr.length()))		ACE_ERROR_RETURN( (LM_ERROR, "%s, %p/n",toString(theAddress).c_str(),"Failed while sending, CDR header"), -1 );	//Create a block with the header and message	const_cast<ACE_Message_Block*>(aceHead.begin())->cont(const_cast<ACE_Message_Block*>(cdr.begin()));	const ssize_t msgLength = headerLength + ACE_CDR::total_length(cdr.begin(), cdr.end());	//	cout << toString(theAddress) << endl;	//Send the message	const ssize_t bytesSent = theStream.send_n(aceHead.begin());	// Clear the block	const_cast<ACE_Message_Block*>(aceHead.begin())->cont(NULL);	//Check whether the send succeeded (sent all the bytes). If not, we assume the socket has disconnected.	//	cout << "Bytes sent: " << bytesSent << " | msgLength: " << msgLength << endl;	if( bytesSent != msgLength ) {		theStream.close_writer();		ACE_ERROR_RETURN((LM_ERROR, "%s - %p/n",toString(theAddress).c_str(),"Socket failed while sending"), -1);	}	//Everything sent successfully	return 0;}

intoperator<< (ACE_OutputCDR &cdr,            const ACE_Log_Record &log_record){  // The written message length can't be more than 32 bits (ACE_CDR::ULong)  // so reduce it here if needed.  ACE_CDR::ULong u_msglen =    ACE_Utils::truncate_cast<ACE_CDR::ULong> (log_record.msg_data_len ());  // Insert each field from <log_record> into the output CDR stream.  cdr << ACE_CDR::Long (log_record.type ());  cdr << ACE_CDR::Long (log_record.pid ());  cdr << ACE_CDR::LongLong (log_record.time_stamp ().sec ());  cdr << ACE_CDR::Long (log_record.time_stamp ().usec ());  cdr << u_msglen;#if defined (ACE_USES_WCHAR)  cdr.write_wchar_array (log_record.msg_data (), u_msglen);#else  cdr.write_char_array (log_record.msg_data (), u_msglen);#endif /* ACE_USES_WCHAR */  return cdr.good_bit ();}

  //   // operator <<  //   bool operator << (ACE_OutputCDR & out, const ::OASIS::PointImpl & s)   {    if (!(out << s.x))      ACE_ERROR_RETURN ((LM_ERROR,                         ACE_TEXT ("%T (%t) - %M - failed to package x/n")),                        false);    if (!(out << s.y))      ACE_ERROR_RETURN ((LM_ERROR,                         ACE_TEXT ("%T (%t) - %M - failed to package y/n")),                        false);    return out.good_bit ();  }

  //FUZZ: disable check_for_lack_ACE_OS  int send (const ACE_Log_Record &log_record) {  //FUZZ: enable check_for_lack_ACE_OS    // Serialize the log record using a CDR stream, allocate    // enough space for the complete <ACE_Log_Record>.    const size_t max_payload_size =     4 // type()     + 8 // timestamp     + 4 // process id     + 4 // data length     + ACE_Log_Record::MAXLOGMSGLEN // data     + ACE_CDR::MAX_ALIGNMENT; // padding;    // Insert contents of <log_record> into payload stream.    ACE_OutputCDR payload (max_payload_size);    payload << log_record;    // Get the number of bytes used by the CDR stream.    ACE_CDR::ULong length = payload.total_length ();    // Send a header so the receiver can determine the byte    // order and size of the incoming CDR stream.    ACE_OutputCDR header (ACE_CDR::MAX_ALIGNMENT + 8);    header << ACE_OutputCDR::from_boolean (ACE_CDR_BYTE_ORDER);    // Store the size of the payload that follows    header << ACE_CDR::ULong (length);    // Use an iovec to send both buffer and payload simultaneously.    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;    // Send header and payload efficiently using "gather-write".    return logging_peer_.sendv_n (iov, 2);  }

ACE_CDR::BooleanUTF16_UCS2_Translator::write_wstring (ACE_OutputCDR & cdr,                                     ACE_CDR::ULong len,                                     const ACE_CDR::WChar *x){  // we'll accept a null pointer but only for an empty string  ACE_ASSERT (x != 0 || len == 0);  if (static_cast <ACE_CDR::Short> (this->major_version(cdr)) == 1      && static_cast <ACE_CDR::Short> (this->minor_version(cdr)) == 2)    {      if (len == 0) // for zero length strings, only write a length of                    // zero.        return this->write_4(cdr, &len);      ACE_CDR::ULong l = len * ACE_UTF16_CODEPOINT_SIZE;      if (this->write_4 (cdr, &l) && x != 0)        {          if (cdr.do_byte_swap())            {              return this->write_swapped_wchar_array_i (cdr, x, len);            }          else            {              return this->write_wchar_array_i (cdr, x, len);            }        }    }  else    {      // pre GIOP 1.2:  include null terminator in length      ACE_CDR::ULong l = len + 1;      if (this->write_4 (cdr, &l))        {          if (x != 0)            {              return this->write_wchar_array_i (cdr, x, len + 1);            }          else            {              ACE_UTF16_T s = 0;              return this->write_2 (cdr, &s);            }        }    }  return 0;}

intCDR_Test_Types::test_put (ACE_OutputCDR &cdr){  for (int i = 0; i < n; ++i)    {      if (cdr.write_octet (this->o) == 0)        ACE_ERROR_RETURN ((LM_ERROR,                           ACE_TEXT ("write_octet[%d] failed/n"),                           i),                          1);      if (cdr.write_short (this->s) == 0)        ACE_ERROR_RETURN ((LM_ERROR,                           ACE_TEXT ("write_short[%d] failed/n"),                           i),                          1);      if (cdr.write_octet (this->o) == 0)        ACE_ERROR_RETURN ((LM_ERROR,                           ACE_TEXT ("write_octet-2[%d] failed/n"),                           i),                          1);      if (cdr.write_long (this->l) == 0)        ACE_ERROR_RETURN ((LM_ERROR,                           ACE_TEXT ("write_long[%d] failed/n"),                           i),                          1);      if (cdr.write_long (this->l) == 0)        ACE_ERROR_RETURN ((LM_ERROR,                           ACE_TEXT ("write_long-2[%d] failed/n"),                           i),                          1);      if (cdr.write_string (this->str) == 0)        ACE_ERROR_RETURN ((LM_ERROR,                           ACE_TEXT ("write_string[%d] failed/n"),                           i),                          1);      if (cdr.write_wstring (this->wstr) == 0)        ACE_ERROR_RETURN ((LM_ERROR,                           ACE_TEXT ("write_wstring[%d] failed/n"),                           i),                          1);    }  return 0;}

CmResult COFPSwitchConfig::StreamTo(ACE_OutputCDR &os){    CmResult lRet = CM_ERROR_FAILURE;        lRet = COFPMessage::StreamTo(os);    if (CM_FAILED(lRet))    {        ACE_ERROR((LM_ERROR, ACE_TEXT("COFPSwitchConfig::StreamTo, COFPMessage::StreamTo fail/n")));        return lRet;    }    os<<m_wFlags;    os<<m_wMissSendLen;        bool bGood = os.good_bit();    CM_ASSERT_RETURN(bGood, CM_ERROR_FAILURE);    return CM_OK;}

CmResult COFP13PacketOutMsg::StreamTo(ACE_OutputCDR &os){    CmResult lRet = CM_ERROR_FAILURE;    ACE_CDR::Octet pad[6];        lRet = COFPMessage::StreamTo(os);    if (CM_FAILED(lRet))    {        ACE_ERROR((LM_ERROR, ACE_TEXT("COFP13PacketOutMsg::StreamTo, COFPMessage::StreamTo fail/n")));        return lRet;    }    ACE_DEBUG((LM_DEBUG, ACE_TEXT("COFP13PacketOutMsg::StreamTo():/n")));    ACE_DEBUG((LM_DEBUG, ACE_TEXT("msgLen=%u/n"), GetMessageLength()));        os<<m_buffer_id;    os<<m_in_port;    os<<m_actions_len;    os.write_octet_array(pad, sizeof(pad));        bool bGood = os.good_bit();    CM_ASSERT_RETURN(bGood, CM_ERROR_FAILURE);    ACE_DEBUG((LM_DEBUG, ACE_TEXT("m_actions_len=%u/n"), m_actions_len));    #if 0    while(!m_action_list.empty())    {        CCmComAutoPtr<COpenFlowProtAction> action;        action = m_action_list.front();        m_action_list.pop_front();        action->EncodeAction(os);    }#endif    ACE_UINT16 i = 0;    std::list<COFP13Action *>::const_iterator it = m_action_list.begin();    while(it != m_action_list.end())    {        lRet = (*it)->StreamTo(os);        CM_ASSERT_RETURN(CM_SUCCEEDED(lRet), CM_ERROR_FAILURE);        ACE_DEBUG((LM_DEBUG, ACE_TEXT("action[%u]'s length=%u/n"), i++, (*it)->GetActionLen()));        bGood = os.good_bit();        CM_ASSERT_RETURN(bGood, CM_ERROR_FAILURE);        it++;    }    // just for debugging    ACE_UINT32 length = sizeof(m_buffer_id)+sizeof(m_in_port)+sizeof(m_actions_len)+sizeof(m_pad);    ACE_DEBUG((LM_DEBUG, ACE_TEXT("length=%u/n"), length));    ACE_INT32 dataLen = GetStreamLen()-COFPMessage::GetStreamLen()-length-m_actions_len;    ACE_DEBUG((LM_DEBUG, ACE_TEXT("dataLen=%d/n"), dataLen));    if (m_PacketData)    {        ACE_DEBUG((LM_DEBUG, ACE_TEXT("m_PacketData->total_length() = %u/n"), m_PacketData->total_length()));        std::string str = m_PacketData->flatten_chained();        bGood = os.write_char_array(str.c_str(), str.length());        CM_ASSERT_RETURN(bGood, CM_ERROR_FAILURE);    }    return CM_OK;}

template<class T, class H> voidCDR_Test<T, H>::do_test (int total, int niter, int use_array,                         char* srcbuf, char* dstbuf,                         int src_offset, int dst_offset){  if (!use_array)    {      dst_offset = src_offset = 0;    }  ACE_DEBUG((LM_DEBUG,             ACE_TEXT( "Starting Test for %s: %d elements " )             ACE_TEXT( "%susing arrays./n" ),             H::name (),             total,             ((use_array) ? ACE_TEXT( "" ) : ACE_TEXT( "not " ))));  if (!use_array && (total % 4) != 0)    {      int lasttotal = total;      total -= (total % 4);      ACE_DEBUG((LM_DEBUG,                 ACE_TEXT( "Rounding from %d to %d elements./n" ),                 lasttotal,                 total));    }  char* src = ACE_ptr_align_binary(srcbuf, H::size ());  T* idata = reinterpret_cast<T*> (src);  idata += src_offset;  src = reinterpret_cast<char*> (idata);  {    int i;    for (i = 0; i < total; i++)      {        idata[i] = CDR_Test<T, H>::checkval (i);      }  }  ACE_DEBUG((LM_DEBUG,             ACE_TEXT( "Writing data.../n" )));  char* toread = 0;  {    ACE_TEST_ASSERT(use_array || total % 4 == 0);    double totalsecs = 0.0;    int n;    for (n = 0; n < niter; n++)      {        size_t size = H::size () * (dst_offset + total) +                      ACE_CDR::MAX_ALIGNMENT;        ACE_OutputCDR os (dstbuf, size);        // This is intrusive...        char* const end = os.begin ()->wr_ptr() + size;        do_seal (end);        double secs = 0.0;        if (use_array)          {            {              int i;              for (i = 0; i < dst_offset; i++)                {                  os << T(0);                }            }            if (n == 0)              {                ACE_DEBUG((LM_DEBUG,                           ACE_TEXT ("* src align = %d, dst align = %d/n"),                           tellalign (src),                           tellalign (os.begin ()->wr_ptr ())));              }            Crono crono;            crono.start ();            H::write_array (os, idata, total);            crono.stop ();            secs = crono.read_seconds ();          }        else          {            int i = 0;            for (; i < dst_offset; i++)              {                os << T(0);              }            i = 0;            Crono crono;            crono.start();            while (i < total)              {                os << idata[i++];//.........这里部分代码省略.........

voidSubDriver::run(){  DBG_ENTRY_LVL("SubDriver", "run", 6);  VDBG((LM_DEBUG, "(%P|%t) DBG:   "             "Initialize our SimpleSubscriber object./n"));  this->reader_.enable_transport(false /*reliable*/, false /*durable*/);  // Write a file so that test script knows we're ready  FILE * file = ACE_OS::fopen ("subready.txt", ACE_TEXT("w"));  ACE_OS::fprintf (file, "Ready/n");  ACE_OS::fclose (file);  VDBG((LM_DEBUG, "(%P|%t) DBG:   Create the 'publications'./n"));  // Set up the publication.  OpenDDS::DCPS::AssociationData publication;  publication.remote_id_ = this->pub_id_;  publication.remote_reliable_ = true;  publication.remote_data_.length(1);  if (shmem_) {    publication.remote_data_[0].transport_type = "shmem";    std::ofstream ofs("sub-pid.txt");    ofs << ACE_OS::getpid() << std::endl;    ofs.close();    for (ACE_stat filestat; -1 == ACE_OS::stat("pub-pid.txt", &filestat);         ACE_OS::sleep(1)) {/*empty loop*/}    std::ifstream ifs("pub-pid.txt");    std::string pid;    getline(ifs, pid);    std::string str = OpenDDS::DCPS::get_fully_qualified_hostname() +                      '/0' + "OpenDDS-" + pid + "-shmem1";    publication.remote_data_[0].data.length(static_cast<CORBA::ULong>(str.size()));    std::memcpy(publication.remote_data_[0].data.get_buffer(),                str.c_str(), str.size());  } else { // tcp    publication.remote_data_[0].transport_type = "tcp";    OpenDDS::DCPS::NetworkAddress network_order_address(      ACE_TEXT_ALWAYS_CHAR(this->pub_addr_str_.c_str()));    ACE_OutputCDR cdr;    cdr << network_order_address;    CORBA::ULong len = static_cast<CORBA::ULong>(cdr.total_length());    publication.remote_data_[0].data =      OpenDDS::DCPS::TransportBLOB(len, len, (CORBA::Octet*)(cdr.buffer()));  }  this->reader_.init(publication, this->num_msgs_);  VDBG((LM_DEBUG, "(%P|%t) DBG:   "             "Ask the SimpleSubscriber object if it has received what, "             "it expected.  If not, sleep for 1 second, and ask again./n"));  // Wait until we receive our expected message from the remote  // publisher.  For this test, we should wait until we receive the  // "Hello World!" message that we expect.  Then this program  // can just shutdown.  while (this->reader_.received_test_message() == 0) {    ACE_OS::sleep(1);  }  this->reader_.print_time();  if (shmem_) {    ACE_OS::unlink("sub-pid.txt");  }  VDBG((LM_DEBUG, "(%P|%t) DBG:   "             "The SimpleSubscriber object has received what it expected.  "             "Release TheTransportFactory - causing all TransportImpl "             "objects to be shutdown()./n"));  this->reader_.disassociate(this->pub_id_);  TheServiceParticipant->shutdown();  VDBG((LM_DEBUG, "(%P|%t) DBG:   "             "TheTransportFactory has finished releasing./n"));}

ssize_tACE_Log_Msg_IPC::log (ACE_Log_Record &log_record){    // Serialize the log record using a CDR stream, allocate enough    // space for the complete <ACE_Log_Record>.    size_t const max_payload_size =        4    // type        + 4  // pid        + 12 // timestamp        + 4  // process id        + 4  // data length#if defined (ACE_USES_WCHAR)        + (log_record.msg_data_len () * ACE_OutputCDR::wchar_maxbytes())  // message#else        + log_record.msg_data_len () // message#endif        + ACE_CDR::MAX_ALIGNMENT;     // padding;    // Insert contents of <log_record> into payload stream.    ACE_OutputCDR payload (max_payload_size);    payload << log_record;    // Get the number of bytes used by the CDR stream. If it becomes desireable    // to support payloads more than 4GB, this field will need to be changed    // to a 64-bit value.    ACE_CDR::ULong length =        ACE_Utils::truncate_cast<ACE_CDR::ULong> (payload.total_length ());    // Send a header so the receiver can determine the byte order and    // size of the incoming CDR stream.    ACE_OutputCDR header (ACE_CDR::MAX_ALIGNMENT + 8);    header << ACE_OutputCDR::from_boolean (ACE_CDR_BYTE_ORDER);    // Store the size of the payload that follows    header << ACE_CDR::ULong (length);    // Use an iovec to send both buffer and payload simultaneously.    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 defined (ACE_HAS_STREAM_PIPES)    // Use the <putpmsg> API if supported to ensure correct message    // queueing according to priority.    ACE_Str_Buf header_msg (static_cast<void *> (header.begin ()->rd_ptr ()),                            static_cast<int> (8));    ACE_Str_Buf payload_msg (static_cast<void *> (payload.begin ()->rd_ptr ()),                             static_cast<int> (length));    return this->message_queue_.send (&header_msg,                                      &payload_msg,                                      static_cast<int> (log_record.priority ()),                                      MSG_BAND);#else    // We're running over sockets, so send header and payload    // efficiently using "gather-write".    return this->message_queue_.sendv_n (iov, 2);#endif /* ACE_HAS_STREAM_PIPES */}

// Logic of StreamTo is similar to StreamFrom.CmResult COFP13HelloMsg::StreamTo(ACE_OutputCDR &os){    bool       bGood = false;    ACE_UINT16 i     = 0;    CmResult   lRet  = CM_ERROR_FAILURE;        lRet = COFPMessage::StreamTo(os);    if (CM_FAILED(lRet))    {        ACE_ERROR((LM_ERROR, ACE_TEXT("COFP13HelloMsg::StreamTo, COFPMessage::StreamTo fail/n")));        return lRet;    }    if (m_wLength < COFPMessage::GetStreamLen())    {        ACE_ERROR((LM_ERROR,                    ACE_TEXT("COFP13HelloMsg::StreamTo(): m_wLength(%u) is less than OFP_MESSAGE_HDR_SIZE(%u)./n"),                   m_wLength,                   COFPMessage::GetStreamLen()));        return CM_ERROR_FAILURE;    }    ACE_UINT16             data_size = m_wElemListLen; //m_wLength-OPENFLOW_MESSAGE_HDR_SIZE;    ACE_UINT16             iter_size = 0;    const ACE_CDR::Octet  *pElem     = m_pElemList;    while (iter_size < m_wElemListLen)    {        ACE_UINT16 elem_type = GetHelloElemType(pElem);        ACE_UINT16 elem_size = GetHelloElemLen(pElem);  // elem size, excluding paddings                os<<elem_type;        os<<elem_size;        bGood = os.good_bit();        CM_ASSERT_RETURN(bGood, CM_ERROR_FAILURE);                ACE_DEBUG((LM_DEBUG, ACE_TEXT("COFP13HelloMsg::StreamTo(): encode hello elem %u/n"), i++));        ACE_DEBUG((LM_DEBUG, ACE_TEXT("elem_type=%u, elem_size=%u/n"), elem_type, elem_size));        if (elem_size < 4)        {            ACE_ERROR((LM_ERROR, ACE_TEXT("COFP13HelloMsg::StreamTo(): elem_size(%u) error, less than 4."), elem_size));            return CM_ERROR_FAILURE;        }        if ((elem_size + 7)/8*8 > data_size)        {            ACE_ERROR((LM_ERROR,                        ACE_TEXT("COFP13HelloMsg::StreamTo(): (elem_size + 7)/8*8=%u, greater than data_size(%u)/n"),                        (elem_size + 7)/8*8,                        data_size));            return CM_ERROR_FAILURE;        }        if (elem_type != OFPHET_VERSIONBITMAP)        {            ACE_ERROR((LM_ERROR,                        ACE_TEXT("COFP13HelloMsg::StreamTo(): elem_type(%u) isnot OFPHET_VERSIONBITMAP(%u)./n"),                       elem_type,                       OFPHET_VERSIONBITMAP));            return CM_ERROR_FAILURE;        }        ACE_UINT16 bitmap_size      = elem_size-4;        CM_ASSERT_RETURN((bitmap_size/4*4 == bitmap_size), CM_ERROR_INVALID_ARG);                ACE_UINT16 iter_bitmap_size = 0;        ACE_UINT16 bitmap_index     = 0;        while (iter_bitmap_size < bitmap_size)//(bitmap_size > 0)        {            ACE_UINT32 bitmap = GetVerBMHelloElemBitmap(pElem, bitmap_index);            os<<bitmap;            bGood = os.good_bit();            CM_ASSERT_RETURN(bGood, CM_ERROR_FAILURE);            ACE_DEBUG((LM_DEBUG, ACE_TEXT("COFP13HelloMsg::StreamTo, bitmap[%u]=%u/n"), bitmap_index++, bitmap));            //bitmap_size -= sizeof(bitmap);            iter_bitmap_size += sizeof(bitmap);        }        // deal with paddings        ACE_UINT16 pad_length = (elem_size + 7)/8*8 - elem_size;        bGood = os.write_octet_array(pElem + 4 + sizeof(ACE_UINT32) * bitmap_index, pad_length);        CM_ASSERT_RETURN(bGood, CM_ERROR_FAILURE);        iter_size += (elem_size + pad_length);        data_size -= (elem_size + pad_length);        pElem     += (elem_size + pad_length);    }    return CM_OK;}

intACE_TMAIN (int argc, ACE_TCHAR *argv[]){  const ACE_TCHAR *logger_host = argc > 1 ? argv[1] : LOGGER_HOST;  u_short logger_port  = argc > 2 ? ACE_OS::atoi (argv[2]) : LOGGER_PORT;  int max_iterations = argc > 3 ? ACE_OS::atoi (argv[3]) : MAX_ITERATIONS;  ACE_SOCK_Stream logger;  ACE_SOCK_Connector connector;  ACE_INET_Addr addr (logger_port, logger_host);  if (connector.connect (logger, addr) == -1)    ACE_ERROR_RETURN ((LM_ERROR, ACE_TEXT ("%p/n"), ACE_TEXT ("open")), -1);  for (int i = 0; i < max_iterations; i++)    {      ACE_Log_Record log_record (LM_DEBUG,                                 ACE_OS::time ((time_t *) 0),                                 ACE_OS::getpid ());      ACE_TCHAR buf[BUFSIZ];      ACE_OS::sprintf (buf, ACE_TEXT ("message = %d/n"), i + 1);      log_record.msg_data (buf);      const size_t max_payload_size =        4 // type()        + 8 // timestamp        + 4 // process id        + 4 // data length        + ACE_Log_Record::MAXLOGMSGLEN // data        + ACE_CDR::MAX_ALIGNMENT; // padding;      // Insert contents of <log_record> into payload stream.      ACE_OutputCDR payload (max_payload_size);      payload << log_record;      // Get the number of bytes used by the CDR stream.      ACE_CDR::ULong length =        ACE_Utils::truncate_cast<ACE_CDR::ULong> (payload.total_length ());      // Send a header so the receiver can determine the byte order and      // size of the incoming CDR stream.      ACE_OutputCDR header (ACE_CDR::MAX_ALIGNMENT + 8);      header << ACE_OutputCDR::from_boolean (ACE_CDR_BYTE_ORDER);      // Store the size of the payload that follows      header << ACE_CDR::ULong (length);      // Use an iovec to send both buffer and payload simultaneously.      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);    }  if (logger.close () == -1)    ACE_ERROR_RETURN ((LM_ERROR, ACE_TEXT ("%p/n"), ACE_TEXT ("close")), -1);  return 0;}