ondemandservermediasubsession.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.
// A 'ServerMediaSubsession' object that creates new, unicast, "RTPSink"s
// on demand.
// Implementation

#include "OnDemandServerMediaSubsession.hh"
#include "RTCP.hh"
#include "BasicUDPSink.hh"
#include <GroupsockHelper.hh>

OnDemandServerMediaSubsession
::OnDemandServerMediaSubsession(UsageEnvironment& env,
				Boolean reuseFirstSource)
  : ServerMediaSubsession(env),
    fReuseFirstSource(reuseFirstSource), fLastStreamToken(NULL),
    fSDPLines(NULL) {
  fDestinationsHashTable = HashTable::create(ONE_WORD_HASH_KEYS);
  gethostname(fCNAME, sizeof fCNAME);
  fCNAME[sizeof fCNAME-1] = '\0'; // just in case
}

class Destinations {
public:
  Destinations(struct in_addr const& destAddr,
	       Port const& rtpDestPort,
	       Port const& rtcpDestPort)
    : isTCP(False), addr(destAddr), rtpPort(rtpDestPort), rtcpPort(rtcpDestPort) {
  }
  Destinations(int tcpSockNum, unsigned char rtpChanId, unsigned char rtcpChanId)
    : isTCP(True), rtpPort(0) /*dummy*/, rtcpPort(0) /*dummy*/,
      tcpSocketNum(tcpSockNum), rtpChannelId(rtpChanId), rtcpChannelId(rtcpChanId) {
  }

public:
  Boolean isTCP;
  struct in_addr addr;
  Port rtpPort;
  Port rtcpPort;
  int tcpSocketNum;
  unsigned char rtpChannelId, rtcpChannelId;
};

OnDemandServerMediaSubsession::~OnDemandServerMediaSubsession() {
  delete[] fSDPLines;

  // Clean out the destinations hash table:
  while (1) {
    Destinations* destinations
      = (Destinations*)(fDestinationsHashTable->RemoveNext());
    if (destinations == NULL) break;
    delete destinations;
  }
  delete fDestinationsHashTable;
}

char const*
OnDemandServerMediaSubsession::sdpLines() {
  if (fSDPLines == NULL) {
    // We need to construct a set of SDP lines that describe this
    // subsession (as a unicast stream).  To do so, we first create
    // dummy (unused) source and "RTPSink" objects,
    // whose parameters we use for the SDP lines: 
    unsigned estBitrate; // unused
    FramedSource* inputSource = createNewStreamSource(0, estBitrate);
    if (inputSource == NULL) return NULL; // file not found

    struct in_addr dummyAddr;
    dummyAddr.s_addr = 0;
    Groupsock dummyGroupsock(envir(), dummyAddr, 0, 0);
    unsigned char rtpPayloadType = 96 + trackNumber()-1; // if dynamic
    RTPSink* dummyRTPSink
      = createNewRTPSink(&dummyGroupsock, rtpPayloadType, inputSource);

    setSDPLinesFromRTPSink(dummyRTPSink, inputSource);
    Medium::close(dummyRTPSink);
    Medium::close(inputSource);
  }

  return fSDPLines;
}

// A class that represents the state of an ongoing stream
class StreamState {
public:
  StreamState(Port const& serverRTPPort, Port const& serverRTCPPort,
	      RTPSink* rtpSink, BasicUDPSink* udpSink,
	      unsigned totalBW, char* CNAME,
	      FramedSource* mediaSource,
	      Groupsock* rtpGS, Groupsock* rtcpGS);
  virtual ~StreamState();

  void startPlaying(Destinations* destinations);
  void pause();
  void endPlaying(Destinations* destinations);
  void reclaim();

  unsigned& referenceCount() { return fReferenceCount; }

  Port const& serverRTPPort() const { return fServerRTPPort; }
  Port const& serverRTCPPort() const { return fServerRTCPPort; }

  RTPSink const* rtpSink() const { return fRTPSink; }

  FramedSource* mediaSource() const { return fMediaSource; }

private:
  Boolean fAreCurrentlyPlaying;
  unsigned fReferenceCount;

  Port fServerRTPPort, fServerRTCPPort;

  RTPSink* fRTPSink;
  BasicUDPSink* fUDPSink;

  unsigned fTotalBW; char* fCNAME; RTCPInstance* fRTCPInstance;

  FramedSource* fMediaSource;

  Groupsock* fRTPgs; Groupsock* fRTCPgs;
};

