rtp.c

完整的RTP RTSP代码库

    }
    if (r->common.version != 2) {
      rtp_message(LOG_WARNING, "Bogus RTCP packet: version number != 2 in sub-packet %d", pc);
      return FALSE;
    }
    l += (ntohs(r->common.length) + 1) * 4;
    r  = (rtcp_t *) (((uint32_t *) r) + ntohs(r->common.length) + 1);
    pc++;	/* count of sub-packets, for debugging... */
  } while (r < end);

  /* Check that the length of the packets matches the length of the UDP */
  /* packet in which they were received...                              */
  if (l != len) {
    rtp_message(LOG_WARNING, "Bogus RTCP packet: RTCP packet length does not match UDP packet length (%d != %d)", l, len);
    return FALSE;
  }
  if (r != end) {
    rtp_message(LOG_WARNING, "Bogus RTCP packet: RTCP packet length does not match UDP packet length (%p != %p)", r, end);
    return FALSE;
  }

  return TRUE;
}

static void process_report_blocks(struct rtp *session, rtcp_t *packet, uint32_t ssrc, rtcp_rr *rrp, struct timeval *event_ts)
{
  int i;
  rtp_event	 event;
  rtcp_rr		*rr;

  /* ...process RRs... */
  if (packet->common.count == 0) {
    if (!filter_event(session, ssrc)) {
      event.ssrc = ssrc;
      event.type = RX_RR_EMPTY;
      event.data = NULL;
      event.ts   = event_ts;
      session->callback(session, &event);
    }
  } else {
    for (i = 0; i < packet->common.count; i++, rrp++) {
      rr = (rtcp_rr *) xmalloc(sizeof(rtcp_rr));
      rr->ssrc          = ntohl(rrp->ssrc);
      rr->fract_lost    = rrp->fract_lost;	/* Endian conversion handled in the */
      rr->total_lost    = rrp->total_lost;	/* definition of the rtcp_rr type.  */
      rr->last_seq      = ntohl(rrp->last_seq);
      rr->jitter        = ntohl(rrp->jitter);
      rr->lsr           = ntohl(rrp->lsr);
      rr->dlsr          = ntohl(rrp->dlsr);

      /* Create a database entry for this SSRC, if one doesn't already exist... */
      create_source(session, rr->ssrc, FALSE);

      /* Store the RR for later use... */
      insert_rr(session, ssrc, rr, event_ts);

      /* Call the event handler... */
      if (!filter_event(session, ssrc)) {
	event.ssrc = ssrc;
	event.type = RX_RR;
	event.data = (void *) rr;
	event.ts   = event_ts;
	session->callback(session, &event);
      }
    }
  }
}

static void process_rtcp_sr(struct rtp *session, rtcp_t *packet, struct timeval *event_ts)
{
  uint32_t	 ssrc;
  rtp_event	 event;
  source	*s;
  rtcp_sr       *sr;

  ssrc = ntohl(packet->r.sr.sr.ssrc);
  s = create_source(session, ssrc, FALSE);
  if (s == NULL) {
    rtp_message(LOG_WARNING, "Source 0x%08x invalid, skipping...", ssrc);
    return;
  }

  /* Mark as an active sender, if we get a sender report... */
  if (s->sender == FALSE) {
    s->sender = TRUE;
    session->sender_count++;
  }

  /* Process the SR... */
  sr = (rtcp_sr *) &s->sr;
  sr->ssrc          = ssrc;
  sr->ntp_sec       = ntohl(packet->r.sr.sr.ntp_sec);
  sr->ntp_frac      = ntohl(packet->r.sr.sr.ntp_frac);
  sr->rtp_ts        = ntohl(packet->r.sr.sr.rtp_ts);
  sr->sender_pcount = ntohl(packet->r.sr.sr.sender_pcount);
  sr->sender_bcount = ntohl(packet->r.sr.sr.sender_bcount);

  /* Store the SR for later retrieval... */
  s->last_sr = *event_ts;

  /* Call the event handler... */
  if (!filter_event(session, ssrc)) {
    event.ssrc = ssrc;
    event.type = RX_SR;
    event.data = (void *) sr;
    event.ts   = event_ts;
    session->callback(session, &event);
  }

  process_report_blocks(session, packet, ssrc, packet->r.sr.rr, event_ts);

  if (((packet->common.count * 6) + 1) < (ntohs(packet->common.length) - 5)) {
    rtp_message(LOG_NOTICE, "Profile specific SR extension ignored");
  }
}

