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

    virtual int svc ()     {  	     	   //int  i =0;  	   ACE_Message_Block *mblk;  	   	   int  len = 0;	          while(!fin.eof())      {      	       	 fin.getline(file_buf, LineSize);      	       	 len = ACE_OS::strlen( file_buf );         ACE_NEW_RETURN (mblk, ACE_Message_Block (len+200), 0);         if (file_buf[len-1] == '/r')         {         	len = len - 1;         }          	     mblk->copy (file_buf, len+1 ); 	     // 通过put_next函数，将消息传递给下一个过滤器	     put_next (mblk); 	  }       ACE_NEW_RETURN(mblk, ACE_Message_Block (0, ACE_Message_Block::MB_STOP),  0);	   put_next (mblk);	   fin.close();	   ACE_DEBUG ((LM_DEBUG, "read svc return ./n"));        return 0;    }

int UDPGenerator::readdatagram(int header_size){    ACE_Message_Block* msg = 0;/*{{{*/    ACE_NEW_RETURN (msg, ACE_Message_Block (1024), -1);    msg->size (header_size); // size of header to read is 20 bytes    // create a message block to read the message body    ACE_Message_Block* body = 0;    ACE_NEW_RETURN (body, ACE_Message_Block (1024), -1);    // The message body will not exceed 1024 bytes, at least not in this test.    msg->cont (body);    // ok lets do the asynch read    size_t number_of_bytes_recvd = 0;    int res = rd_.recv (                  msg,                  number_of_bytes_recvd,                  0,                  PF_INET,                  this->act_);    /*}}}*/    switch (res)/*{{{*/    {    case 0:        // this is a good error.  The proactor will call our handler when the        // read has completed.        break;    case 1:        // actually read something, we will handle it in the handler callback        ACE_DEBUG ((LM_DEBUG, "********************/n"));        ACE_DEBUG ((LM_DEBUG,                    "%s = %d/n",                    "bytes recieved immediately",                    number_of_bytes_recvd));        ACE_DEBUG ((LM_DEBUG, "********************/n"));        res = 0;        break;    case -1:        // Something else went wrong.        ACE_ERROR ((LM_ERROR,                    "%p/n",                    "ACE_Asynch_Read_Dgram::recv"));        // the handler will not get called in this case so lets clean up our msg        msg->release ();        break;    default:        // Something undocumented really went wrong.        ACE_ERROR ((LM_ERROR,                    "%p/n",                    "ACE_Asynch_Read_Dgram::recv"));        msg->release ();        break;    }/*}}}*/    return res;}

static void *supplier (void *){  ACE_UPIPE_Stream s_stream;  ACE_UPIPE_Addr c_addr (ACE_TEXT("pattern"));  ACE_Auto_Basic_Array_Ptr<char> mybuf (new char[size]);  for (int i = 0; i < size; i++)    mybuf[i] = 'a';  ACE_DEBUG ((LM_DEBUG,              "(%t) supplier starting connect thread/n"));  ACE_UPIPE_Connector con;  if (con.connect (s_stream, c_addr) == -1)    ACE_ERROR ((LM_ERROR,                "(%t) %p/n",                "ACE_UPIPE_Acceptor.connect failed"));  // Test asynchronicity (the "acausal principle" ;-)).  s_stream.enable (ACE_SIGIO);  ACE_Message_Block *mb_p;  for (int j = 0; j < iterations; j++)    {      ACE_NEW_RETURN (mb_p,                      ACE_Message_Block (size,                                         ACE_Message_Block::MB_DATA,                                         (ACE_Message_Block *) 0,                                         mybuf.get ()),                      0);      if (s_stream.send (mb_p) == -1)        ACE_ERROR_RETURN ((LM_ERROR,                           "(%t) %p/n",                           "send failed"),                          0);    }  ACE_NEW_RETURN (mb_p,                  ACE_Message_Block ((size_t) 0),                  0);  // Insert a 0-sized message block to signal the other side to shut  // down.  if (s_stream.send (mb_p) == -1)    ACE_ERROR_RETURN ((LM_ERROR,                       "(%t) %p/n",                       "send failed"),                          0);  s_stream.close ();  return 0;}

static void *producer (ACE_Message_Queue<ACE_MT_SYNCH> *msg_queue){  ACE_Read_Buffer rb (ACE_STDIN);  // Keep reading stdin, until we reach EOF.  for (;;)    {      // Allocate a new buffer.      char *buffer = rb.read ('/n');      ACE_Message_Block *mb = 0;      if (buffer == 0)        {          // Send a 0-sized shutdown message to the other thread and          // exit.          ACE_NEW_RETURN (mb, ACE_Message_Block ((size_t) 0), 0);          if (msg_queue->enqueue_tail (mb) == -1)            ACE_ERROR ((LM_ERROR, "(%t) %p/n", "put_next"));          break;        }      // Enqueue the message in priority order.      else        {          // Allocate a new message, but have it "borrow" its memory          // from the buffer.          ACE_NEW_RETURN (mb,                          ACE_Message_Block (rb.size (),                          ACE_Message_Block::MB_DATA,                          0,                          buffer),                          0);          mb->msg_priority (ACE_Utils::truncate_cast<unsigned long> (rb.size ()));          mb->wr_ptr (rb.size ());          ACE_DEBUG ((LM_DEBUG,                      "enqueueing message of size %d/n",                      mb->msg_priority ()));          // Enqueue in priority order.          if (msg_queue->enqueue_prio (mb) == -1)            ACE_ERROR ((LM_ERROR, "(%t) %p/n", "put_next"));        }    }  // Now read all the items out in priority order (i.e., ordered by  // the size of the lines!).  consumer (msg_queue);  return 0;}

template <ACE_SYNCH_DECL> intACE_Stream<ACE_SYNCH_USE>::control (ACE_IO_Cntl_Msg::ACE_IO_Cntl_Cmds cmd,                                    void *a){    ACE_TRACE ("ACE_Stream<ACE_SYNCH_USE>::control");    ACE_IO_Cntl_Msg ioc (cmd);    ACE_Message_Block *db;    // Try to create a data block that contains the user-supplied data.    ACE_NEW_RETURN (db,                    ACE_Message_Block (sizeof (int),                                       ACE_Message_Block::MB_IOCTL,                                       0,                                       (char *) a),                    -1);    // Try to create a control block <cb> that contains the control    // field and a pointer to the data block <db> in <cb>'s continuation    // field.    ACE_Message_Block *cb = 0;    ACE_NEW_RETURN (cb,                    ACE_Message_Block (sizeof ioc,                                       ACE_Message_Block::MB_IOCTL,                                       db,                                       (char *) &ioc),                    -1);    // @@ Michael: The old semantic assumed that cb returns == 0    //             if no memory was available. We will now return immediately    //             without release (errno is set to ENOMEM by the macro).    // If we can't allocate <cb> then we need to delete db and return    // -1.    if (cb == 0)    {        db->release ();        errno = ENOMEM;        return -1;    }    int result;    if (this->stream_head_->writer ()->put (cb) == -1)        result = -1;    else if (this->stream_head_->reader ()->getq (cb) == -1)        result = -1;    else        result = ((ACE_IO_Cntl_Msg *) cb->rd_ptr ())->rval ();    // This will also release db if it's reference count == 0.    cb->release ();    return result;}

/*static*/TAO_Queued_Data *TAO_Queued_Data::make_queued_data (ACE_Allocator *message_buffer_alloc,                                   ACE_Allocator *input_cdr_alloc,                                   ACE_Data_Block *db){  // Get a node for the queue..  TAO_Queued_Data *qd = 0;  if (message_buffer_alloc)    {      ACE_NEW_MALLOC_RETURN (qd,                             static_cast<TAO_Queued_Data *> (                               message_buffer_alloc->malloc (sizeof (TAO_Queued_Data))),                             TAO_Queued_Data (message_buffer_alloc),                             0);    }  else    {      // No allocator, so use the global pool!      ACE_NEW_RETURN (qd,                      TAO_Queued_Data,                      0);    }  // Providing an ACE_Data_Block indicates that the caller wants  // an aligned ACE_Message_Block added to the TAO_Queued_Data.  if (db != 0)    {      // If this allocation fails, the TAO_Queued_Data will be leaked.      if (input_cdr_alloc == 0)        ACE_NEW_RETURN (qd->msg_block_,                        ACE_Message_Block (db,                                           0,                                           input_cdr_alloc),                        0);      else        ACE_NEW_MALLOC_RETURN (qd->msg_block_,                               static_cast<ACE_Message_Block*> (                                 input_cdr_alloc->malloc (sizeof (ACE_Message_Block))),                               ACE_Message_Block (db,                                                  0,                                                  input_cdr_alloc),                               0);      ACE_CDR::mb_align (qd->msg_block_);    }  return qd;}

static void *producer (void *args){  ACE_Message_Queue<ACE_MT_SYNCH> *msg_queue =    reinterpret_cast<ACE_Message_Queue<ACE_MT_SYNCH> *> (args);  ACE_Message_Block *mb = 0;  for (const char *c = ACE_ALPHABET; *c != '/0'; c++)    {      ++message_count;      // Allocate a new message      ACE_NEW_RETURN (mb,                      ACE_Message_Block (1),                      0);      *mb->wr_ptr () = *c;      // Set the priority.      mb->msg_priority (message_count);      mb->wr_ptr (1);      // Enqueue in priority order.      if (msg_queue->enqueue_prio (mb) == -1)        ACE_ERROR_RETURN ((LM_ERROR,                           ACE_TEXT ("(%t) %p/n"),                           ACE_TEXT ("put_next")),                          0);    }  // Now send a 0-sized shutdown message to the other thread  ACE_NEW_RETURN (mb,                  ACE_Message_Block ((size_t) 0),                  0);  if (msg_queue->enqueue_tail (mb) == -1)    ACE_ERROR ((LM_ERROR,                ACE_TEXT ("(%t) %p/n"),                ACE_TEXT ("put_next")));  ++message_count;  // Now read all the items out in priority order (i.e., ordered by  // the size of the lines!).  consumer (msg_queue);  return 0;}

voidReliableSession::send_naks(DisjointSequence& received){  const std::vector<SequenceRange> ranges(received.missing_sequence_ranges());  CORBA::ULong size = ACE_Utils::truncate_cast<CORBA::ULong>(ranges.size());  size_t len = sizeof(this->remote_peer_)             + sizeof(size)             + size * 2 * sizeof(SequenceNumber);  ACE_Message_Block* data;  ACE_NEW(data, ACE_Message_Block(len));  Serializer serializer(data);  serializer << this->remote_peer_;  serializer << size;  for (std::vector<SequenceRange>::const_iterator iter = ranges.begin();       iter != ranges.end(); ++iter) {    serializer << iter->first;    serializer << iter->second;    if (OpenDDS::DCPS::DCPS_debug_level > 0) {      ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("(%P|%t) ReliableSession::send_naks")                            ACE_TEXT (" local %d remote %d [%q - %q]/n"),                            this->link_->local_peer(), remote_peer_,                            iter->first.getValue(), iter->second.getValue()));    }  }  // Send control sample to remote peer:  send_control(MULTICAST_NAK, data);}

voidJAWS_Asynch_IO::send_message (const char *buffer,                              int length,                              int act){  ACE_Message_Block *mb = 0;  ACE_NEW (mb, ACE_Message_Block (buffer, length));  if (mb == 0)    {      this->handler_->error_message_complete ();      return;    }  ACE_Asynch_Write_Stream aw;  if (aw.open (*this, this->handle_) == -1      || aw.write (*mb, length, (void *) static_cast<intptr_t> (act)) == -1)    {      mb->release ();      if (act == CONFORMATION)        this->handler_->confirmation_message_complete ();      else        this->handler_->error_message_complete ();    }}

template <class HANDLER> intACE_Asynch_Acceptor<HANDLER>::accept (size_t bytes_to_read, const void *act){  ACE_TRACE ("ACE_Asynch_Acceptor<>::accept");  ACE_Message_Block *message_block = 0;  // The space_needed calculation is drive by needs of Windows. POSIX doesn't  // need to extra 16 bytes, but it doesn't hurt.  size_t space_needed = sizeof (sockaddr_in) + 16;#if defined (ACE_HAS_IPV6)  if (PF_INET6 == this->addr_family_)    space_needed = sizeof (sockaddr_in6) + 16;#endif /* ACE_HAS_IPV6 */  space_needed = (2 * space_needed) + bytes_to_read;  // Create a new message block big enough for the addresses and data  ACE_NEW_RETURN (message_block,                  ACE_Message_Block (space_needed),                  -1);  // Initiate asynchronous accepts  if (this->asynch_accept_.accept (*message_block,                                   bytes_to_read,                                   ACE_INVALID_HANDLE,                                   act,                                   0,                                   ACE_SIGRTMIN,                                   this->addr_family_) == -1)    {      // Cleanup on error      message_block->release ();      return -1;    }  return 0;}

/// send a data frame.intTimeStamp_Protocol_Object::send_frame (ACE_Message_Block *frame,                                       TAO_AV_frame_info *){  ACE_DEBUG ((LM_DEBUG,              "TimeStamp_Protocol_Object::send_frame/n"));  ACE_Message_Block* timestamp;  ACE_NEW_RETURN (timestamp,                  ACE_Message_Block (BUFSIZ),                  -1);  ACE_hrtime_t now = ACE_OS::gethrtime ();  ACE_UINT64 usec = now;  ACE_UINT32 val_1 = ACE_CU64_TO_CU32 (usec);  ACE_DEBUG ((LM_DEBUG,              "Time Stamp %u usecs/n",              val_1));  ACE_OS_String::memcpy (timestamp->wr_ptr (), &now, sizeof (now));  timestamp->wr_ptr (sizeof (now));  frame->cont (timestamp);  ssize_t result = this->transport_->send (frame);  if (result < 0)    return result;  return 0;}

intSupplier::svc (void){  ACE_Message_Block *mb = 0;  // Send one message for each letter of the alphabet, then send an empty  // message to mark the end.  for (const char *c = ACE_ALPHABET; *c != '/0'; c++)    {      // Allocate a new message.      char d[2];      d[0] = *c;      d[1] = '/0';      ACE_NEW_RETURN (mb,                      ACE_Message_Block (2),                      -1);      ACE_OS::strcpy (mb->wr_ptr (), d);      mb->wr_ptr (2);      if (this->put_next (mb) == -1)        ACE_ERROR ((LM_ERROR, ACE_TEXT ("(%t) %p/n"),                    ACE_TEXT ("put_next")));    }  ACE_NEW_RETURN(mb, ACE_Message_Block, -1);  if (this->put_next (mb) == -1)    ACE_ERROR ((LM_ERROR, ACE_TEXT ("(%t) %p/n"), ACE_TEXT ("put_next")));  return 0;}

intMessage_Handler::svc (void){  for (int i = 0;; i++)    {      ACE_Message_Block *mb;      ACE_NEW_RETURN (mb,                      ACE_Message_Block (1),                      0);      mb->msg_priority (i);      ACE_OS::sleep (1);      // Note that this putq() call with automagically invoke the      // notify() hook of our ACE_Reactor_Notification_Strategy,      // thereby informing the <ACE_Reactor> Singleton to call our      // <handle_input> method.      if (this->putq (mb) == -1)        {          if (errno == ESHUTDOWN)            ACE_ERROR_RETURN ((LM_ERROR,                               "(%t) queue is deactivated"), 0);          else            ACE_ERROR_RETURN ((LM_ERROR,                               "(%t) %p/n",                               "putq"),                              -1);        }    }  ACE_NOTREACHED (return 0);}

int ACE_TMAIN (int, ACE_TCHAR *[]){  Manager tp;  tp.activate ();  // Wait for a moment every time you send a message.  ACE_Time_Value tv;  tv.msec (100);  ACE_Message_Block *mb = 0;  for (int i = 0; i < 30; i++)    {      ACE_NEW_RETURN        (mb, ACE_Message_Block(sizeof(int)), -1);      ACE_OS::memcpy (mb->wr_ptr (), &i, sizeof(int));      ACE_OS::sleep (tv);      // Add a new work item.      tp.putq (mb);    }  ACE_Thread_Manager::instance ()->wait ();  return 0;}

intACE_Activation_Queue::enqueue (ACE_Method_Request *mr,                               ACE_Time_Value *tv){  ACE_Message_Block *mb = 0;  // We pass sizeof (*mr) here so that flow control will work  // correctly.  Since we also pass <mr> note that no unnecessary  // memory is actually allocated -- just the size field is set.  ACE_NEW_MALLOC_RETURN (mb,                         static_cast<ACE_Message_Block *> (this->allocator_->malloc (sizeof (ACE_Message_Block))),                         ACE_Message_Block (sizeof (*mr),    // size                                            ACE_Message_Block::MB_DATA, // type                                            0,       // cont                                            (char *) mr,    // data                                            0,       // allocator                                            0,       // locking strategy                                            mr->priority (), // priority                                            ACE_Time_Value::zero,     // execution time                                            ACE_Time_Value::max_time, // absolute time of deadline                                            this->data_block_allocator_,  // data_block allocator                                            this->allocator_), // message_block allocator                         -1);  // Enqueue in priority order.  int const result = this->queue_->enqueue_prio (mb, tv);  // Free ACE_Message_Block if enqueue_prio failed.  if (result == -1)      ACE_DES_FREE (mb, this->allocator_->free, ACE_Message_Block);  return result;}

ACE_Message_Block* CMessageBlockManager::Create(uint32 u4Size){	ACE_Guard<ACE_Recursive_Thread_Mutex> WGuard(m_ThreadWriteLock);	ACE_Message_Block* pmb = NULL;	if(u4Size <= 0)	{		//如果申请的空间为0,则直接返回空。		return NULL;	}	ACE_NEW_MALLOC_NORETURN(pmb, 		                    static_cast<ACE_Message_Block*>(m_pmsgallocator->malloc(sizeof(ACE_Message_Block))),	 	                    ACE_Message_Block(u4Size, // size							ACE_Message_Block::MB_DATA, // type							0,							0,							m_pbuff_allocator, // allocator_strategy							0, // locking strategy							ACE_DEFAULT_MESSAGE_BLOCK_PRIORITY, // priority							ACE_Time_Value::zero,							ACE_Time_Value::max_time,							m_pdata_allocator,							m_pmsgallocator							));	return pmb;}

intReceiver::initiate_read_stream (void){  if (this->flg_cancel_ )     return -1;  if (this->get_ref_cnt_r() != 0)     return 0;  // don' start second read  u_int blksize = this->config().r_blksize();  u_int winsize = this->config().w_size();  // flow control  if ((u_long)(this->total_rcv_ - this->total_snd_) > winsize)    return 0;  ACE_Message_Block *mb = 0;  ACE_NEW_RETURN (mb,                  ACE_Message_Block (blksize+1),                  -1);  // Inititiate read  if (this->stream_.read (*mb, blksize) == -1)    {      mb->release ();      this->cancel_i();      return -1;    }  this->ref_cnt_r_++;  this->total_r_++;  return 0;}

voidMulticastSession::send_synack(){  size_t len = sizeof(this->remote_peer_);  ACE_Message_Block* data;  ACE_NEW(data, ACE_Message_Block(len));  Serializer serializer(data);  serializer << this->remote_peer_;  VDBG_LVL((LM_DEBUG, "(%P|%t) MulticastSession[%C]::send_synack "                      "local %#08x%08x remote %#08x%08x active %d/n",                      this->link()->config()->name().c_str(),                      (unsigned int)(this->link()->local_peer() >> 32),                      (unsigned int) this->link()->local_peer(),                      (unsigned int)(this->remote_peer_ >> 32),                      (unsigned int) this->remote_peer_,                      this->active_ ? 1 : 0), 2);  // Send control sample to remote peer:  send_control(MULTICAST_SYNACK, data);  // Send naks before sending synack to  // reduce wait time for resends from remote.  send_naks();}

ACE_Message_Block*DataLink::create_control(char submessage_id,                         DataSampleHeader& header,                         ACE_Message_Block* data){  DBG_ENTRY_LVL("DataLink", "create_control", 6);  header.byte_order_ = ACE_CDR_BYTE_ORDER;  header.message_id_ = TRANSPORT_CONTROL;  header.submessage_id_ = submessage_id;  header.message_length_ = static_cast<ACE_UINT32>(data->total_length());  ACE_Message_Block* message;  ACE_NEW_MALLOC_RETURN(message,                        static_cast<ACE_Message_Block*>(                          this->mb_allocator_->malloc(sizeof(ACE_Message_Block))),                        ACE_Message_Block(header.max_marshaled_size(),                                          ACE_Message_Block::MB_DATA,                                          data,                                          0,  // data                                          0,  // allocator_strategy                                          0,  // locking_strategy                                          ACE_DEFAULT_MESSAGE_BLOCK_PRIORITY,                                          ACE_Time_Value::zero,                                          ACE_Time_Value::max_time,                                          this->db_allocator_,                                          this->mb_allocator_),                        0);  *message << header;  return message;}

void HttpdPeer::handle_read_stream( const ACE_Asynch_Read_Stream::Result &result ){	if (!result.success() || result.bytes_transferred() == 0)	{		ACE_ERROR ((LM_ERROR,			"%p ",			"HttpdPeer::Read"));		ACE_OS::printf("%d/n",ACE_OS::last_error());		delete this;			}	else	{		//write response		if (connect_succeed_)		{			init_read();			return;		}		ACE_Message_Block *lpMb_ = NULL;		ACE_NEW_NORETURN(lpMb_,ACE_Message_Block(HTTP_RESPONSE,ACE_OS::strlen(HTTP_RESPONSE)));		lpMb_->wr_ptr(ACE_OS::strlen(HTTP_RESPONSE));		putQ(lpMb_ );		init_write();		init_read();		connect_succeed_ = true;		bIsIniting_ =true;		sentinel_ =0;	}}

voidJAWS_Asynch_IO::send_message (JAWS_IO_Handler *ioh,                              const char *buffer,                              unsigned int length,                              long act){  ioh->idle ();  JAWS_Asynch_IO_Handler *aioh =    dynamic_cast<JAWS_Asynch_IO_Handler *> (ioh);  ACE_Message_Block *mb;  ACE_NEW (mb, ACE_Message_Block (buffer, length));  if (mb == 0)    {      this->handler_->error_message_complete ();      return;    }  ACE_Asynch_Write_Stream aw;  if (aw.open (*(aioh->handler ()), aioh->handle ()) == -1      || aw.write (*mb, length, (void *) static_cast<intptr_t> (act)) == -1)    {      mb->release ();      if (act == CONFIRMATION)        ioh->confirmation_message_complete ();      else        ioh->error_message_complete ();    }}

ACE_Message_Block*OpenDDS::DCPS::RepoIdSetMap::marshal(bool byte_order){  DBG_ENTRY_LVL("RepoIdSetMap","marshal",6);  ACE_Message_Block* data = 0;  ACE_NEW_RETURN(data,                 ACE_Message_Block(this->marshaled_size(),                                   ACE_Message_Block::MB_DATA,                                   0, //cont                                   0, //data                                   0, //allocator_strategy                                   0, //locking_strategy                                   ACE_DEFAULT_MESSAGE_BLOCK_PRIORITY,                                   ACE_Time_Value::zero,                                   ACE_Time_Value::max_time,                                   0,                                   0),                 0);  Serializer writer(data, byte_order);  CORBA::ULong sz = this->size();  writer << sz;  for (MapType::iterator itr = map_.begin();       itr != map_.end();       ++itr) {    writer << itr->first;    itr->second->serialize(writer);  }  return data;}

TAO_BEGIN_VERSIONED_NAMESPACE_DECL/*! * @brief Allocate and return a new empty message block of size /a span_size * mimicking parameters of /a mb. * * This function allocates a new aligned message block using the same * allocators and flags as found in /a mb.  The size of the new message * block is at least /a span_size; the size may be adjusted up in order * to accomodate alignment requirements and still fit /a span_size bytes * into the aligned buffer. * * @param mb message block whose parameters should be mimicked * @param span_size size of the new message block (will be adjusted for proper * alignment) * @return an aligned message block with rd_ptr sitting at correct * alignment spot, 0 on failure */static ACE_Message_Block*clone_mb_nocopy_size (ACE_Message_Block *mb, size_t span_size){  // Calculate the required size of the cloned block with alignment  size_t const aligned_size = ACE_CDR::first_size (span_size + ACE_CDR::MAX_ALIGNMENT);  // Get the allocators  ACE_Allocator *data_allocator = 0;  ACE_Allocator *data_block_allocator = 0;  ACE_Allocator *message_block_allocator = 0;  mb->access_allocators (data_allocator,                         data_block_allocator,                         message_block_allocator);  // Create a new Message Block  ACE_Message_Block *nb = 0;  ACE_NEW_MALLOC_RETURN (nb,                         static_cast<ACE_Message_Block*> (                                         message_block_allocator->malloc (                                           sizeof (ACE_Message_Block))),                         ACE_Message_Block(aligned_size,                                           mb->msg_type(),                                           mb->cont(),                                           0, //we want the data block created                                           data_allocator,                                           mb->locking_strategy(),                                           mb->msg_priority(),                                           mb->msg_execution_time (),                                           mb->msg_deadline_time (),                                           data_block_allocator,                                           message_block_allocator),                         0);  ACE_CDR::mb_align (nb);  // Copy the flags over, but be SURE to clear the DONT_DELETE flag, since  // we just dynamically allocated the two things.  nb->set_flags (mb->flags());  nb->clr_flags (ACE_Message_Block::DONT_DELETE);  return nb;}

    int construct_message(void)  	{        // A very fast message creation algorithm// would lead to the need for queuing messages..// here. These messages are created and then sent// using the SLOW send_message() routine which is// running in a different thread so that the message//construction thread isn't blocked.            ACE_DEBUG((LM_DEBUG, "(%t)Constructing message::>> "));        // Create a new message to send            ACE_Message_Block * mb;        const char *data = "Hello Connector";        ACE_NEW_RETURN(mb, ACE_Message_Block(16,       //Message 16 bytes long                                              ACE_Message_Block::MB_DATA,       //Set header to data                                              0,        //No continuations.                                              data      //The data we want to send                        ), 0);        mb->wr_ptr(16);        //Set the write pointer.// Enqueue the message into the message queue// we COULD have done a timed wait for enqueuing in case// someone else holds the lock to the queue so it doesn't block//forever..        ACE_ASSERT(this->putq(mb) != -1);        ACE_DEBUG((LM_DEBUG, "Enqueued msg successfully/n"));    }

intSender::initiate_write (void){  if ( this->msg_queue ()->message_count () < 20) // flow control    {      size_t nbytes = ACE_OS::strlen (send_buf_);      ACE_Message_Block *mb = 0;      ACE_NEW_RETURN (mb,                      ACE_Message_Block (nbytes+8),                      -1);      mb->init (send_buf_, nbytes);      mb->rd_ptr (mb->base ());      mb->wr_ptr (mb->base ());      mb->wr_ptr (nbytes);      ACE_Time_Value tv = ACE_Time_Value::zero;      int qcount =this->putq (mb, & tv);      if (qcount <= 0)        {          ACE_Message_Block::release (mb);          return -1;        }    }  return initiate_io (ACE_Event_Handler::WRITE_MASK);}

int ACE_TMAIN (int, ACE_TCHAR *[]){  LF_ThreadPool tp;  tp.activate (THR_NEW_LWP| THR_JOINABLE, 5);  // Wait for a few seconds...  ACE_OS::sleep (2);  ACE_Time_Value tv (1L);  ACE_Message_Block *mb;  for (int i = 0; i < 30; i++)    {      ACE_NEW_RETURN (mb, ACE_Message_Block (sizeof(int)), -1);      ACE_OS::memcpy (mb->wr_ptr (), &i, sizeof(int));      ACE_OS::sleep (tv);      // Add a new work item.      tp.putq (mb);    }  ACE_Thread_Manager::instance ()->wait ();  ACE_OS::sleep (10);  return 0;}

int Sender::initiate_read_stream (void){  ACE_Guard<MyMutex> locker (m_Mtx) ;  // Create a new <Message_Block>.  Note that this message block will  // be used both to <read> data asynchronously from the socket and to  // <write> data asynchronously to the file.  ACE_DEBUG ((LM_DEBUG,              "initiate_read_stream called/n"));  ACE_Message_Block *mb = 0;  ACE_NEW_RETURN (mb,                  ACE_Message_Block (BUFSIZ + 1),                  -1);  // Inititiate read  if (this->rs_.read (*mb, mb->size ()- 1) == -1)    {      mb->release () ;      ACE_ERROR_RETURN ((LM_ERROR,                         "%p/n",                         "ACE_Asynch_Read_Stream::read"),                        -1);    }  nIOCount++ ;  return 0;}

int DC_Service_Handler::recv_data(ACE_Message_Block *&blk){	int ret;	size_t maxlen;	ACE_Time_Value tv(0);	if(blk == NULL)	{		ACE_NEW_RETURN(blk,ACE_Message_Block(512),-1);	}	else	{	}	maxlen = blk->size()-blk->length();	ret = KSGNetUtil::recv_buffer(peer().get_handle(),(unsigned char*)blk->wr_ptr(),maxlen,&tv);	if(ret)	{		return -1;	}	if(maxlen==0)		return 0;	blk->wr_ptr(maxlen);	// TODO: 判断包长度	return 1;}

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

int HDCCUSvrHandler::handle_input(ACE_HANDLE fd){    ACE_Message_Block * mb;    ACE_NEW_RETURN(mb,ACE_Message_Block(MAX_MESBUF_LEN),0);    // read data    ACE_INT32 n = 0;    ACE_INT32 m = 0;    while( (n = peer().recv(mb->wr_ptr(),mb->size() - m)) >= 0 )    {        mb->wr_ptr(n);    }    if(mb->length() <= 0)    {        mb->release();        return -1;    }    // 放入队列    if(putq(mb) == -1)    {        ACE_DEBUG((LM_ERROR,"保存失败"));        return -1;    }    _close_time = 1;    // 线程已经启动,返回 0 表示可以继续处理事件    REACTOR::instance()->remove_handler(this,                                        ACE_Event_Handler::READ_MASK|ACE_Event_Handler::DONT_CALL|                                        ACE_Event_Handler::WRITE_MASK);    activate(THR_NEW_LWP|THR_JOINABLE,1);    //ACE_Time_Value reschedule(_max_timeout_sec.sec()/2);    REACTOR::instance()->schedule_timer(this,NULL,_max_timeout_sec);    return 0;}