rtcp.cpp

流媒体传输协议的实现代码,非常有用.可以支持rtsp mms等流媒体传输协议

      }
    }
      
    if (!packetOK) {
#ifdef DEBUG
      fprintf(stderr, "rejected bad RTCP subpacket: header 0x%08x\n", rtcpHdr);
#endif
      break;
    } else {
#ifdef DEBUG
      fprintf(stderr, "validated entire RTCP packet\n");
#endif
    }
      
    onReceive(typeOfPacket, totPacketSize, reportSenderSSRC);
  } while (0);
}

void RTCPInstance::onReceive(int typeOfPacket, int totPacketSize,
			     unsigned ssrc) {
  fTypeOfPacket = typeOfPacket;
  fLastReceivedSize = totPacketSize;
  fLastReceivedSSRC = ssrc;

  int members = (int)numMembers();
  int senders = (fSink != NULL) ? 1 : 0;

  OnReceive(this, // p
	    this, // e
	    &members, // members
	    &fPrevNumMembers, // pmembers
	    &senders, // senders
	    &fAveRTCPSize, // avg_rtcp_size
	    &fPrevReportTime, // tp
	    dTimeNow(), // tc
	    fNextReportTime);
}

void RTCPInstance::sendReport() {
#ifdef DEBUG
  fprintf(stderr, "sending REPORT\n");
#endif
  // Begin by including a SR and/or RR report:
  addReport();

  // Then, include a SDES:
  addSDES();

  // Send the report:
  sendBuiltPacket();

  // Periodically clean out old members from our SSRC membership database:
  const unsigned membershipReapPeriod = 5;
  if ((++fOutgoingReportCount) % membershipReapPeriod == 0) {
    unsigned threshold = fOutgoingReportCount - membershipReapPeriod;
    fKnownMembers->reapOldMembers(threshold);
  }
}

void RTCPInstance::sendBYE() {
#ifdef DEBUG
  fprintf(stderr, "sending BYE\n");
#endif
  // The packet must begin with a SR and/or RR report:
  addReport();

  addBYE();
  sendBuiltPacket();
}

void RTCPInstance::sendBuiltPacket() {
#ifdef DEBUG
  fprintf(stderr, "sending RTCP packet\n");
  unsigned char* p = fOutBuf->packet();
  for (unsigned i = 0; i < fOutBuf->curPacketSize(); ++i) {
    if (i%4 == 0) fprintf(stderr," ");
    fprintf(stderr, "%02x", p[i]);
  }
  fprintf(stderr, "\n");
#endif
  unsigned reportSize = fOutBuf->curPacketSize();
  fRTCPInterface.sendPacket(fOutBuf->packet(), reportSize);
  fOutBuf->resetOffset();

  fLastSentSize = IP_UDP_HDR_SIZE + reportSize;
  fHaveJustSentPacket = True;
  fLastPacketSentSize = reportSize;
}

int RTCPInstance::checkNewSSRC() {
  return fKnownMembers->noteMembership(fLastReceivedSSRC,
				       fOutgoingReportCount);
}

void RTCPInstance::removeLastReceivedSSRC() {
  removeSSRC(fLastReceivedSSRC);
}

void RTCPInstance::removeSSRC(u_int32_t ssrc) {
  fKnownMembers->remove(ssrc);

  // Also, remove records of this SSRC from any reception or transmission stats
  if (fSource != NULL) fSource->receptionStatsDB().removeRecord(ssrc);
  if (fSink != NULL) fSink->transmissionStatsDB().removeRecord(ssrc);
}

void RTCPInstance::onExpire(RTCPInstance* instance) {
  instance->onExpire1();
}

// Member functions to build specific kinds of report:

void RTCPInstance::addReport() {
  // Include a SR or a RR, depending on whether we
  // have an associated sink or source:
  if (fSink != NULL) {
    addSR();
  } else if (fSource != NULL) {
    addRR();
  }
}

