rtcp.cpp

H.264 RTSP 串流(live 555)視窗版本

/**********
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.,
51 Franklin Street, Fifth Floor, Boston, MA 02110-1301  USA
**********/
// "liveMedia"
// Copyright (c) 1996-2010 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, True);
    }
  } 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),
    fSRHandlerTask(NULL), fSRHandlerClientData(NULL),
    fRRHandlerTask(NULL), fRRHandlerClientData(NULL),
    fSpecificRRHandlerTable(NULL) {
#ifdef DEBUG
  fprintf(stderr, "RTCPInstance[%p]::RTCPInstance()\n", this);
#endif
  if (fTotSessionBW == 0) { // not allowed!
    env << "RTCPInstance::RTCPInstance error: totSessionBW parameter should not be zero!\n";
    fTotSessionBW = 1;
  }

  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;
  fNumBytesAlreadyRead = 0;

  // 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);
}

struct RRHandlerRecord {
  TaskFunc* rrHandlerTask;
  void* rrHandlerClientData;
};

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();

  if (fSpecificRRHandlerTable != NULL) {
    AddressPortLookupTable::Iterator iter(*fSpecificRRHandlerTable);
    RRHandlerRecord* rrHandler;
    while ((rrHandler = (RRHandlerRecord*)iter.next()) != NULL) {
      delete rrHandler;
    }
    delete fSpecificRRHandlerTable;
  }

  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::setSRHandler(TaskFunc* handlerTask, void* clientData) {
  fSRHandlerTask = handlerTask;
  fSRHandlerClientData = clientData;
}

void RTCPInstance::setRRHandler(TaskFunc* handlerTask, void* clientData) {
  fRRHandlerTask = handlerTask;
  fRRHandlerClientData = clientData;
}

void RTCPInstance
::setSpecificRRHandler(netAddressBits fromAddress, Port fromPort,
		       TaskFunc* handlerTask, void* clientData) {
  if (handlerTask == NULL && clientData == NULL) {
    unsetSpecificRRHandler(fromAddress, fromPort);
    return;
  }

  RRHandlerRecord* rrHandler = new RRHandlerRecord;
  rrHandler->rrHandlerTask = handlerTask;
  rrHandler->rrHandlerClientData = clientData;
  if (fSpecificRRHandlerTable == NULL) {
    fSpecificRRHandlerTable = new AddressPortLookupTable;
  }
  fSpecificRRHandlerTable->Add(fromAddress, (~0), fromPort, rrHandler);
}

void RTCPInstance
::unsetSpecificRRHandler(netAddressBits fromAddress, Port fromPort) {
  if (fSpecificRRHandlerTable == NULL) return;

  RRHandlerRecord* rrHandler
    = (RRHandlerRecord*)(fSpecificRRHandlerTable->Lookup(fromAddress, (~0), fromPort));
  if (rrHandler != NULL) {
    fSpecificRRHandlerTable->Remove(fromAddress, (~0), fromPort);
    delete rrHandler;
  }
}

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) {
  // First, turn off background read handling for the default (UDP) socket:
  fRTCPInterface.stopNetworkReading();

  // Add the RTCP-over-TCP interface:
  fRTCPInterface.addStreamSocket(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() {
  do {
    int tcpReadStreamSocketNum = fRTCPInterface.nextTCPReadStreamSocketNum();
    unsigned char tcpReadStreamChannelId = fRTCPInterface.nextTCPReadStreamChannelId();
    unsigned packetSize = 0;
    unsigned numBytesRead;
    struct sockaddr_in fromAddress;
    Boolean packetReadWasIncomplete;
    Boolean readResult
      = fRTCPInterface.handleRead(&fInBuf[fNumBytesAlreadyRead], maxPacketSize - fNumBytesAlreadyRead,
				  numBytesRead, fromAddress, packetReadWasIncomplete);
    if (packetReadWasIncomplete) {
      fNumBytesAlreadyRead += numBytesRead;
      return; // more reads are needed to get the entire packet
    } else { // normal case: We've read the entire packet 
      packetSize = fNumBytesAlreadyRead + numBytesRead;