rtp.c

用C实现的在LINUX系统下的聊天程序.无需登录就可以下载

  remove.ssrc = ssrc;

  return ((rtp_source_t *) ohtbl_remove (htbl, &remove));
}

/* Parses received network data and checks for validity */
rtp_packet_t *rtp_recv (rtp_t *s, vstr_t *p) {

  uint8_t padding;
  rtp_packet_t *rtp_packet = (rtp_packet_t *) alloc (RTP_PACKET_SIZE);
  vstr_t map;
  rtp_source_t *member;

  /* maps RTP packet into vstr */
  vstr_map (&map, rtp_packet, RTP_PACKET_SIZE);
  vstr_set (&map, RTP_PACKET_IDX, RTP_PACKET_IDX);
  vstr_put_tail (&map, p -> head, p -> size);

  rtp_packet -> seq = ntohs(rtp_packet -> seq);
  rtp_packet -> ts = ntohl(rtp_packet -> ts);
  rtp_packet -> ssrc = ntohl(rtp_packet -> ssrc);

  /* check SSRC */
  pthread_mutex_lock (&s -> guard);
  member = rtp_find_member(&s -> members, rtp_packet -> ssrc);
  pthread_mutex_unlock (&s -> guard);
  if (member == NULL) {
    free (rtp_packet);
    return (NULL);
  }

  vstr_adv_head (&map, RTP_HEADER_SIZE);

  /* checks for RTP packet version */
  if(rtp_packet -> ver != RTP_VERSION) {
    free (rtp_packet);
    return (NULL);
  }

  /* check for validity of payload type:
     it must not be equal to any RTCP type */
  if (rtp_packet -> pt >= RTCP_TYPE_MIN && rtp_packet -> pt <= RTCP_TYPE_MAX) {
    free (rtp_packet);
    return (NULL);
  }

  /* checks for CSRC list and maps it */
  if (rtp_packet -> cc > 0)
    vstr_adv_head (&map, rtp_packet -> cc * sizeof(uint32_t));

  /* checks for header extension */
  if (rtp_packet -> x == 1) {
    vstr_get_head (&map, &rtp_packet -> xdef, sizeof(uint32_t));
    rtp_packet -> xdef = htons (rtp_packet -> xdef);
    rtp_packet -> xlen = htons (rtp_packet -> xlen);
    if (rtp_packet -> xlen != 0) {
      rtp_packet -> xhdr = (uint32_t *) map.head;
      vstr_adv_head (&map, rtp_packet -> xlen * sizeof(uint32_t));
    }
  }

  /* maps payload */
  rtp_packet -> payload = map.head;
  rtp_packet -> pay_len = map.size;

  /* checks for header padding */
  if (rtp_packet -> p == 1) {
    vstr_get_tail (&map, &padding, sizeof(uint8_t));

    if (padding > rtp_packet -> pay_len) {
      free (rtp_packet);
      return (NULL);
    }

    if (padding == 0) {
      free (rtp_packet);
      return (NULL);
    }

    /* updates payload lenght */
    rtp_packet -> pay_len -= padding;
  }

  pthread_mutex_lock (&member -> guard);
  rtp_update_jitter (member, rtp_packet -> ts);
  pthread_mutex_unlock (&member -> guard);

  return (rtp_packet);
}

int _rtp_enqueue_min_solve (uint16_t cur, uint16_t next) {

  if (cur > (uint16_t) RTP_MAX_QUEUE)
    return (next < cur);
  else
    return (next + (uint16_t) RTP_MAX_QUEUE < cur + (uint16_t) RTP_MAX_QUEUE);
}

int _rtp_enqueue_max_solve (uint16_t cur, uint16_t next) {

  if (cur < (uint16_t) -RTP_MAX_QUEUE)
    return (next > cur);
  else
    return (next + (uint16_t) RTP_MAX_QUEUE > cur + (uint16_t) RTP_MAX_QUEUE);
}