void RTCPInstance::addSR() {
  // ASSERT: fSink != NULL

  enqueueCommonReportPrefix(RTCP_PT_SR, fSink->SSRC(),
			    5 /* extra words in a SR */);

  // Now, add the 'sender info' for our sink

  // Insert the NTP and RTP timestamps for the 'wallclock time':
  struct timeval timeNow;
  gettimeofday(&timeNow, NULL);
  fOutBuf->enqueueWord(timeNow.tv_sec + 0x83AA7E80);
      // NTP timestamp most-significant word (1970 epoch -> 1900 epoch)
  double fractionalPart = (timeNow.tv_usec/15625.0)*0x04000000; // 2^32/10^6
  fOutBuf->enqueueWord((unsigned)(fractionalPart+0.5));
      // NTP timestamp least-significant word
  unsigned rtpTimestamp = fSink->convertToRTPTimestamp(timeNow);
  fOutBuf->enqueueWord(rtpTimestamp); // RTP ts

  // Insert the packet and byte counts:
  fOutBuf->enqueueWord(fSink->packetCount());
  fOutBuf->enqueueWord(fSink->octetCount());

  enqueueCommonReportSuffix();
}

void RTCPInstance::addRR() {
  // ASSERT: fSource != NULL

  enqueueCommonReportPrefix(RTCP_PT_RR, fSource->SSRC());
  enqueueCommonReportSuffix();
}

void RTCPInstance::enqueueCommonReportPrefix(unsigned char packetType,
					     unsigned SSRC,
					     unsigned numExtraWords) {
  unsigned numReportingSources;
  if (fSource == NULL) {
    numReportingSources = 0; // we don't receive anything
  } else {
    RTPReceptionStatsDB& allReceptionStats
      = fSource->receptionStatsDB();
    numReportingSources = allReceptionStats.numActiveSourcesSinceLastReset();
    // This must be <32, to fit in 5 bits:
    if (numReportingSources >= 32) { numReportingSources = 32; }
    // Later: support adding more reports to handle >32 sources (unlikely)#####
  }

  unsigned rtcpHdr = 0x80000000; // version 2, no padding
  rtcpHdr |= (numReportingSources<<24);
  rtcpHdr |= (packetType<<16);
  rtcpHdr |= (1 + numExtraWords + 6*numReportingSources);
      // each report block is 6 32-bit words long
  fOutBuf->enqueueWord(rtcpHdr);

  fOutBuf->enqueueWord(SSRC);
}

void RTCPInstance::enqueueCommonReportSuffix() {
  // Output the report blocks for each source:
  if (fSource != NULL) { 
    RTPReceptionStatsDB& allReceptionStats
      = fSource->receptionStatsDB();

    RTPReceptionStatsDB::Iterator iterator(allReceptionStats);
    while (1) {
      RTPReceptionStats* receptionStats = iterator.next();
      if (receptionStats == NULL) break;
      enqueueReportBlock(receptionStats);
    }

    allReceptionStats.reset(); // because we have just generated a report
  }
}

