rtcp.cpp

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

/**********
This library is free software; you can redistribute it and/or modify it under
the terms of the GNU Lesser General Public License as published by the
Free Software Foundation; either version 2.1 of the License, or (at your
option) any later version. (See <http://www.gnu.org/copyleft/lesser.html>.)

This library is distributed in the hope that it will be useful, but WITHOUT
ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
FOR A PARTICULAR PURPOSE.  See the GNU Lesser General Public License for
more details.

You should have received a copy of the GNU Lesser General Public License
along with this library; if not, write to the Free Software Foundation, Inc.,
59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
**********/
// "liveMedia"
// Copyright (c) 1996-2004 Live Networks, Inc.  All rights reserved.
// RTCP
// Implementation

#include "RTCP.hh"
#include "GroupsockHelper.hh"
#include "rtcp_from_spec.h"

////////// RTCPMemberDatabase //////////

class RTCPMemberDatabase {
public:
  RTCPMemberDatabase(RTCPInstance& ourRTCPInstance)
    : fOurRTCPInstance(ourRTCPInstance), fNumMembers(1 /*ourself*/),
      fTable(HashTable::create(ONE_WORD_HASH_KEYS)) {
  }

  virtual ~RTCPMemberDatabase() {
	delete fTable;
  }

  Boolean isMember(unsigned ssrc) const {
    return fTable->Lookup((char*)(long)ssrc) != NULL;
  }

  Boolean noteMembership(unsigned ssrc, unsigned curTimeCount) {
    Boolean isNew = !isMember(ssrc);

    if (isNew) {
      ++fNumMembers;
    }

    // Record the current time, so we can age stale members
    fTable->Add((char*)(long)ssrc, (void*)(long)curTimeCount);

    return isNew;
  }

  Boolean remove(unsigned ssrc) {
    Boolean wasPresent = fTable->Remove((char*)(long)ssrc);
    if (wasPresent) {
      --fNumMembers;
    }
    return wasPresent;
  }

  unsigned numMembers() const {
    return fNumMembers;
  }

  void reapOldMembers(unsigned threshold);

private:
  RTCPInstance& fOurRTCPInstance;
  unsigned fNumMembers;
  HashTable* fTable;
};

void RTCPMemberDatabase::reapOldMembers(unsigned threshold) {
  Boolean foundOldMember;
  unsigned oldSSRC = 0;

  do {
    foundOldMember = False;

    HashTable::Iterator* iter
      = HashTable::Iterator::create(*fTable);
    unsigned long timeCount;
    char const* key;
    while ((timeCount = (unsigned long)(iter->next(key))) != 0) {
#ifdef DEBUG
      fprintf(stderr, "reap: checking SSRC 0x%lx: %ld (threshold %d)\n", (unsigned long)key, timeCount, threshold);
#endif
      if (timeCount < (unsigned long)threshold) { // this SSRC is old
        unsigned long ssrc = (unsigned long)key;
        oldSSRC = (unsigned)ssrc;
        foundOldMember = True;
      }
    }
    delete iter;

    if (foundOldMember) {
#ifdef DEBUG
        fprintf(stderr, "reap: removing SSRC 0x%x\n", oldSSRC);
#endif
      fOurRTCPInstance.removeSSRC(oldSSRC);
    }
  } while (foundOldMember);
}


////////// RTCPInstance //////////

static double dTimeNow() {
    struct timeval timeNow;
    gettimeofday(&timeNow, NULL);
    return (double) (timeNow.tv_sec + timeNow.tv_usec/1000000.0);
}

static unsigned const maxPacketSize = 1450;
	// bytes (1500, minus some allowance for IP, UDP, UMTP headers)
static unsigned const preferredPacketSize = 1000; // bytes

