rtcp.cpp

这是广泛使用的通信开源项目,对于大容量,高并发的通讯要求完全能够胜任,他广泛可用于网络游戏医学图像网关的高qos要求.更详细的内容可阅读相应的材料

}

TAO_AV_RTCP_Object::TAO_AV_RTCP_Object (TAO_AV_Callback *client_cb,
                                        TAO_AV_RTCP_Callback *&rtcp_cb,
                                        TAO_AV_Transport *transport)
  :TAO_AV_Protocol_Object (&rtcp_cb_, transport)
{
  rtcp_cb = &this->rtcp_cb_;
  this->client_cb_ = client_cb;

}

TAO_AV_RTCP_Object::~TAO_AV_RTCP_Object (void)
{
}

int
TAO_AV_RTCP_Object::destroy (void)
{
  this->callback_->handle_destroy ();
  delete this;

  return 0;
}

int
TAO_AV_RTCP_Object::set_policies (const TAO_AV_PolicyList &/*policy_list*/)
{
  return -1;
}

int
TAO_AV_RTCP_Object::start (void)
{
  return this->callback_->handle_start ();
}

int
TAO_AV_RTCP_Object::stop (void)
{
  return this->callback_->handle_stop ();
}

int
TAO_AV_RTCP_Object::handle_control_input (ACE_Message_Block *frame,
                                          const ACE_Addr &peer_address)
{
  return this->callback_->receive_frame (frame,
                                         0,
                                         peer_address);
}

int
TAO_AV_RTCP_Object::handle_control_output (ACE_Message_Block *frame)
{
  TAO_AV_RTCP_Callback *cb = ACE_dynamic_cast (TAO_AV_RTCP_Callback*,
                                               this->callback_);

  return cb->send_frame (frame);
}

void
TAO_AV_RTCP_Object::ts_offset (ACE_UINT32 ts_offset)
{
  TAO_AV_RTCP_Callback *cb = ACE_dynamic_cast (TAO_AV_RTCP_Callback*,
                                               this->callback_);
  cb->ts_offset (ts_offset);
}

// TAO_AV_RTCP_Callback
TAO_AV_RTCP_Callback::TAO_AV_RTCP_Callback (void)
  :is_initial_timeout_(1),
   packet_size_(0)
{
  char cname[256];
  char host[256];
  ACE_OS::hostname(host, sizeof(host));

  // TODO: determine username auto-magically?
  ACE_OS::sprintf(cname, "username@%s", host);

  this->output_.cname(cname);
}

TAO_AV_RTCP_Callback::~TAO_AV_RTCP_Callback (void)
{
}

void
TAO_AV_RTCP_Callback::schedule (int ms)
{
  this->timeout_ = ms;
}

int
TAO_AV_RTCP_Callback::handle_start (void)
{
  return 0;
}

int
TAO_AV_RTCP_Callback::handle_stop (void)
{
  return this->send_report(1);
}

int
TAO_AV_RTCP_Callback::handle_timeout (void * /*arg*/)
{
  return this->send_report(0);
}