static void process_rtcp_rr(struct rtp *session, rtcp_t *packet, struct timeval *event_ts)
{
  uint32_t		 ssrc;
  source		*s;

  ssrc = ntohl(packet->r.rr.ssrc);
  s = create_source(session, ssrc, FALSE);
  if (s == NULL) {
    rtp_message(LOG_WARNING, "Source 0x%08x invalid, skipping...", ssrc);
    return;
  }

  process_report_blocks(session, packet, ssrc, packet->r.rr.rr, event_ts);

  if (((packet->common.count * 6) + 1) < ntohs(packet->common.length)) {
    rtp_message(LOG_INFO, "Profile specific RR extension ignored");
  }
}

static void process_rtcp_sdes(struct rtp *session, rtcp_t *packet, struct timeval *event_ts)
{
  int 			count = packet->common.count;
  rtcp_sdes_t 	*sd   = &packet->r.sdes;
  rtcp_sdes_item 		*rsp; 
  rtcp_sdes_item		*rspn;
  rtcp_sdes_item 		*end  = (rtcp_sdes_item *) ((uint32_t *)packet + packet->common.length + 1);
  source 			*s;
  rtp_event		 event;

  while (--count >= 0) {
    rsp = &sd->item[0];
    if (rsp >= end) {
      break;
    }
    sd->ssrc = ntohl(sd->ssrc);
    s = create_source(session, sd->ssrc, FALSE);
    if (s == NULL) {
      rtp_message(LOG_NOTICE, "Cannot get valid source entry for 0x%08x, skipping...", sd->ssrc);
    } else {
      for (; rsp->type; rsp = rspn ) {
	rspn = (rtcp_sdes_item *)((char*)rsp+rsp->length+2);
	if (rspn >= end) {
	  rsp = rspn;
	  break;
	}
	if (rtp_set_sdes(session, sd->ssrc, rsp->type, rsp->data, rsp->length)) {
	  if (!filter_event(session, sd->ssrc)) {
	    event.ssrc = sd->ssrc;
	    event.type = RX_SDES;
	    event.data = (void *) rsp;
	    event.ts   = event_ts;
	    session->callback(session, &event);
	  }
	} else {
	  rtp_message(LOG_WARNING, "Invalid sdes item for source 0x%08x, skipping...", sd->ssrc);
	}
      }
    }
    sd = (rtcp_sdes_t *) ((uint32_t *)sd + (((char *)rsp - (char *)sd) >> 2)+1);
  }
  if (count >= 0) {
    rtp_message(LOG_INFO, "Invalid RTCP SDES packet, some items ignored.");
  }
}

static void process_rtcp_bye(struct rtp *session, rtcp_t *packet, struct timeval *event_ts)
{
  int		 i;
  uint32_t	 ssrc;
  rtp_event	 event;
  source		*s;

  for (i = 0; i < packet->common.count; i++) {
    ssrc = ntohl(packet->r.bye.ssrc[i]);
    /* This is kind-of strange, since we create a source we are about to delete. */
    /* This is done to ensure that the source mentioned in the event which is    */
    /* passed to the user of the RTP library is valid, and simplify client code. */
    s = create_source(session, ssrc, FALSE);
    /* Call the event handler... */
    if (!filter_event(session, ssrc)) {
      event.ssrc = ssrc;
      event.type = RX_BYE;
      event.data = NULL;
      event.ts   = event_ts;
      session->callback(session, &event);
    }
    /* Mark the source as ready for deletion. Sources are not deleted immediately */
    /* since some packets may be delayed and arrive after the BYE...              */
    // redundant from above s = get_source(session, ssrc);
    s->got_bye = TRUE;
    check_source(s);
    session->bye_count++;
  }
}

