rtp.cpp

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

      // set the ssrc on the control object so the RTCP traffic can be matched
      // to the RTP traffic
      rtcp_prot_obj->ssrc(this->ssrc_);

      ACE_NEW_RETURN (rtp_packet,
                      RTP_Packet (0,                            // padding
                                  frame_info->boundary_marker,  // marker
                                  this->format_,                // payload type
                                  frame_info->sequence_num,     // sequence num
                                  frame_info->timestamp,        // time stamp
                                  this->ssrc_,                  // ssrc
                                  csrc_count,                   // csrc count
                                  csrc_list,                    // csrc list
                                  frame->rd_ptr (),             // data
                                  (ACE_UINT16)frame->length ()),// data size
                                  -1);

      frame_info->sequence_num ++;
    }
  else
    {
      // TODO: For periodic RTP packets (constant rate), the RFC suggests
      //       increasing the clock by the number of samples each frame rather
      //       than relying on the system time

      // The RFC specifies at least one timestamp unit per sample as well as a
      //  random offset.  It used to be in milliseconds so I left it that way
      //  for non-audio streams.
      unsigned int samples_per_sec;
      double samples_per_usec;

      switch (this->format_)
      {
        case RTP_PT_PCMU:
        case RTP_PT_CELP:
        case RTP_PT_G721:
        case RTP_PT_GSM:
        case RTP_PT_DVI:
        case RTP_PT_LPC:
        case RTP_PT_PCMA:
        case RTP_PT_G722:
          samples_per_sec = 8000;
          break;
        case RTP_PT_L16_STEREO:
        case RTP_PT_L16_MONO:
          samples_per_sec = 44100;
          break;
        default:
          samples_per_sec = 1000000;
      };

      samples_per_usec = samples_per_sec/1000000.0;

      ACE_Time_Value now = ACE_OS::gettimeofday();

      ACE_UINT32 ts = (ACE_UINT32)
                      (now.sec () * samples_per_sec +
                       ((double)now.usec () * samples_per_usec) +
                       this->timestamp_offset_);

      ACE_NEW_RETURN (rtp_packet,
                      RTP_Packet (0,                            // padding
                                  0,                            // marker
                                  this->format_,                // payload type
                                  this->sequence_num_,          // sequence num
                                  ts,                           // time stamp
                                  this->ssrc_,                  // ssrc
                                  csrc_count,                   // csrc count
                                  csrc_list,                    // csrc list
                                  frame->rd_ptr (),             // data
                                  (ACE_UINT16)frame->length ()),// data size
                                  -1);

      this->sequence_num_ ++;
    }

  char *data_ptr;
  ACE_UINT16 data_length;
  rtp_packet->get_packet_data (&data_ptr, data_length);

  ACE_Message_Block mb (data_ptr, data_length);
  mb.wr_ptr (data_length);

  result = this->transport_->send (&mb);
  if (result < 0)
    ACE_ERROR_RETURN ( (LM_ERROR,"TAO_AV_RTP::send_frame failed\n"),result);

  TAO_AV_RTCP_Object *rtcp_prot_obj = ACE_dynamic_cast (TAO_AV_RTCP_Object*,
                                                        this->control_object_);
  if (rtcp_prot_obj)
    rtcp_prot_obj->handle_control_output (&mb);

  delete rtp_packet;

  return 0;
}