RTCPInstance::RTCPInstance(UsageEnvironment& env, Groupsock* RTCPgs,
			   unsigned totSessionBW,
			   unsigned char const* cname,
			   RTPSink* sink, RTPSource const* source,
			   Boolean isSSMSource)
  : Medium(env), fRTCPInterface(this, RTCPgs), fTotSessionBW(totSessionBW),
    fSink(sink), fSource(source), fIsSSMSource(isSSMSource),
    fCNAME(RTCP_SDES_CNAME, cname), fOutgoingReportCount(1),
    fAveRTCPSize(0), fIsInitial(1), fPrevNumMembers(0),
    fLastSentSize(0), fLastReceivedSize(0), fLastReceivedSSRC(0),
    fTypeOfEvent(EVENT_UNKNOWN), fTypeOfPacket(PACKET_UNKNOWN_TYPE),
    fHaveJustSentPacket(False), fLastPacketSentSize(0),
    fByeHandlerTask(NULL), fByeHandlerClientData(NULL) {
#ifdef DEBUG
  fprintf(stderr, "RTCPInstance[%p]::RTCPInstance()\n", this);
#endif
  if (isSSMSource) RTCPgs->multicastSendOnly(); // don't receive multicast
    
  double timeNow = dTimeNow();
  fPrevReportTime = fNextReportTime = timeNow;

  fKnownMembers = new RTCPMemberDatabase(*this);
  fInBuf = new unsigned char[maxPacketSize];
  if (fKnownMembers == NULL || fInBuf == NULL) return;

  // A hack to save buffer space, because RTCP packets are always small:
  unsigned savedMaxSize = OutPacketBuffer::maxSize;
  OutPacketBuffer::maxSize = maxPacketSize;
  fOutBuf = new OutPacketBuffer(preferredPacketSize, maxPacketSize);
  OutPacketBuffer::maxSize = savedMaxSize;
  if (fOutBuf == NULL) return;

  // Arrange to handle incoming reports from others:
  TaskScheduler::BackgroundHandlerProc* handler
    = (TaskScheduler::BackgroundHandlerProc*)&incomingReportHandler;
  fRTCPInterface.startNetworkReading(handler);
  
  // Send our first report.
  fTypeOfEvent = EVENT_REPORT;
  onExpire(this);
}

RTCPInstance::~RTCPInstance() {
#ifdef DEBUG
  fprintf(stderr, "RTCPInstance[%p]::~RTCPInstance()\n", this);
#endif
  // Turn off background read handling:
  fRTCPInterface.stopNetworkReading();

  // Begin by sending a BYE.  We have to do this immediately, without
  // 'reconsideration', because "this" is going away.
  fTypeOfEvent = EVENT_BYE; // not used, but...
  sendBYE();

  delete fKnownMembers;
  delete fOutBuf;
  delete[] fInBuf;
}

RTCPInstance* RTCPInstance::createNew(UsageEnvironment& env, Groupsock* RTCPgs,
				      unsigned totSessionBW,
				      unsigned char const* cname,
				      RTPSink* sink, RTPSource const* source,
				      Boolean isSSMSource) {
  return new RTCPInstance(env, RTCPgs, totSessionBW, cname, sink, source,
			  isSSMSource);
}

Boolean RTCPInstance::lookupByName(UsageEnvironment& env,
				   char const* instanceName,
				   RTCPInstance*& resultInstance) {
  resultInstance = NULL; // unless we succeed

  Medium* medium;
  if (!Medium::lookupByName(env, instanceName, medium)) return False;

  if (!medium->isRTCPInstance()) {
    env.setResultMsg(instanceName, " is not a RTCP instance");
    return False;
  }

  resultInstance = (RTCPInstance*)medium;
  return True;
}

Boolean RTCPInstance::isRTCPInstance() const {
  return True;
}

unsigned RTCPInstance::numMembers() const {
  if (fKnownMembers == NULL) return 0;

  return fKnownMembers->numMembers();
}

void RTCPInstance::setByeHandler(TaskFunc* handlerTask, void* clientData,
				 Boolean handleActiveParticipantsOnly) {
  fByeHandlerTask = handlerTask;
  fByeHandlerClientData = clientData;
  fByeHandleActiveParticipantsOnly = handleActiveParticipantsOnly;
}

void RTCPInstance::setStreamSocket(int sockNum,
				   unsigned char streamChannelId) {
  // Turn off background read handling:
  fRTCPInterface.stopNetworkReading();

  // Switch to RTCP-over-TCP:
  fRTCPInterface.setStreamSocket(sockNum, streamChannelId);

  // Turn background reading back on:
  TaskScheduler::BackgroundHandlerProc* handler
    = (TaskScheduler::BackgroundHandlerProc*)&incomingReportHandler;
  fRTCPInterface.startNetworkReading(handler);
}