int
TAO_AV_RTCP_Callback::send_report (int bye)
{
  // get the RTCP control object in order to get the ssrc
  TAO_AV_RTCP_Object *rtcp_prot_obj = ACE_dynamic_cast (TAO_AV_RTCP_Object*,
                                               this->protocol_object_);
  ACE_UINT32 my_ssrc = rtcp_prot_obj->ssrc ();

  RTCP_Packet *cp;
  RTCP_SDES_Packet sdes;
  ACE_CString value = "";
  ACE_CString note = "";
  unsigned char sdes_type = 0;
  RTCP_BYE_Packet *bye_packet = 0;  // only used for bye
  ACE_UINT32 ssrc_list[1];          // only used for bye

  // get an iterator for the incoming channels.
  ACE_Hash_Map_Iterator<ACE_UINT32, RTCP_Channel_In*, ACE_Null_Mutex> iter (this->inputs_);
  iter = this->inputs_.begin();

  // first send an SR/RR
  RR_Block *blocks = 0;
  RR_Block *b_iter = 0;
  RR_Block *b_ptr = 0;

  while (iter != this->inputs_.end() )
    {
      if (!b_iter)
      {
        b_ptr = (*iter).int_id_->getRRBlock ();
        if (b_ptr)
          {
            blocks = b_ptr;
            b_iter = b_ptr;
          }
      }
      else
      {
        b_ptr = (*iter).int_id_->getRRBlock ();
        if (b_ptr)
          {
            b_iter->next_ = b_ptr;
          }
      }

      iter++;
    }

  if (b_iter)
    b_iter->next_ = 0;

  if (this->output_.active ())
    {
      // get the NTP timestamp
      ACE_Time_Value unix_now = ACE_OS::gettimeofday ();
      TAO_AV_RTCP::ntp64 ntp_now = ntp64time (unix_now);
      ACE_UINT32 rtp_ts = unix_now.sec () * 8000 + unix_now.usec () / 125 +
                      this->timestamp_offset_;
      ACE_NEW_RETURN(cp,
                     RTCP_SR_Packet (my_ssrc,
                                     ntp_now.upper,
                                     ntp_now.lower,
                                     rtp_ts,
                                     this->output_.packets_sent (),
                                     this->output_.octets_sent (),
                                     blocks),
                     -1);
    }
  else
    {
      ACE_NEW_RETURN(cp,
                     RTCP_RR_Packet (my_ssrc,
                                     blocks),
                     -1);
    }

  /*
   * We always send a cname plus one other sdes
   * There's a schedule for what we send sequenced by sdes_seq_:
   *   - send 'email' every 0th & 4th packet
   *   - send 'note' every 2nd packet
   *   - send 'tool' every 6th packet
   *   - send 'name' in all the odd slots
   * (if 'note' is not the empty string, we switch the roles
   *  of name & note)
   */

  // TODO: need capability to change these settings
  switch (this->sdes_count_%8)
  {
    case 0:
    case 4:
      value = "[e-mail]";
      sdes_type = RTCP_SDES_EMAIL;
      break;
    case 2:
      if (note.length () > 0)
        {
          value = "Joe User";
          sdes_type = RTCP_SDES_NAME;
        }
      else
        {
          value = "An important note...";
          sdes_type = RTCP_SDES_NOTE;
        }
      break;
    case 6:
      value = "TAO A/V Service";
      sdes_type = RTCP_SDES_TOOL;
      break;
    case 1:
    case 3:
    case 5:
    case 7:
      if (note.length () == 0)
        {
          value = "Joe User";
          sdes_type = RTCP_SDES_NAME;
        }
      else
        {
          value = "An important note...";
          sdes_type = RTCP_SDES_NOTE;
        }
      break;
  }

  ++this->sdes_count_;

  sdes.add_item (my_ssrc,
                 RTCP_SDES_CNAME,
                 ACE_static_cast (unsigned char,
                                  ACE_OS::strlen(this->output_.cname())),
                 this->output_.cname());
  if (bye)
    {
      ssrc_list[0] = rtcp_prot_obj->ssrc ();

      ACE_NEW_RETURN (bye_packet,
                      RTCP_BYE_Packet(ssrc_list,
                                      sizeof(ssrc_list)/sizeof(ssrc_list[0]),
                                      "Got bored."),
                      -1);
    }
  else
    {
      unsigned char length = (unsigned char)(value.length() & 0xFF);
      sdes.add_item (my_ssrc, sdes_type, length, value.c_str ());
    }

  // create the message block
  char *cp_ptr;
  char *sdes_ptr;
  char *bye_ptr = 0;
  ACE_UINT16 cp_length;
  ACE_UINT16 sdes_length;
  ACE_UINT16 bye_length = 0;
  cp->get_packet_data (&cp_ptr, cp_length);
  sdes.get_packet_data (&sdes_ptr, sdes_length);
  if (bye_packet)
    bye_packet->get_packet_data(&bye_ptr, bye_length);

  ACE_Message_Block mb (cp_length + sdes_length + bye_length);

  memcpy (mb.wr_ptr (), cp_ptr, cp_length);
  mb.wr_ptr (cp_length);
  memcpy (mb.wr_ptr (), sdes_ptr, sdes_length);
  mb.wr_ptr (sdes_length);
  if (bye_length)
    {
      memcpy (mb.wr_ptr (), bye_ptr, bye_length);
      mb.wr_ptr (bye_length);
    }

  // send the report
  this->protocol_object_->send_frame (&mb);

  this->packet_size_ = cp_length + sdes_length + bye_length;

  delete cp;
  if (bye_packet)
    delete bye_packet;

  return 0;
}

void
//TAO_AV_RTCP_Callback::get_timeout (ACE_Time_Value *tv,
TAO_AV_RTCP_Callback::get_timeout (ACE_Time_Value *&tv,
                                   void *& /*arg*/)
{
  int senders = 0;
  int members = 1;  // count self as member

  // TODO: this should be 5% of the session bw
  double rtcp_bw = 1000;
  double interval;

  ACE_Hash_Map_Iterator<ACE_UINT32, RTCP_Channel_In*, ACE_Null_Mutex> iter (this->inputs_);
  iter = this->inputs_.begin();

  if (this->output_.active ())
    senders++;

  // determine the number of senders and members of this session
  while (iter != this->inputs_.end ())
    {
      if ((*iter).int_id_->active ())
        {
          if ((*iter).int_id_->sender ())
            senders++;
          members++;
        }
      iter++;
    }

  // Here we do the RTCP timeout calculation.
  interval = TAO_AV_RTCP::rtcp_interval (members,                  // members
                                         senders,                  // senders
                                         rtcp_bw,                  // rtcp_bw
                                         this->output_.active (),  // we_sent
                                         this->packet_size_,       // packet_size
                                         &this->avg_rtcp_size_,    // avg_rtcp_size
                                         this->is_initial_timeout_);  // initial)

  this->is_initial_timeout_ = 0;

  ACE_NEW (tv,
           ACE_Time_Value);

  tv->sec ((int)interval);
  tv->usec ((int)((interval - (int)interval) * 1000000));
}