static void process_rtcp_app(struct rtp *session, rtcp_t *packet, struct timeval *event_ts)
{
  uint32_t         ssrc;
  rtp_event        event;
  rtcp_app        *app;
  source          *s;
  int              data_len;

  /* Update the database for this source. */
  ssrc = ntohl(packet->r.app.ssrc);
  s = create_source(session, ssrc, FALSE);
  // redundant - wmay - s = get_source(session, ssrc);
  if (s == NULL) {
    /* This should only occur in the event of database malfunction. */
    rtp_message(LOG_NOTICE, "Source 0x%08x invalid, skipping...", ssrc);
    return;
  }
  check_source(s);

  /* Copy the entire packet, converting the header (only) into host byte order. */
  app = (rtcp_app *) xmalloc(RTP_MAX_PACKET_LEN);
  app->version        = packet->common.version;
  app->p              = packet->common.p;
  app->subtype        = packet->common.count;
  app->pt             = packet->common.pt;
  app->length         = ntohs(packet->common.length);
  app->ssrc           = ssrc;
  app->name[0]        = packet->r.app.name[0];
  app->name[1]        = packet->r.app.name[1];
  app->name[2]        = packet->r.app.name[2];
  app->name[3]        = packet->r.app.name[3];
  data_len            = (app->length - 2) * 4;
  memcpy(app->data, packet->r.app.data, data_len);

  /* Callback to the application to process the app packet... */
  if (!filter_event(session, ssrc)) {
    event.ssrc = ssrc;
    event.type = RX_APP;
    event.data = (void *) app;       /* The callback function MUST free this! */
    event.ts   = event_ts;
    session->callback(session, &event);
  }
}


static void rtp_process_rtcp (struct rtp *session, uint8_t *buffer, uint32_t buflen)
{
  rtp_event	 event;
  struct timeval	 event_ts;
  rtcp_t		*packet;
  int		 first;
  uint32_t	 packet_ssrc = rtp_my_ssrc(session);

  gettimeofday(&event_ts, NULL);
  if (validate_rtcp(buffer, buflen)) {
    first  = TRUE;
    packet = (rtcp_t *) buffer;
    while (packet < (rtcp_t *) (buffer + buflen)) {
      switch (packet->common.pt) {
      case RTCP_SR:
	if (first && !filter_event(session, ntohl(packet->r.sr.sr.ssrc))) {
	  event.ssrc  = ntohl(packet->r.sr.sr.ssrc);
	  event.type  = RX_RTCP_START;
	  event.data  = &buflen;
	  event.ts    = &event_ts;
	  packet_ssrc = event.ssrc;
	  session->callback(session, &event);
	}
	process_rtcp_sr(session, packet, &event_ts);
	break;
      case RTCP_RR:
	if (first && !filter_event(session, ntohl(packet->r.rr.ssrc))) {
	  event.ssrc  = ntohl(packet->r.rr.ssrc);
	  event.type  = RX_RTCP_START;
	  event.data  = &buflen;
	  event.ts    = &event_ts;
	  packet_ssrc = event.ssrc;
	  session->callback(session, &event);
	}
	process_rtcp_rr(session, packet, &event_ts);
	break;
      case RTCP_SDES:
	if (first && !filter_event(session, ntohl(packet->r.sdes.ssrc))) {
	  event.ssrc  = ntohl(packet->r.sdes.ssrc);
	  event.type  = RX_RTCP_START;
	  event.data  = &buflen;
	  event.ts    = &event_ts;
	  packet_ssrc = event.ssrc;
	  session->callback(session, &event);
	}
	process_rtcp_sdes(session, packet, &event_ts);
	break;
      case RTCP_BYE:
	if (first && !filter_event(session, ntohl(packet->r.bye.ssrc[0]))) {
	  event.ssrc  = ntohl(packet->r.bye.ssrc[0]);
	  event.type  = RX_RTCP_START;
	  event.data  = &buflen;
	  event.ts    = &event_ts;
	  packet_ssrc = event.ssrc;
	  session->callback(session, &event);
	}
	process_rtcp_bye(session, packet, &event_ts);
	break;
      case RTCP_APP:
	if (first && !filter_event(session, ntohl(packet->r.app.ssrc))) {
	  event.ssrc  = ntohl(packet->r.app.ssrc);
	  event.type  = RX_RTCP_START;
	  event.data  = &buflen;
	  event.ts    = &event_ts;
	  packet_ssrc = event.ssrc;
	  session->callback(session, &event);
	}
	process_rtcp_app(session, packet, &event_ts);
	break;
      default: 
	rtp_message(LOG_WARNING, "RTCP packet with unknown type (%d) ignored.", packet->common.pt);
	break;
      }
      packet = (rtcp_t *) ((uint8_t *) packet + (4 * (ntohs(packet->common.length) + 1)));
      first  = FALSE;
    }
    if (session->avg_rtcp_size < 0) {
      /* This is the first RTCP packet we've received, set our initial estimate */
      /* of the average  packet size to be the size of this packet.             */
      session->avg_rtcp_size = buflen + RTP_LOWER_LAYER_OVERHEAD;
    } else {
      /* Update our estimate of the average RTCP packet size. The constants are */
      /* 1/16 and 15/16 (section 6.3.3 of draft-ietf-avt-rtp-new-02.txt).       */
      session->avg_rtcp_size = (0.0625 * (buflen + RTP_LOWER_LAYER_OVERHEAD)) + (0.9375 * session->avg_rtcp_size);
    }
    /* Signal that we've finished processing this packet */
    if (!filter_event(session, packet_ssrc)) {
      event.ssrc = packet_ssrc;
      event.type = RX_RTCP_FINISH;
      event.data = NULL;
      event.ts   = &event_ts;
      session->callback(session, &event);
    }
  } else {
    rtp_message(LOG_INFO, "Invalid RTCP packet discarded");
    session->invalid_rtcp_count++;
  }
}