void OnDemandServerMediaSubsession
::getStreamParameters(unsigned clientSessionId,
		      netAddressBits clientAddress,
		      Port const& clientRTPPort,
		      Port const& clientRTCPPort,
		      int tcpSocketNum,
		      unsigned char rtpChannelId,
		      unsigned char rtcpChannelId,
		      netAddressBits& destinationAddress,
		      u_int8_t& /*destinationTTL*/,
		      Boolean& isMulticast,
		      Port& serverRTPPort,
		      Port& serverRTCPPort,
		      void*& streamToken) {
  if (destinationAddress == 0) destinationAddress = clientAddress;
  struct in_addr destinationAddr; destinationAddr.s_addr = destinationAddress;
  isMulticast = False;

  if (fLastStreamToken != NULL && fReuseFirstSource) {
    // Special case: Rather than creating a new 'StreamState',
    // we reuse the one that we've already created:
    serverRTPPort = ((StreamState*)fLastStreamToken)->serverRTPPort();
    serverRTCPPort = ((StreamState*)fLastStreamToken)->serverRTCPPort();
    ++((StreamState*)fLastStreamToken)->referenceCount();
    streamToken = fLastStreamToken;
  } else {
    // Normal case: Create a new media source:
    unsigned streamBitrate;
    FramedSource* mediaSource
      = createNewStreamSource(clientSessionId, streamBitrate);

    // Create a new 'groupsock' for the RTP destination, and make sure that
    // its port number is even:
    struct in_addr dummyAddr; dummyAddr.s_addr = 0;
    Groupsock* rtpGroupsock;
    Groupsock* rtpGroupsock_old = NULL;
    portNumBits serverRTPPortNum = 0;
    while (1) {
      rtpGroupsock = new Groupsock(envir(), dummyAddr, 0, 255);
      if (!getSourcePort(envir(), rtpGroupsock->socketNum(), serverRTPPort)) break;
      serverRTPPortNum = ntohs(serverRTPPort.num());

      // If the port number's even, we're done:
      if ((serverRTPPortNum&1) == 0) break;

      // Try again (while keeping the old 'groupsock' around, so that we get
      // a different socket number next time):
      delete rtpGroupsock_old;
      rtpGroupsock_old = rtpGroupsock;
    }
    delete rtpGroupsock_old;

    // Create a sink for this stream:
    RTPSink* rtpSink;
    BasicUDPSink* udpSink;
    Groupsock* rtcpGroupsock;
    if (clientRTCPPort.num() == 0) {
      // We're streaming raw UDP (not RTP):
      rtpSink = NULL;
      udpSink = BasicUDPSink::createNew(envir(), rtpGroupsock);
      rtcpGroupsock = NULL;
    } else {
      // Normal case: We're streaming RTP (over UDP or TCP):
      unsigned char rtpPayloadType = 96 + trackNumber()-1; // if dynamic
      rtpSink = createNewRTPSink(rtpGroupsock, rtpPayloadType, mediaSource);
      udpSink = NULL;
    
      // Create a 'groupsock' for a 'RTCP instance' to be created later:
      rtcpGroupsock = new Groupsock(envir(), dummyAddr, serverRTPPortNum+1, 255);
      getSourcePort(envir(), rtcpGroupsock->socketNum(), serverRTCPPort);
    }
    
    // Turn off the destinations for each groupsock.  They'll get set later
    // (unless TCP is used instead):
    if (rtpGroupsock != NULL) rtpGroupsock->removeAllDestinations();
    if (rtcpGroupsock != NULL) rtcpGroupsock->removeAllDestinations();

    // Set up the state of the stream.  The stream will get started later:
    streamToken = fLastStreamToken
      = new StreamState(serverRTPPort, serverRTCPPort, rtpSink, udpSink,
			streamBitrate, fCNAME, mediaSource,
			rtpGroupsock, rtcpGroupsock);
  }
  
  // Record these destinations as being for this client session id:
  Destinations* destinations;
  if (tcpSocketNum < 0) { // UDP
    destinations = new Destinations(destinationAddr, clientRTPPort, clientRTCPPort);
  } else { // TCP
    destinations = new Destinations(tcpSocketNum, rtpChannelId, rtcpChannelId);
  }
  fDestinationsHashTable->Add((char const*)clientSessionId, destinations);
}

void OnDemandServerMediaSubsession::startStream(unsigned clientSessionId,
						void* streamToken,
						unsigned short& rtpSeqNum,