int
TAO_AV_RTCP_Callback::handle_destroy (void)
{
  return 0;
}

int
TAO_AV_RTCP_Callback::receive_frame (ACE_Message_Block *frame,
                                     TAO_AV_frame_info *,
                                     const ACE_Addr &peer_address)
{
  RTCP_Channel_In *c;

  RTP_Packet packet (frame->rd_ptr(), ACE_static_cast (int, frame->length()));

  if (this->inputs_.find (packet.ssrc(), c) < 0)
    {
      ACE_NEW_RETURN (c,
                      RTCP_Channel_In (packet.ssrc(),
                                       &peer_address),
                      -1);

      this->inputs_.bind (packet.ssrc(), c);
    }

  c->recv_rtp_packet (frame, &peer_address);
  return 0;
}

int
TAO_AV_RTCP_Callback::send_frame (ACE_Message_Block *frame)
{
  RTP_Packet packet (frame->rd_ptr(), ACE_static_cast (int, frame->length()));
  this->output_.updateStatistics (&packet);

  return 0;
}

void
TAO_AV_RTCP_Callback::ts_offset (ACE_UINT32 offset)
{
  this->timestamp_offset_ = offset;
}


ACE_FACTORY_DEFINE (TAO_AV, TAO_AV_RTCP_Flow_Factory)
ACE_STATIC_SVC_DEFINE (TAO_AV_RTCP_Flow_Factory,
                       ACE_TEXT ("RTCP_Flow_Factory"),
                       ACE_SVC_OBJ_T,
                       &ACE_SVC_NAME (TAO_AV_RTCP_Flow_Factory),
                       ACE_Service_Type::DELETE_THIS |
                       ACE_Service_Type::DELETE_OBJ,
                       0)


#if defined (ACE_HAS_EXPLICIT_TEMPLATE_INSTANTIATION)

template class ACE_Hash_Map_Entry<ACE_UINT32,RTCP_Channel_In *>;
template class ACE_Hash_Map_Manager<ACE_UINT32,RTCP_Channel_In *,ACE_Null_Mutex>;
template class ACE_Hash_Map_Manager_Ex<ACE_UINT32, RTCP_Channel_In *, ACE_Hash<ACE_UINT32>, ACE_Equal_To<ACE_UINT32>, ACE_Null_Mutex>;
template class ACE_Hash_Map_Iterator<ACE_UINT32,RTCP_Channel_In *,ACE_Null_Mutex>;
template class ACE_Hash_Map_Iterator_Ex<ACE_UINT32, RTCP_Channel_In *, ACE_Hash<ACE_UINT32>, ACE_Equal_To<ACE_UINT32>, ACE_Null_Mutex>;
template class ACE_Hash_Map_Iterator_Base_Ex<ACE_UINT32, RTCP_Channel_In *, ACE_Hash<ACE_UINT32>, ACE_Equal_To<ACE_UINT32>, ACE_Null_Mutex>;
template class ACE_Hash_Map_Reverse_Iterator<ACE_UINT32,RTCP_Channel_In *,ACE_Null_Mutex>;
template class ACE_Hash_Map_Reverse_Iterator_Ex<ACE_UINT32, RTCP_Channel_In *, ACE_Hash<ACE_UINT32>, ACE_Equal_To<ACE_UINT32>, ACE_Null_Mutex>;

#elif defined (ACE_HAS_TEMPLATE_INSTANTIATION_PRAGMA)

#pragma instantiate ACE_Hash_Map_Entry<ACE_UINT32,RTCP_Channel_In *>
#pragma instantiate ACE_Hash_Map_Manager<ACE_UINT32,RTCP_Channel_In *,ACE_Null_Mutex>
#pragma instantiate ACE_Hash_Map_Manager_Ex<ACE_UINT32, RTCP_Channel_In *, ACE_Hash<ACE_UINT32>, ACE_Equal_To<ACE_UINT32>, ACE_Null_Mutex>
#pragma instantiate ACE_Hash_Map_Iterator<ACE_UINT32,RTCP_Channel_In *,ACE_Null_Mutex>
#pragma instantiate ACE_Hash_Map_Iterator_Ex<ACE_UINT32, RTCP_Channel_In *, ACE_Hash<ACE_UINT32>, ACE_Equal_To<ACE_UINT32>, ACE_Null_Mutex>
#pragma instantiate ACE_Hash_Map_Iterator_Base_Ex<ACE_UINT32, RTCP_Channel_In *, ACE_Hash<ACE_UINT32>, ACE_Equal_To<ACE_UINT32>, ACE_Null_Mutex>
#pragma instantiate ACE_Hash_Map_Reverse_Iterator<ACE_UINT32,RTCP_Channel_In *,ACE_Null_Mutex>
#pragma instantiate ACE_Hash_Map_Reverse_Iterator_Ex<ACE_UINT32, RTCP_Channel_In *, ACE_Hash<ACE_UINT32>, ACE_Equal_To<ACE_UINT32>, ACE_Null_Mutex>

#endif /* ACE_HAS_EXPLICIT_TEMPLATE_INSTANTIATION */