int
TAO_AV_RTP_Object::send_frame (const iovec *iov,
                               int iovcnt,
                               TAO_AV_frame_info *frame_info)
{
  int result = -1;
  RTP_Packet *rtp_packet = 0;
  ACE_UINT32 csrc_count = 0;  // Assume for now no mixers/translators
  ACE_UINT32 *csrc_list = 0;

  if (this->connection_gone_)
    {
      errno = ECONNRESET;
      return -1;
    }

  if (frame_info != 0)
    {
      if (frame_info->format != this->format_)
        ACE_DEBUG ((LM_DEBUG,
                    "TAO_AV_RTP_Object::send_frame - error: format type mismatch"));
      this->sequence_num_ = frame_info->sequence_num;
      if (frame_info->ssrc != 0)
        this->ssrc_ = frame_info->ssrc;

      TAO_AV_RTCP_Object *rtcp_prot_obj = ACE_dynamic_cast (TAO_AV_RTCP_Object*,
                                                            this->control_object_);
      // set the ssrc on the control object so the RTCP traffic can be matched
      // to the RTP traffic
      rtcp_prot_obj->ssrc(this->ssrc_);


      ACE_NEW_RETURN (rtp_packet,
                      RTP_Packet (0,                            // padding
                                  frame_info->boundary_marker,  // marker
                                  this->format_,                // payload type
                                  frame_info->sequence_num,     // sequence num
                                  frame_info->timestamp,        // time stamp
                                  this->ssrc_,                  // ssrc
                                  csrc_count,                   // csrc count
                                  csrc_list,                    // csrc list
                                  (char *)iov[0].iov_base,              // data
                                  iov[0].iov_len),              // data size
                                  -1);

      frame_info->sequence_num ++;
    }
  else
    {
      // TODO: For periodic RTP packets (constant rate), the RFC suggests
      //       increasing the clock by the number of samples each frame rather
      //       than relying on the system time

      // The RFC specifies at least one timestamp unit per sample as well as a
      //  random offset.  It used to be in milliseconds so I left it that way
      //  for non-audio streams.

      unsigned int samples_per_sec;
      double samples_per_usec;

      switch (this->format_)
      {
        case RTP_PT_PCMU:
        case RTP_PT_CELP:
        case RTP_PT_G721:
        case RTP_PT_GSM:
        case RTP_PT_DVI:
        case RTP_PT_LPC:
        case RTP_PT_PCMA:
        case RTP_PT_G722:
          samples_per_sec = 8000;
          break;
        case RTP_PT_L16_STEREO:
        case RTP_PT_L16_MONO:
          samples_per_sec = 44100;
          break;
        default:
          samples_per_sec = 1000000;
      };

      samples_per_usec = samples_per_sec/1000000.0;

      ACE_Time_Value now = ACE_OS::gettimeofday();

      ACE_UINT32 ts = (ACE_UINT32)
                      (now.sec () * samples_per_sec +
                       ((double)now.usec () * samples_per_usec) +
                       this->timestamp_offset_);

      ACE_NEW_RETURN (rtp_packet,
                      RTP_Packet (0,                            // padding
                                  0,                            // marker
                                  this->format_,                // payload type
                                  this->sequence_num_,          // sequence num
                                  ts,                           // time stamp
                                  this->ssrc_,                  // ssrc
                                  csrc_count,                   // csrc count
                                  csrc_list,                    // csrc list
                                  (char *)iov[0].iov_base,              // data
                                  iov[0].iov_len),              // data size
                                  -1);

      this->sequence_num_ ++;
    }

  char *data_ptr;
  ACE_UINT16 data_length;
  rtp_packet->get_packet_data (&data_ptr, data_length);

  iovec send_iov[ACE_IOV_MAX];
  send_iov [0].iov_base = data_ptr;
  send_iov [0].iov_len  = data_length;
  for (int i=1;i<iovcnt; i++)
    send_iov [i] = iov [i];
  result = this->transport_->send (send_iov, iovcnt);

  delete rtp_packet;

  if (result < 0)
    ACE_ERROR_RETURN ((LM_ERROR,"TAO_AV_RTP::send_frame failed\n"),result);

  return 0;
}

int
TAO_AV_RTP_Object::send_frame (const char*,
                               size_t)
{
  return 0;
}


TAO_AV_RTP_Object::TAO_AV_RTP_Object (TAO_AV_Callback *callback,
                                      TAO_AV_Transport *transport)
  :TAO_AV_Protocol_Object (callback,transport),
   control_object_ (0),
   connection_gone_ (0)
{
  this->sequence_num_ = ACE_OS::rand ();
  this->timestamp_offset_ = ACE_OS::rand ();

  char buf [BUFSIZ];
  int result = ACE_OS::hostname (buf, BUFSIZ);
  unsigned long ipaddr = 0;
  if (result == 0)
    ipaddr = ACE_OS::inet_addr (buf);
  this->ssrc_ = TAO_AV_RTCP::alloc_srcid (ipaddr);

  this->frame_.size (2 * this->transport_->mtu ());
}