void rtp_process_ctrl (struct rtp *session, uint8_t *buffer, uint32_t buflen)
{
  /* This routine processes incoming RTCP packets */
  if (buflen > 0) {
    if (session->rtcp_encryption_enabled)
      {
	/* Decrypt the packet... */
	int ret = (session->rtcp_decrypt_func)(session->encrypt_userdata, buffer, &buflen);
	if (ret == FALSE) return;
      }
    rtp_process_rtcp(session, buffer, buflen);
  }
}

/**
 * rtp_recv:
 * @session: the session pointer (returned by rtp_init())
 * @timeout: the amount of time that rtcp_recv() is allowed to block
 * @curr_rtp_ts: the current time expressed in units of the media
 * timestamp.
 *
 * Receive RTP packets and dispatch them.
 *
 * Returns: TRUE if data received, FALSE if the timeout occurred.
 */
int rtp_recv(struct rtp *session, struct timeval *timeout, uint32_t curr_rtp_ts)
{
  check_database(session);
  udp_fd_zero(session->udp_session);
  udp_fd_set(session->udp_session, session->rtp_socket);
  udp_fd_set(session->udp_session, session->rtcp_socket);
  if (udp_select(session->udp_session, timeout) > 0) {
    if (udp_fd_isset(session->udp_session, session->rtp_socket)) {
      rtp_recv_data(session, curr_rtp_ts);
    }
    if (udp_fd_isset(session->udp_session, session->rtcp_socket)) {
      uint8_t		 buffer[RTP_MAX_PACKET_LEN];
      int		 buflen;
      buflen = udp_recv(session->rtcp_socket, buffer, RTP_MAX_PACKET_LEN);
      rtp_process_ctrl(session, buffer, buflen);
    }
    check_database(session);
    return TRUE;
  }
  check_database(session);
  return FALSE;
}



/**
 * rtp_add_csrc:
 * @session: the session pointer (returned by rtp_init()) 
 * @csrc: Constributing SSRC identifier
 * 
 * Adds @csrc to list of contributing sources used in SDES items.
 * Used by mixers and transcoders.
 * 
 * Return value: TRU