/* Adds RTP packet to member packet queue */
int rtp_enqueue (rtp_t *s, rtp_packet_t *rtp_packet) {

  rtp_source_t *member;

  /* check SSRC */
  pthread_mutex_lock (&s -> guard);
  member = rtp_find_member(&s -> members, rtp_packet -> ssrc);
  pthread_mutex_unlock (&s -> guard);
  if (member == NULL)
    return (-1);

  /* check for initialization */
  pthread_mutex_lock (&member -> guard);
  if (member -> fail_count >= RTP_MAX_FAIL_COUNT) {
    member -> fail_count = 0;
    member -> min_seq = rtp_packet -> seq;
    member -> max_seq = rtp_packet -> seq;
  }
  else {

    /* update minimal sequence number received */
    if (member -> min_seq == 0 || _rtp_enqueue_min_solve (member -> min_seq, rtp_packet -> seq))
      member -> min_seq = rtp_packet -> seq;

    /* update maximal sequence number received */
    if (_rtp_enqueue_max_solve (member -> max_seq, rtp_packet -> seq))
      member -> max_seq = rtp_packet -> seq;
  }

  /* update packets received from this source */
  member -> received++;

  /* update extended last sequence number received */
  member -> last_seq = rtp_packet -> seq + (member -> cycles * 65536);

  pthread_mutex_unlock (&member -> guard);

  /* insert RTP packet in queue and return */
  return (phtbl_insert (&member -> pqueue, rtp_packet));
}

/* Get (remove) packet from queue */
rtp_packet_t *rtp_get_package (phtbl_t *htbl, uint16_t seq) {

  rtp_packet_t rtp_packet;

  rtp_packet.seq = seq;
  return ((rtp_packet_t *) phtbl_remove (htbl, &rtp_packet));
}

/* Get next RTP packet from packet queue */
rtp_packet_t *rtp_dequeue (rtp_t *s, uint32_t ssrc) {

  rtp_packet_t *packet;
  rtp_source_t *member;

  /* check SSRC */
  pthread_mutex_lock (&s -> guard);
  member = rtp_find_member(&s -> members, ssrc);
  pthread_mutex_unlock (&s -> guard);
  if (member == NULL)
    return (NULL);

  pthread_mutex_lock (&member -> guard);
  if ((packet = rtp_get_package (&member -> pqueue, member -> min_seq)) == NULL) {
    member -> fail_count++;
    member -> lost++;
    if (member -> received != 0)
      member -> frac = (uint8_t) (((uint32_t) 256 * member -> lost) / member -> received);
  }
  else
    member -> fail_count = 0;

  member -> min_seq++;
  if (member -> min_seq == 0)
    member -> cycles++;
  pthread_mutex_unlock (&member -> guard);

  return (packet);
}