TAO_AV_RTP_Object::~TAO_AV_RTP_Object (void)
{
}

int
TAO_AV_RTP_Object::destroy (void)
{
  if(this->control_object_)
     this->control_object_->destroy ();

  this->callback_->handle_destroy ();
  delete this;

  return 0;
}

int
TAO_AV_RTP_Object::set_policies (const TAO_AV_PolicyList &policy_list)
{
  this->policy_list_ = policy_list;
  u_int num_policies = this->policy_list_.length ();
  TAO_AV_Policy *policy = 0;

  for (u_int i=0; i< num_policies;i++)
    {
      policy = this->policy_list_ [i];
      switch (policy->type ())
        {
        case TAO_AV_PAYLOAD_TYPE_POLICY:
          {
            TAO_AV_Payload_Type_Policy *payload_policy =
              ACE_static_cast (TAO_AV_Payload_Type_Policy *,policy);
            if (payload_policy == 0)
              ACE_ERROR_RETURN ( (LM_ERROR,"TAO_AV_RTP_Object::send_frame:Payload policy not defined\n"),-1);
            this->format_ = payload_policy->value ();
          }
          break;
        case TAO_AV_SSRC_POLICY:
          {
            TAO_AV_SSRC_Policy *ssrc_policy =
              ACE_static_cast (TAO_AV_SSRC_Policy *,policy);
            if (ssrc_policy == 0)
              ACE_ERROR_RETURN ( (LM_ERROR,"TAO_AV_RTP_Object::send_frame:SSRC policy not defined\n"),-1);
            this->ssrc_ = ssrc_policy->value ();;
          }
          break;
        default:
          break;
        }
    }

  return 0;
}

void
TAO_AV_RTP_Object::control_object (TAO_AV_Protocol_Object *object)
{
  this->control_object_ = object;
  TAO_AV_RTCP_Object *rtcp_prot_obj = ACE_dynamic_cast (TAO_AV_RTCP_Object*,
                                                        this->control_object_);
  rtcp_prot_obj->ssrc (this->ssrc_);
  rtcp_prot_obj->ts_offset (this->timestamp_offset_);
}

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

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

// TAO_AV_RTP_Flow_Factory
TAO_AV_RTP_Flow_Factory::TAO_AV_RTP_Flow_Factory (void)
{
}

TAO_AV_RTP_Flow_Factory::~TAO_AV_RTP_Flow_Factory (void)
{
}

int
TAO_AV_RTP_Flow_Factory::init (int /* argc */,
                                char * /* argv */ [])
{
  return 0;
}

TAO_AV_Protocol_Object*
TAO_AV_RTP_Flow_Factory::make_protocol_object (TAO_FlowSpec_Entry *entry,
                                               TAO_Base_StreamEndPoint *endpoint,
                                               TAO_AV_Flow_Handler *handler,
                                               TAO_AV_Transport *transport)
{
  TAO_AV_Callback *callback = 0;

  if( endpoint->get_callback (entry->flowname (), callback) ) {
    ACE_ERROR_RETURN ((LM_ERROR, "(%N,%l) Invalid callback\n"), 0);
  }

  TAO_AV_Protocol_Object *object = 0;
  ACE_NEW_RETURN (object,
                  TAO_AV_RTP_Object (callback,
                                     transport),
                  0);
  callback->open (object,
                  handler);
  endpoint->set_protocol_object (entry->flowname (),
                                 object);

  return object;
}

int
TAO_AV_RTP_Flow_Factory::match_protocol (const char *flow_string)
{
  if (ACE_OS::strncasecmp (flow_string,"RTP",3) == 0)
    {
      return 1;
    }
  return 0;
}

const char *
TAO_AV_RTP_Flow_Factory::control_flow_factory (void)
{
  return "RTCP";
}

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