void
RTCPInstance::enqueueReportBlock(RTPReceptionStats* stats) {
  fOutBuf->enqueueWord(stats->SSRC());

  unsigned highestExtSeqNumReceived = stats->highestExtSeqNumReceived();

  unsigned totNumExpected
    = highestExtSeqNumReceived - stats->baseExtSeqNumReceived();
  int totNumLost = totNumExpected - stats->totNumPacketsReceived();
  // 'Clamp' this loss number to a 24-bit signed value:
  if (totNumLost > 0x007FFFFF) {
    totNumLost = 0x007FFFFF;
  } else if (totNumLost < 0) {
    if (totNumLost < -0x00800000) totNumLost = 0x00800000; // unlikely, but...
    totNumLost &= 0x00FFFFFF;
  }

  unsigned numExpectedSinceLastReset
    = highestExtSeqNumReceived - stats->lastResetExtSeqNumReceived();
  int numLostSinceLastReset
    = numExpectedSinceLastReset - stats->numPacketsReceivedSinceLastReset(); 
  unsigned char lossFraction;
  if (numExpectedSinceLastReset == 0 || numLostSinceLastReset < 0) {
    lossFraction = 0;
  } else {
    lossFraction = (unsigned char)
      ((numLostSinceLastReset << 8) / numExpectedSinceLastReset);
  }
  
  fOutBuf->enqueueWord((lossFraction<<24) | totNumLost);
  fOutBuf->enqueueWord(highestExtSeqNumReceived);

  fOutBuf->enqueueWord(stats->jitter());

  unsigned NTPmsw = stats->lastReceivedSR_NTPmsw();
  unsigned NTPlsw = stats->lastReceivedSR_NTPlsw();
  unsigned LSR = ((NTPmsw&0xFFFF)<<16)|(NTPlsw>>16); // middle 32 bits
  fOutBuf->enqueueWord(LSR);

  // Figure out how long has elapsed since the last SR rcvd from this src:
  struct timeval const& LSRtime = stats->lastReceivedSR_time(); // "last SR"
  struct timeval timeNow, timeSinceLSR;
  gettimeofday(&timeNow, NULL);
  if (timeNow.tv_usec < LSRtime.tv_usec) {
    timeNow.tv_usec += 1000000;
    timeNow.tv_sec -= 1;
  }
  timeSinceLSR.tv_sec = timeNow.tv_sec - LSRtime.tv_sec;
  timeSinceLSR.tv_usec = timeNow.tv_usec - LSRtime.tv_usec;
  // The enqueued time is in units of 1/65536 seconds.
  // (Note that 65536/1000000 == 1024/15625) 
  unsigned DLSR;
  if (LSR == 0) {
    DLSR = 0;
  } else {
    DLSR = (timeSinceLSR.tv_sec<<16)
         | ( (((timeSinceLSR.tv_usec<<11)+15625)/31250) & 0xFFFF);
  }
  fOutBuf->enqueueWord(DLSR);
}

void RTCPInstance::addSDES() {
  // For now we support only the CNAME item; later support more #####

  // Begin by figuring out the size of the entire SDES report:
  unsigned numBytes = 4;
      // counts the SSRC, but not the header; it'll get subtracted out
  numBytes += fCNAME.totalSize(); // includes id and length
  numBytes += 1; // the special END item

  unsigned num4ByteWords = (numBytes + 3)/4;

  unsigned rtcpHdr = 0x81000000; // version 2, no padding, 1 SSRC chunk
  rtcpHdr |= (RTCP_PT_SDES<<16);
  rtcpHdr |= num4ByteWords;
  fOutBuf->enqueueWord(rtcpHdr);

  if (fSource != NULL) {
    fOutBuf->enqueueWord(fSource->SSRC());
  } else if (fSink != NULL) {
    fOutBuf->enqueueWord(fSink->SSRC());
  }

  // Add the CNAME:
  fOutBuf->enqueue(fCNAME.data(), fCNAME.totalSize());

  // Add the 'END' item (i.e., a zero byte), plus any more needed to pad:
  unsigned numPaddingBytesNeeded = 4 - (fOutBuf->curPacketSize() % 4);
  unsigned char const zero = '\0';
  while (numPaddingBytesNeeded-- > 0) fOutBuf->enqueue(&zero, 1);
}

void RTCPInstance::addBYE() {
  unsigned rtcpHdr = 0x81000000; // version 2, no padding, 1 SSRC
  rtcpHdr |= (RTCP_PT_BYE<<16);
  rtcpHdr |= 1; // 2 32-bit words total (i.e., with 1 SSRC)
  fOutBuf->enqueueWord(rtcpHdr);

  if (fSource != NULL) {
    fOutBuf->enqueueWord(fSource->SSRC());
  } else if (fSink != NULL) {
    fOutBuf->enqueueWord(fSink->SSRC());
  }
}

void RTCPInstance::schedule(double nextTime) {
  fNextReportTime = nextTime;

  double secondsToDelay = nextTime - dTimeNow();
#ifdef DEBUG
  fprintf(stderr, "schedule(%f->%f)\n", secondsToDelay, nextTime);
#endif
  int usToGo = (int)(secondsToDelay * 1000000);
  nextTask() = envir().taskScheduler().scheduleDelayedTask(usToGo,
				(TaskFunc*)RTCPInstance::onExpire, this);
}

