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

void
ReliableSession::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);
}

void
JAWS_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> int
ACE_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.
int
TimeStamp_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;
}

int
Supplier::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;
}

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

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

int
Receiver::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;
}

void
MulticastSession::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;
	}
}