/* Appends a RTCP Sender Report + Receiver Reports (one for each member) */
void rtcp_send_ctrl (rtp_t *s, vstr_t *vstr) {

  struct timeval ts;
  rtcp_packet_t *rtcp_packet;
  rtp_source_t *member;

  uint32_t ntp_sec;             /* NTP time (seconds)  */
  uint32_t ntp_frac;            /* NTP time (fraction) */
  int i;                        /* just a counter      */

  /* maps RTCP packet into vstr */
  rtcp_packet = (rtcp_packet_t *) vstr -> tail;

  /* RTCP header */
  rtcp_packet -> ver = RTP_VERSION;                       /* RTP version     */
  rtcp_packet -> p = 0;                                   /* padding flag    */
  vstr_adv_tail (vstr, RTCP_HEADER_SIZE);

  pthread_mutex_lock (&s -> guard);
  rtcp_packet -> count = ohtbl_size (&s -> members);      /* report count    */
  rtcp_packet -> r.sr.ssrc = htonl(s -> ssrc);            /* SSRC of sender  */

  /* sender info */
  if (s -> flags & RTP_SENDER) {
    rtcp_packet -> pt = RTCP_SR;           /* RTCP packet is a Sender Report */
    /* calculates NTP time */
    gettimeofday(&ts, NULL);
    ntp_sec = ts.tv_sec + SECS_BETWEEN_1900_1970;
    ntp_frac = (ts.tv_usec << 12) + (ts.tv_usec << 8) - ((ts.tv_usec * 3650) >> 6);

    rtcp_packet -> r.sr.ntp_sec = htonl(ntp_sec);         /* NTP timestamp   */
    rtcp_packet -> r.sr.ntp_frac = htonl(ntp_frac);       /* NTP timestamp   */
    rtcp_packet -> r.sr.rtp_ts = htonl(ts.tv_usec + ts.tv_sec * 1000000);
    rtcp_packet -> r.sr.pcount = htonl(s -> pcount);      /* packet count    */
    rtcp_packet -> r.sr.ocount = htonl(s -> ocount);      /* octet count     */
    vstr_adv_tail (vstr, RTCP_SR_SIZE);
  }
  else
    rtcp_packet -> pt = RTCP_RR;         /* RTCP packet is a Receiver Report */

  /* report blocks */
  ohtbl_flush (&s -> members);
  for (i = 0; i < ohtbl_size (&s -> members); i++) {
    member = (rtp_source_t *) ohtbl_traverse (&s -> members);
    pthread_mutex_lock (&member -> guard);
    rtcp_packet -> r.sr.rr[i].ssrc = htonl(member -> ssrc); /* SSRC of source  */
    rtcp_packet -> r.sr.rr[i].frac = member -> frac;        /* fraction lost   */
    rtcp_packet -> r.sr.rr[i].lost = htonl(member -> lost); /* packets lost    */
    rtcp_packet -> r.sr.rr[i].last_seq = htonl(member -> last_seq);  /* ext. seq. */
    rtcp_packet -> r.sr.rr[i].jitter = htonl(member -> jitter);  /* jitter        */
    rtcp_packet -> r.sr.rr[i].lsr = htonl(member -> lsr);   /* last SR         */
    rtcp_packet -> r.sr.rr[i].dlsr = htonl(member -> dlsr); /* delay since LSR */
    pthread_mutex_unlock (&member -> guard);
    vstr_adv_tail (vstr, RTCP_RR_SIZE);
  }

  pthread_mutex_unlock (&s -> guard);

  rtcp_packet -> len = vstr -> size / 4 - 1;               /* packet length   */

  return;
}

/* Appends a RTCP Source Description report */
void rtcp_append_sdes (rtp_t *s, vstr_t *vstr, int flags) {

  rtcp_packet_t *rtcp_packet = (rtcp_packet_t *) vstr -> tail;
  rtcp_sdes_item_t *sdes_item;

  int i;                                /* just a counter */

  /* RTCP header */
  rtcp_packet -> ver = RTP_VERSION;                       /* RTP version     */
  rtcp_packet -> p = 0;                                   /* padding flag    */
  rtcp_packet -> count = 1;                               /* SDES count      */
  rtcp_packet -> pt = RTCP_SDES;                          /* PT = SDES       */
  vstr_adv_tail (vstr, RTCP_SDES_HEADER_SIZE + sizeof(uint32_t));

  pthread_mutex_lock (&s -> guard);
  rtcp_packet -> r.sdes.ssrc = htonl (s -> ssrc);         /* SSRC of sender  */

  /* SDES items */
  for (i = 0; i < RTCP_SDES_LAST; i++) {
    if (flags & (1 << i)) {

      sdes_item = (rtcp_sdes_item_t *) vstr -> tail;
      vstr_adv_tail (vstr, RTCP_SDES_ITEM_SIZE);
      sdes_item -> type = i + 1;

      switch (sdes_item -> type) {
        case RTCP_SDES_CNAME:
          sdes_item -> length = strlen (s -> cname);
          vstr_put_tail (vstr, s -> cname, strlen (s -> cname));
          break;
        case RTCP_SDES_NAME:
          sdes_item -> length = strlen (s -> name);
          vstr_put_tail (vstr, s -> name, strlen (s -> name));
          break;
        case RTCP_SDES_EMAIL:
          sdes_item -> length = strlen (s -> email);
          vstr_put_tail (vstr, s -> email, strlen (s -> email));
          break;
        case RTCP_SDES_PHONE:
          sdes_item -> length = strlen (s -> phone);
          vstr_put_tail (vstr, s -> phone, strlen (s -> phone));
          break;
        case RTCP_SDES_LOC:
          sdes_item -> length = strlen (s -> loc);
          vstr_put_tail (vstr, s -> loc, strlen (s -> loc));
          break;
        case RTCP_SDES_TOOL:
          sdes_item -> length = strlen (s -> tool);
          vstr_put_tail (vstr, s -> tool, strlen (s -> tool));
          break;
        case RTCP_SDES_NOTE:
          sdes_item -> length = strlen (s -> note);
          vstr_put_tail (vstr, s -> note, strlen (s -> note));
          break;
      }
    }
  }

  pthread_mutex_unlock (&s -> guard);

  /* end chunk */
  vstr_zero_tail (vstr, sizeof(uint32_t) - vstr -> size % sizeof(uint32_t));

  rtcp_packet -> len = vstr -> size / 4 - 1;               /* packet length   */

  return;
}