void RTCPInstance::addStreamSocket(int sockNum,
				   unsigned char streamChannelId) {
  // Add the RTCP-over-TCP interface:
  fRTCPInterface.setStreamSocket(sockNum, streamChannelId);

  // Turn on background reading for this socket (in case it's not on already):
  TaskScheduler::BackgroundHandlerProc* handler
    = (TaskScheduler::BackgroundHandlerProc*)&incomingReportHandler;
  fRTCPInterface.startNetworkReading(handler);
}

static unsigned const IP_UDP_HDR_SIZE = 28;
    // overhead (bytes) of IP and UDP hdrs

#define ADVANCE(n) pkt += (n); packetSize -= (n)

void RTCPInstance::incomingReportHandler(RTCPInstance* instance,
					 int /*mask*/) {
  instance->incomingReportHandler1();
}

void RTCPInstance::incomingReportHandler1() {
  unsigned char* pkt = fInBuf;
  unsigned packetSize;
  struct sockaddr_in fromAddress;
  int typeOfPacket = PACKET_UNKNOWN_TYPE;

  do {
    if (!fRTCPInterface.handleRead(pkt, maxPacketSize,
				   packetSize, fromAddress)) {
      break;
    }

    // Ignore the packet if it was looped-back from ourself:
    if (RTCPgs()->wasLoopedBackFromUs(envir(), fromAddress)) {
      // However, we still want to handle incoming RTCP packets from
      // *other processes* on the same machine.  To distinguish this
      // case from a true loop-back, check whether we've just sent a
      // packet of the same size.  (This check isn't perfect, but it seems
      // to be the best we can do.)
      if (fHaveJustSentPacket && fLastPacketSentSize == packetSize) {
	// This is a true loop-back:
	fHaveJustSentPacket = False;
	break; // ignore this packet
      }
    }

    if (fIsSSMSource) {
      // This packet was received via unicast.  'Reflect' it by resending
      // it to the multicast group.
      // NOTE: Denial-of-service attacks are possible here.
      // Users of this software may wish to add their own,
      // application-specific mechanism for 'authenticating' the
      // validity of this packet before relecting it.
      fRTCPInterface.sendPacket(pkt, packetSize);
      fHaveJustSentPacket = True;
      fLastPacketSentSize = packetSize;
    }

#ifdef DEBUG
    fprintf(stderr, "[%p]saw incoming RTCP packet (from address %s, port %d)\n", this, our_inet_ntoa(fromAddress.sin_addr), ntohs(fromAddress.sin_port));
    unsigned char* p = pkt;
    for (unsigned i = 0; i < packetSize; ++i) {
      if (i%4 == 0) fprintf(stderr, " ");
      fprintf(stderr, "%02x", p[i]);
    }
    fprintf(stderr, "\n");
#endif
    int totPacketSize = IP_UDP_HDR_SIZE + packetSize;

    // Check the RTCP packet for validity:
    // It must at least contain a header (4 bytes), and this header
    // must be version=2, with no padding bit, and a payload type of
    // SR (200) or RR (201):
    if (packetSize < 4) break;
    unsigned rtcpHdr = ntohl(*(unsigned*)pkt);
    if ((rtcpHdr & 0xE0FE0000) != (0x80000000 | (RTCP_PT_SR<<16))) {
#ifdef DEBUG
      fprintf(stderr, "rejected bad RTCP packet: header 0x%08x\n", rtcpHdr);
#endif
      break;
    }

    // Process each of the individual RTCP 'subpackets' in (what may be)
    // a compound RTCP packet.
    unsigned reportSenderSSRC = 0;
    Boolean packetOK = False;
    while (1) {
      unsigned rc = (rtcpHdr>>24)&0x1F;
      unsigned pt = (rtcpHdr>>16)&0xFF;
      unsigned length = 4*(rtcpHdr&0xFFFF); // doesn't count hdr
      ADVANCE(4); // skip over the header
      if (length > packetSize) break;

      // Assume that each RTCP subpacket begins with a 4-byte SSRC:
      if (length < 4) break; length -= 4;
      reportSenderSSRC = ntohl(*(unsigned*)pkt); ADVANCE(4);

      Boolean subPacketOK = False;
      switch (pt) {
        case RTCP_PT_SR: {
#ifdef DEBUG
	  fprintf(stderr, "SR\n");
#endif
	  if (length < 20) break; length -= 20;

	  // Extract the NTP timestamp, and note this:
	  unsigned NTPmsw = ntohl(*(unsigned*)pkt); ADVANCE(4);
	  unsigned NTPlsw = ntohl(*(unsigned*)pkt); ADVANCE(4);
	  unsigned rtpTimestamp = ntohl(*(unsigned*)pkt); ADVANCE(4);
	  if (fSource != NULL) {
	    RTPReceptionStatsDB& receptionStats
	      = fSource->receptionStatsDB();
	    receptionStats.noteIncomingSR(reportSenderSSRC,
					  NTPmsw, NTPlsw, rtpTimestamp);
	  }
	  ADVANCE(8); // skip over packet count, octet count
	  // The rest of the SR is handled like a RR (so, no "break;" here)
	}
        case RTCP_PT_RR: {
#ifdef DEBUG
	  fprintf(stderr, "RR\n");
#endif
	  unsigned reportBlocksSize = rc*(6*4);
	  if (length < reportBlocksSize) break;
	  length -= reportBlocksSize;

          if (fSink != NULL) {
	    // Use this information to update stats about our transmissions:
            RTPTransmissionStatsDB& transmissionStats = fSink->transmissionStatsDB();
            for (unsigned i = 0; i < rc; ++i) {
              unsigned senderSSRC = ntohl(*(unsigned*)pkt); ADVANCE(4);
              // We care only about reports about our own transmission, not others'
              if (senderSSRC == fSink->SSRC()) {
                unsigned lossStats = ntohl(*(unsigned*)pkt); ADVANCE(4);
                unsigned highestReceived = ntohl(*(unsigned*)pkt); ADVANCE(4);
                unsigned jitter = ntohl(*(unsigned*)pkt); ADVANCE(4);
                unsigned timeLastSR = ntohl(*(unsigned*)pkt); ADVANCE(4);
                unsigned timeSinceLastSR = ntohl(*(unsigned*)pkt); ADVANCE(4);
                transmissionStats.noteIncomingRR(reportSenderSSRC, fromAddress,
						 lossStats,
						 highestReceived, jitter,
						 timeLastSR, timeSinceLastSR);
              } else {
                ADVANCE(4*5);
              }
            }
          } else {
            ADVANCE(reportBlocksSize);
          }

	  subPacketOK = True;
	  typeOfPacket = PACKET_RTCP_REPORT;
	  break;
	}
        case RTCP_PT_BYE: {
#ifdef DEBUG
	  fprintf(stderr, "BYE\n");
#endif
	  // If a 'BYE handler' was set, call it now:
	  TaskFunc* byeHandler = fByeHandlerTask;
	  if (byeHandler != NULL
	      && (!fByeHandleActiveParticipantsOnly
		  || (fSource != NULL
		      && fSource->receptionStatsDB().lookup(reportSenderSSRC) != NULL)
		  || (fSink != NULL
		      && fSink->transmissionStatsDB().lookup(reportSenderSSRC) != NULL))) {
	    fByeHandlerTask = NULL;
	        // we call this only once by default 
	    (*byeHandler)(fByeHandlerClientData);
	  }

	  // We should really check for & handle >1 SSRCs being present #####

	  subPacketOK = True;
	  typeOfPacket = PACKET_BYE;
	  break;
	}
	// Later handle SDES, APP, and compound RTCP packets #####
        default:
#ifdef DEBUG
	  fprintf(stderr, "UNSUPPORTED TYPE(0x%x)\n", pt);
#endif
	  subPacketOK = True;
	  break;
      }  
      if (!subPacketOK) break;

      // need to check for (& handle) SSRC collision! #####

#ifdef DEBUG
      fprintf(stderr, "validated RTCP subpacket (type %d): %d, %d, %d, 0x%08x\n", typeOfPacket, rc, pt, length, reportSenderSSRC);
#endif
      
      // Skip over any remaining bytes in this subpacket:
      ADVANCE(length);

      // Check whether another RTCP 'subpacket' follows:
      if (packetSize == 0) {
	packetOK = True;
	break;
      } else if (packetSize < 4) {
#ifdef DEBUG
	fprintf(stderr, "extraneous %d bytes at end of RTCP packet!\n", packetSize);
#endif
	break;
      }
      rtcpHdr = ntohl(*(unsigned*)pkt);
      if ((rtcpHdr & 0xC0000000) != 0x80000000) {
#ifdef DEBUG
	fprintf(stderr, "bad RTCP subpacket: header 0x%08x\n", rtcpHdr);
#endif
	break;