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

/***************************************************************************** Function:    handleMediaFrame Description: 拼完一帧PS数据后处理媒体数据 Input:       rtpFrameList PS数据包列表 Output:       Return:      N/A*****************************************************************************/void CPs2EsProcessor::handleMediaFrame(RTP_FRAME_LIST_T &rtpFrameList){	if(NULL == m_pExtendHeader )	{		ERROR_LOG("m_pExtendHeader is NULL");		return;	}	if(NULL == m_pRtpFrameCache)	{		ERROR_LOG("m_pRtpFrameCache is NULL");		return;	}    if (rtpFrameList.empty())    {        ERROR_LOG("Handle PS media frame abnormal , the frame list is empty");        return;    }    if (MAX_RTP_PACKET_COUNT < rtpFrameList.size())    {        ERROR_LOG("Handle PS media frame abnormal , the frame list exceeds the Threshold[1024], the rtp packet count: %d",rtpFrameList.size());        return;    }    // 新的一帧到达，缓存应该是空的    if (m_pWritePos != m_pRtpFrameCache)    {        m_pWritePos = m_pRtpFrameCache;        //BP_RUN_LOG_INF("Handle PS media frame abnormal",            "Write postion not in cache head, serviceID=%s.", m_strServiceID.c_str());    }    // 将收到的一帧PS数据拷贝至缓冲区    ACE_Message_Block* pRtpBlock = NULL;    bool bFirst = true;    CRtpPacket rtpPacket;    unsigned int unCacheSize = RTP_FRAME_CACHE_SIZE;    int iRet = IVS_SUCCEED;    for (RTP_FRAME_LIST_T_ITER iter = rtpFrameList.begin(); iter != rtpFrameList.end(); ++iter)    {        pRtpBlock = *iter;        iRet = rtpPacket.ParsePacket(pRtpBlock->rd_ptr(), pRtpBlock->length());        if (IVS_SUCCEED != iRet)        {            m_pWritePos = m_pRtpFrameCache;            m_pExtendHeader->reset();            ERROR_LOG("Parse rtp packet fail ,retcode:%d",iRet);            return;        }        if (bFirst)        {            if (1 == rtpPacket.GetExtension())            {                if(NULL != rtpPacket.GetMuExtData())                {                    m_pExtendHeader->copy((char *)rtpPacket.GetMuExtData(), sizeof(RTP_EXTENSION_DATA_MU_S));                }                else if(NULL != rtpPacket.GetExtData())                {                    m_pExtendHeader->copy((char *)rtpPacket.GetExtData(), sizeof(RTP_EXTENSION_DATA_S));                }                else                {                   ERROR_LOG("Error extension label");                }            }            if (m_bAppendExtInfo)            {                REAL_RECORD_TIME* realRecordTime = (REAL_RECORD_TIME*)(pRtpBlock->base());                if (NULL != realRecordTime)                {                    m_uiRealRecordSecond = realRecordTime->uiSecond;                    m_uiRealRecordMSecond = realRecordTime->uiMSecond;                    uint32_t* pStreamRate = (uint32_t*)(pRtpBlock->base() + sizeof(REAL_RECORD_TIME));                    m_uiReserved = *pStreamRate;                }                else                {                    ERROR_LOG("Error real record time info");                }            }            bFirst = false;        }        // 移除RTP消息头        pRtpBlock->rd_ptr(rtpPacket.GetHeadLen());        if (unCacheSize >= pRtpBlock->length())        {            memcpy(m_pWritePos, pRtpBlock->rd_ptr(), pRtpBlock->length());            m_pWritePos += pRtpBlock->length();//.........这里部分代码省略.........

int WorldSocket::SendPacket(WorldPacket const& pct){    ACE_GUARD_RETURN (LockType, Guard, m_OutBufferLock, -1);    if (closing_)        return -1;    // Dump outgoing packet    if (sPacketLog->CanLogPacket())        sPacketLog->LogPacket(pct, SERVER_TO_CLIENT);    WorldPacket const* pkt = &pct;    // Empty buffer used in case packet should be compressed    // Disable compression for now :)    /* WorldPacket buff;     if (m_Session && pkt->size() > 0x400)     {         buff.Compress(m_Session->GetCompressionStream(), pkt);         pkt = &buff;     }*/    uint16 opcodeNumber = serverOpcodeTable[pkt->GetOpcode()]->OpcodeNumber;    if (m_Session)        TC_LOG_TRACE("network.opcode", "S->C: %s %s", m_Session->GetPlayerInfo().c_str(), GetOpcodeNameForLogging(pkt->GetOpcode(), true).c_str());    sScriptMgr->OnPacketSend(this, *pkt);    ServerPktHeader header(!m_Crypt.IsInitialized() ? pkt->size() + 2 : pct.size(), opcodeNumber, &m_Crypt);    if (m_OutBuffer->space() >= pkt->size() + header.getHeaderLength() && msg_queue()->is_empty())    {        // Put the packet on the buffer.        if (m_OutBuffer->copy((char*) header.header, header.getHeaderLength()) == -1)            ACE_ASSERT (false);        if (!pkt->empty())            if (m_OutBuffer->copy((char*) pkt->contents(), pkt->size()) == -1)                ACE_ASSERT (false);    }    else    {        // Enqueue the packet.        ACE_Message_Block* mb;        ACE_NEW_RETURN(mb, ACE_Message_Block(pkt->size() + header.getHeaderLength()), -1);        mb->copy((char*) header.header, header.getHeaderLength());        if (!pkt->empty())            mb->copy((const char*)pkt->contents(), pkt->size());        if (msg_queue()->enqueue_tail(mb, (ACE_Time_Value*)&ACE_Time_Value::zero) == -1)        {            TC_LOG_ERROR("network", "WorldSocket::SendPacket enqueue_tail failed");            mb->release();            return -1;        }    }    return 0;}

intPeer_Handler::transmit_stdin (void){  // If return value is -1, then first_time_ must be reset to 1.  int result = 0;  if (this->connection_id_ != -1)    {      ACE_Message_Block *mb;      ACE_NEW_RETURN (mb,                      ACE_Message_Block (sizeof (Event)),                      -1);      // Cast the message block payload into an <Event> pointer.      Event *event = (Event *) mb->rd_ptr ();      ssize_t n = ACE_OS::read (ACE_STDIN,                                event->data_,                                sizeof event->data_);      switch (n)        {        case 0:          ACE_DEBUG ((LM_DEBUG,                      ACE_TEXT ("stdin closing down/n")));          // Take stdin out of the ACE_Reactor so we stop trying to          // send events.          ACE_Reactor::instance ()->remove_handler            (ACE_STDIN,             ACE_Event_Handler::DONT_CALL | ACE_Event_Handler::READ_MASK);          mb->release ();          result = 0; //          break;          /* NOTREACHED */        case -1:          mb->release ();          ACE_ERROR ((LM_ERROR,                      ACE_TEXT ("%p/n"),                      ACE_TEXT ("read")));          result = 0; //          break;          /* NOTREACHED */        default:          // Do not return directly, save the return value.          result = this->transmit (mb, n, ROUTING_EVENT);          break;          /* NOTREACHED */        }      // Do not return at here, but at exit of function.      /*return 0;*/    }  else  {  ACE_DEBUG ((LM_DEBUG,              ACE_TEXT ("Must transmit over an opened channel./n")));    result = -1; // Save return value at here, return at exit of function.  }  // If transmit error, the stdin-thread will be cancelled, so should  // reset first_time_ to 1, which will register_stdin_handler again.  if (result == -1)    first_time_ = 1;  return result;}

intSender::handle_output (ACE_HANDLE h){  ACE_GUARD_RETURN (ACE_Recursive_Thread_Mutex, locker, this->mutex_, -1);  ACE_Time_Value tv = ACE_Time_Value::zero;  ACE_Message_Block *mb = 0;  int     err=0;  ssize_t res=0;  size_t  bytes=0;  int     qcount = this->getq (mb , & tv);  if (mb != 0)  // qcount >= 0    {      bytes = mb->length ();      res = this->peer ().send (mb->rd_ptr (), bytes);      this->total_w_++;      if (res < 0)        err = errno ;      else        this->total_snd_ += res;      if (loglevel == 0 || res <= 0 || err!= 0)        {          LogLocker log_lock;          ACE_DEBUG ((LM_DEBUG, "**** Sender::handle_output () SessionId=%d****/n", index_));          ACE_DEBUG ((LM_DEBUG, "%C = %d/n", "bytes_to_write", bytes));          ACE_DEBUG ((LM_DEBUG, "%C = %d/n", "handle", h));          ACE_DEBUG ((LM_DEBUG, "%C = %d/n", "bytes_transferred", res));          ACE_DEBUG ((LM_DEBUG, "%C = %d/n", "error", err));          ACE_DEBUG ((LM_DEBUG, "%C = %s/n", "message_block", mb->rd_ptr ()));          ACE_DEBUG ((LM_DEBUG, "**** end of message ****************/n"));        }    }  ACE_Message_Block::release (mb);  if (err != 0  || res < 0)    return -1;  int rc = 0;  if (qcount <= 0)  // no more message blocks in queue    {      if (duplex != 0 &&   // full duplex, continue write          (this->total_snd_ - this->total_rcv_ ) < 1024*32 )  // flow control        rc = initiate_write ();      else        rc = terminate_io (ACE_Event_Handler::WRITE_MASK);      if (rc == -1)        return -1;    }  rc = initiate_io (ACE_Event_Handler::READ_MASK);  if (rc == -1)    return -1;  return check_destroy ();}

intConsumer_Handler::handle_output (ACE_HANDLE){  ACE_Message_Block *event = 0;  ACE_DEBUG ((LM_DEBUG,              ACE_TEXT("(%t) Receiver signalled 'resume transmission' %d/n"),              this->get_handle ()));  // WIN32 Notes: When the receiver blocked, we started adding to the  // consumer handler's message Q. At this time, we registered a  // callback with the reactor to tell us when the TCP layer signalled  // that we could continue to send messages to the consumer. However,  // Winsock only sends this notification ONCE, so we have to assume  // at the application level, that we can continue to send until we  // get any subsequent blocking signals from the receiver's buffer.#if defined (ACE_WIN32)  // Win32 Winsock doesn't trigger multiple "You can write now"  // signals, so we have to assume that we can continue to write until  // we get another EWOULDBLOCK.  // We cancel the wakeup callback we set earlier.  if (ACE_Reactor::instance ()->cancel_wakeup      (this, ACE_Event_Handler::WRITE_MASK) == -1)    ACE_ERROR_RETURN ((LM_ERROR,                       ACE_TEXT ("(%t) %p/n"),                       ACE_TEXT ("Error in ACE_Reactor::cancel_wakeup()")),                      -1);  // The list had better not be empty, otherwise there's a bug!  while (this->msg_queue ()->dequeue_head         (event, (ACE_Time_Value *) &ACE_Time_Value::zero) != -1)    {      switch (this->nonblk_put (event))        {        case -1:                // Error sending message to consumer.          {            // We are responsible for releasing an ACE_Message_Block if            // failures occur.            event->release ();            ACE_ERROR ((LM_ERROR,                        ACE_TEXT ("(%t) %p/n"),                        ACE_TEXT ("transmission failure")));            break;          }        case 0:                 // Partial Send - we got flow controlled by the receiver          {            ACE_DEBUG ((LM_DEBUG,                        ACE_TEXT ("%D Partial Send due to flow control")                        ACE_TEXT ("- scheduling new wakeup with reactor/n")));            // Re-schedule a wakeup call from the reactor when the            // flow control conditions abate.            if (ACE_Reactor::instance ()->schedule_wakeup                (this,                 ACE_Event_Handler::WRITE_MASK) == -1)              ACE_ERROR_RETURN ((LM_ERROR,                                 ACE_TEXT ("(%t) %p/n"),                                 ACE_TEXT ("Error in ACE_Reactor::schedule_wakeup()")),                                -1);            // Didn't write everything this time, come back later...            return 0;          }        default:                // Sent the whole thing          {            ACE_DEBUG ((LM_DEBUG,                        ACE_TEXT ("Sent message from message Q, Q size = %d/n"),                        this->msg_queue()->message_count ()));            break;          }        }    }  // If we drop out of the while loop, then the message Q should be  // empty...or there's a problem in the dequeue_head() call...but  // thats another story.  ACE_DEBUG ((LM_DEBUG,              ACE_TEXT ("%D Sent all messages from consumers message Q/n")));  ACE_DEBUG ((LM_DEBUG,              ACE_TEXT ("(%t) queueing deactivated on handle %d to routing id %d/n"),              this->get_handle (),              this->connection_id ()));#else /* !defined (ACE_WIN32) */  // The list had better not be empty, otherwise there's a bug!  if (this->msg_queue ()->dequeue_head      (event, (ACE_Time_Value *) &ACE_Time_Value::zero) != -1)    {      switch (this->nonblk_put (event))        {        case 0:           // Partial send.          ACE_ASSERT (errno == EWOULDBLOCK);          ACE_DEBUG ((LM_DEBUG,                      ACE_TEXT ("%D Partial Send/n")));          // Didn't write everything this time, come back later...          break;//.........这里部分代码省略.........