/* Parses received RTCP packet, checks for validity and updates RTP session */
int rtcp_recv (rtp_t *s, vstr_t *packet) {

  rtcp_packet_t *rtcp_packet = (rtcp_packet_t *) packet -> head;
  rtp_source_t *member;
  uint8_t rlen;
  uint32_t *ssrc;
  int i;

  while (packet -> size > 0) {

    /* check RTCP header */
    if (rtcp_packet -> ver != RTP_VERSION)
      return (-1);

    vstr_adv_head (packet, sizeof(uint32_t));

    /* parse RTCP content */
    switch (rtcp_packet -> pt) {
      case RTCP_SR:

        ssrc = (uint32_t *) packet -> head;
        *ssrc = htonl (*ssrc);
        pthread_mutex_lock (&s -> guard);
        member = rtp_find_member (&s -> members, *ssrc);

        /* update data */
        if (member != NULL) {
          pthread_mutex_unlock (&s -> guard);
          pthread_mutex_lock (&member -> guard);
          rtp_update_sr (member, packet);
          pthread_mutex_unlock (&member -> guard);
        }
        else {
          member = rtp_new_member (*ssrc);
          rtp_update_sr (member, packet);
          rtp_add_member (&s -> members, member);
          pthread_mutex_unlock (&s -> guard);
        }
        vstr_adv_head (packet, RTCP_SR_SIZE);

      case RTCP_RR:
        for (i = 0; i < rtcp_packet -> count; i++) {

          ssrc = (uint32_t *) packet -> head;
          *ssrc = htonl (*ssrc);
          pthread_mutex_lock (&s -> guard);
          member = rtp_find_member (&s -> members, *ssrc);
          pthread_mutex_unlock (&s -> guard);

          /* update data */
          if (member != NULL) {
            pthread_mutex_lock (&member -> guard);
            rtp_update_rr (member, packet);
            pthread_mutex_unlock (&member -> guard);
          }

          vstr_adv_head (packet, RTCP_RR_SIZE);
        }
        break;

      case RTCP_SDES:
        for (i = 0; i < rtcp_packet -> count; i++) {

          ssrc = (uint32_t *) packet -> head;
          *ssrc = htonl (*ssrc);
          pthread_mutex_lock (&s -> guard);
          member = rtp_find_member (&s -> members, *ssrc);
          pthread_mutex_unlock (&s -> guard);

          vstr_adv_head (packet, sizeof(uint32_t));

          /* update data */
          if (member != NULL) {
            pthread_mutex_lock (&member -> guard);
            rtp_update_sdes (member, packet);
            pthread_mutex_unlock (&member -> guard);
          }
          else
            rtp_discard_sdes (packet);
        }
        break;

      case RTCP_BYE:
        for (i = 0; i < rtcp_packet -> count; i++) {

          ssrc = (uint32_t *) packet -> head;
          *ssrc = htonl (*ssrc);
          vstr_adv_head (packet, sizeof(uint32_t));
          pthread_mutex_lock (&s -> guard);
          member = rtp_find_member (&s -> members, *ssrc);

          /* remove members */
          if (member != NULL)
            rtp_remove_member (&s -> members, *ssrc);

          pthread_mutex_unlock (&s -> guard);
        }

        if (rtcp_packet -> len > rtcp_packet -> count * sizeof(uint32_t)) {
          vstr_get_head (packet, &rlen, sizeof(uint8_t));
          vstr_adv_head (packet, rlen);
        }
        break;

      case RTCP_APP:

      default:
        return (-1);
    }

    rtcp_packet = (rtcp_packet_t *) packet -> head;
  }

  return (0);
}