void RTCPInstance::reschedule(double nextTime) {
  envir().taskScheduler().unscheduleDelayedTask(nextTask());
  schedule(nextTime);
}

void RTCPInstance::onExpire1() {
  // Note: fTotSessionBW is kbits per second
  double rtcpBW = 0.05*fTotSessionBW*1024/8; // -> bytes per second

  OnExpire(this, // event
	   numMembers(), // members
	   (fSink != NULL) ? 1 : 0, // senders
	   rtcpBW, // rtcp_bw
	   (fSink != NULL) ? 1 : 0, // we_sent
	   &fAveRTCPSize, // ave_rtcp_size
	   &fIsInitial, // initial
	   dTimeNow(), // tc
	   &fPrevReportTime, // tp
	   &fPrevNumMembers // pmembers
	   );
}

////////// SDESItem //////////

SDESItem::SDESItem(unsigned char tag, unsigned char const* value) {
  unsigned length = strlen((char const*)value);
  if (length > 511) length = 511;

  fData[0] = tag;
  fData[1] = (unsigned char)length;
  memmove(&fData[2], value, length);

  // Pad the trailing bytes to a 4-byte boundary:
  while ((length)%4 > 0) fData[2 + length++] = '\0';
}

unsigned SDESItem::totalSize() const {
  return 2 + (unsigned)fData[1];
}


////////// Implementation of routines imported by the "rtcp_from_spec" C code

extern "C" void Schedule(double nextTime, event e) {
  RTCPInstance* instance = (RTCPInstance*)e;
  if (instance == NULL) return;

  instance->schedule(nextTime);
}

extern "C" void Reschedule(double nextTime, event e) {
  RTCPInstance* instance = (RTCPInstance*)e;
  if (instance == NULL) return;

  instance->reschedule(nextTime);
}

extern "C" void SendRTCPReport(event e) {
  RTCPInstance* instance = (RTCPInstance*)e;
  if (instance == NULL) return;

  instance->sendReport();
}

extern "C" void SendBYEPacket(event e) {
  RTCPInstance* instance = (RTCPInstance*)e;
  if (instance == NULL) return;

  instance->sendBYE();
}

extern "C" int TypeOfEvent(event e) {
  RTCPInstance* instance = (RTCPInstance*)e;
  if (instance == NULL) return EVENT_UNKNOWN;

  return instance->typeOfEvent();
}

extern "C" int SentPacketSize(event e) {
  RTCPInstance* instance = (RTCPInstance*)e;
  if (instance == NULL) return 0;

  return instance->sentPacketSize();
}

extern "C" int PacketType(packet p) {
  RTCPInstance* instance = (RTCPInstance*)p;
  if (instance == NULL) return PACKET_UNKNOWN_TYPE;

  return instance->packetType();
}

extern "C" int ReceivedPacketSize(packet p) {
  RTCPInstance* instance = (RTCPInstance*)p;
  if (instance == NULL) return 0;

  return instance->receivedPacketSize();
}

extern "C" int NewMember(packet p) {
  RTCPInstance* instance = (RTCPInstance*)p;
  if (instance == NULL) return 0;

  return instance->checkNewSSRC();
}

extern "C" int NewSender(packet /*p*/) {
  return 0; // we don't yet recognize senders other than ourselves #####
}

extern "C" void AddMember(packet /*p*/) {
  // Do nothing; all of the real work was done when NewMember() was called
}

extern "C" void AddSender(packet /*p*/) {
  // we don't yet recognize senders other than ourselves #####
}

extern "C" void RemoveMember(packet p) {
  RTCPInstance* instance = (RTCPInstance*)p;
  if (instance == NULL) return;

  instance->removeLastReceivedSSRC();
}

extern "C" void RemoveSender(packet /*p*/) {
  // we don't yet recognize senders other than ourselves #####
}

extern "C" double drand30() {
  unsigned tmp = our_random()&0x3FFFFFFF; // a random 30-bit integer
  return tmp/(double)(1024*1024*1024);
}