playcommon.cpp

Live media server used to transfer mpeg or other video/audio media

	
	*env << "kBytes_received_total\t" << curQOSRecord->kBytesTotal << "\n";
	
	*env << "measurement_sampling_interval_ms\t" << qosMeasurementIntervalMS << "\n";
	
	if (curQOSRecord->kbits_per_second_max == 0) {
	  // special case: we didn't receive any data:
	  *env <<
	    "kbits_per_second_min\tunavailable\n"
	    "kbits_per_second_ave\tunavailable\n"
	    "kbits_per_second_max\tunavailable\n";
	} else {
	  *env << "kbits_per_second_min\t" << curQOSRecord->kbits_per_second_min << "\n";
	  *env << "kbits_per_second_ave\t"
	       << (measurementTime == 0.0 ? 0.0 : 8*curQOSRecord->kBytesTotal/measurementTime) << "\n";
	  *env << "kbits_per_second_max\t" << curQOSRecord->kbits_per_second_max << "\n";
	}
	
	*env << "packet_loss_percentage_min\t" << 100*curQOSRecord->packet_loss_fraction_min << "\n";
	double packetLossFraction = numPacketsExpected == 0 ? 1.0
	  : 1.0 - numPacketsReceived/(double)numPacketsExpected;
	if (packetLossFraction < 0.0) packetLossFraction = 0.0;
	*env << "packet_loss_percentage_ave\t" << 100*packetLossFraction << "\n";
	*env << "packet_loss_percentage_max\t"
	     << (packetLossFraction == 1.0 ? 100.0 : 100*curQOSRecord->packet_loss_fraction_max) << "\n";
	
#ifdef SUPPORT_REAL_RTSP
	if (session->isRealNetworksRDT) {
	  RealRDTSource* rdt = (RealRDTSource*)src;
	  *env << "inter_packet_gap_ms_min\t" << rdt->minInterPacketGapUS()/1000.0 << "\n";
	  struct timeval totalGaps = rdt->totalInterPacketGaps();
	  double totalGapsMS = totalGaps.tv_sec*1000.0 + totalGaps.tv_usec/1000.0;
	  unsigned totNumPacketsReceived = rdt->totNumPacketsReceived();
	  *env << "inter_packet_gap_ms_ave\t"
	       << (totNumPacketsReceived == 0 ? 0.0 : totalGapsMS/totNumPacketsReceived) << "\n";
	  *env << "inter_packet_gap_ms_max\t" << rdt->maxInterPacketGapUS()/1000.0 << "\n";
	} else {
#endif
	  RTPReceptionStatsDB::Iterator statsIter(src->receptionStatsDB());
	  // Assume that there's only one SSRC source (usually the case):
	  RTPReceptionStats* stats = statsIter.next(True);
	  if (stats != NULL) {
	    *env << "inter_packet_gap_ms_min\t" << stats->minInterPacketGapUS()/1000.0 << "\n";
	    struct timeval totalGaps = stats->totalInterPacketGaps();
	    double totalGapsMS = totalGaps.tv_sec*1000.0 + totalGaps.tv_usec/1000.0;
	    unsigned totNumPacketsReceived = stats->totNumPacketsReceived();
	    *env << "inter_packet_gap_ms_ave\t"
		 << (totNumPacketsReceived == 0 ? 0.0 : totalGapsMS/totNumPacketsReceived) << "\n";
	    *env << "inter_packet_gap_ms_max\t" << stats->maxInterPacketGapUS()/1000.0 << "\n";
	  }
#ifdef SUPPORT_REAL_RTSP
	}
#endif
	
	curQOSRecord = curQOSRecord->fNext;
      }
    }
  }    

  *env << "end_QOS_statistics\n";
  delete qosRecordHead;
}

void shutdown(int exitCode) {
  if (env != NULL) {
    env->taskScheduler().unscheduleDelayedTask(sessionTimerTask);
    env->taskScheduler().unscheduleDelayedTask(arrivalCheckTimerTask);
    env->taskScheduler().unscheduleDelayedTask(interPacketGapCheckTimerTask);
    env->taskScheduler().unscheduleDelayedTask(qosMeasurementTimerTask);
  }

  if (qosMeasurementIntervalMS > 0) {
    printQOSData(exitCode);
  }

  // Close our output files:
  closeMediaSinks();

  // Teardown, then shutdown, any outstanding RTP/RTCP subsessions
  tearDownStreams();
  Medium::close(session);

  // Finally, shut down our client:
  Medium::close(ourClient);

  // Adios...
  exit(exitCode);
}

void signalHandlerShutdown(int /*sig*/) {
  *env << "Got shutdown signal\n";
  shutdown(0);
}

void checkForPacketArrival(void* /*clientData*/) {
  if (!notifyOnPacketArrival) return; // we're not checking 

  // Check each subsession, to see whether it has received data packets:
  unsigned numSubsessionsChecked = 0;
  unsigned numSubsessionsWithReceivedData = 0;
  unsigned numSubsessionsThatHaveBeenSynced = 0;

  MediaSubsessionIterator iter(*session);
  MediaSubsession* subsession;
  while ((subsession = iter.next()) != NULL) {
    RTPSource* src = subsession->rtpSource();
    if (src == NULL) continue;
    ++numSubsessionsChecked;

    if (src->receptionStatsDB().numActiveSourcesSinceLastReset() > 0) {
      // At least one data packet has arrived
      ++numSubsessionsWithReceivedData;
    }
    if (src->hasBeenSynchronizedUsingRTCP()) {
      ++numSubsessionsThatHaveBeenSynced;
    }
  }

  unsigned numSubsessionsToCheck = numSubsessionsChecked;
  // Special case for "QuickTimeFileSink"s and "AVIFileSink"s:
  // They might not use all of the input sources:
  if (qtOut != NULL) {
    numSubsessionsToCheck = qtOut->numActiveSubsessions();
  } else if (aviOut != NULL) {
    numSubsessionsToCheck = aviOut->numActiveSubsessions();
  }

  Boolean notifyTheUser;
  if (!syncStreams) {
    notifyTheUser = numSubsessionsWithReceivedData > 0; // easy case
  } else {
    notifyTheUser = numSubsessionsWithReceivedData >= numSubsessionsToCheck
      && numSubsessionsThatHaveBeenSynced == numSubsessionsChecked;
    // Note: A subsession with no active sources is considered to be synced
  }
  if (notifyTheUser) {
    struct timeval timeNow;
    gettimeofday(&timeNow, NULL);
	char timestampStr[100];
	sprintf(timestampStr, "%ld%03ld", timeNow.tv_sec, timeNow.tv_usec/1000);
    *env << (syncStreams ? "Synchronized d" : "D")
		<< "ata packets have begun arriving [" << timestampStr << "]\007\n";
    return;
  }

  // No luck, so reschedule this check again, after a delay:
  int uSecsToDelay = 100000; // 100 ms
  arrivalCheckTimerTask
    = env->taskScheduler().scheduleDelayedTask(uSecsToDelay,
			       (TaskFunc*)checkForPacketArrival, NULL);
}

void checkInterPacketGaps(void* /*clientData*/) {
  if (interPacketGapMaxTime == 0) return; // we're not checking 

  // Check each subsession, counting up how many packets have been received:
  unsigned newTotNumPacketsReceived = 0;

  MediaSubsessionIterator iter(*session);
  MediaSubsession* subsession;
  while ((subsession = iter.next()) != NULL) {
    RTPSource* src = subsession->rtpSource();
    if (src == NULL) continue;
    newTotNumPacketsReceived += src->receptionStatsDB().totNumPacketsReceived();
  }

  if (newTotNumPacketsReceived == totNumPacketsReceived) {
    // No additional packets have been received since the last time we
    // checked, so end this stream:
    *env << "Closing session, because we stopped receiving packets.\n";
    interPacketGapCheckTimerTask = NULL;
    sessionAfterPlaying();
  } else {
    totNumPacketsReceived = newTotNumPacketsReceived;
    // Check again, after the specified delay: 
    interPacketGapCheckTimerTask
      = env->taskScheduler().scheduleDelayedTask(interPacketGapMaxTime*1000000,
				 (TaskFunc*)checkInterPacketGaps, NULL);
  }
}

// WORK IN PROGRESS #####
class RTPTranslator: public FramedFilter {
public:
  static RTPTranslator* createNew(UsageEnvironment& env,
				  FramedSource* source);

private:
  RTPTranslator(UsageEnvironment& env, FramedSource* source);
  virtual ~RTPTranslator();

  static void afterGettingFrame(void* clientData,
                                unsigned numBytesRead,
				unsigned numTruncatedBytes,
                                struct timeval presentationTime,
				unsigned durationInMicroseconds);
  void afterGettingFrame1(unsigned numBytesRead,
			  unsigned numTruncatedBytes,
			  struct timeval presentationTime,
			  unsigned durationInMicroseconds);

private: // redefined virtual function:
  virtual void doGetNextFrame();

private:
  //unsigned char fBuffer[50000];//##### Later: parameterize
};

RTPTranslator* RTPTranslator::createNew(UsageEnvironment& env,
					FramedSource* source) {
  // Check whether source is a "RTPSource"??? #####
  return new RTPTranslator(env, source);
}

RTPTranslator::RTPTranslator(UsageEnvironment& env, FramedSource* source)
  : FramedFilter(env, source) {
}

RTPTranslator::~RTPTranslator() {
}

void RTPTranslator::doGetNextFrame() {
  // For now, do a direct relay #####
  fInputSource->getNextFrame(fTo, fMaxSize,
			     afterGettingFrame, this,
			     handleClosure, this);
}

void RTPTranslator::afterGettingFrame(void* clientData,
				      unsigned numBytesRead,
				      unsigned numTruncatedBytes,
				      struct timeval presentationTime,
				      unsigned durationInMicroseconds) {
  RTPTranslator* rtpTranslator = (RTPTranslator*)clientData;
  rtpTranslator->afterGettingFrame1(numBytesRead, numTruncatedBytes,
				    presentationTime, durationInMicroseconds);
}

void RTPTranslator::afterGettingFrame1(unsigned numBytesRead,
				       unsigned numTruncatedBytes,
				       struct timeval presentationTime,
				       unsigned durationInMicroseconds) {
  fFrameSize = numBytesRead;
  fPresentationTime = presentationTime;
  fNumTruncatedBytes = numTruncatedBytes;
  fDurationInMicroseconds = durationInMicroseconds;
  afterGetting(this);
}

//#####
RTSPClient* rtspClientOutgoing = NULL;
Boolean setupDestinationRTSPServer() {
  do {
    rtspClientOutgoing
      = RTSPClient::createNew(*env, verbosityLevel, progName);
    if (rtspClientOutgoing == NULL) break;

    // Construct the SDP description to announce into the RTSP server:

    // First, get our own IP address, and that of the RTSP server:
    struct in_addr ourIPAddress;
    ourIPAddress.s_addr = ourSourceAddressForMulticast(*env);
    char* ourIPAddressStr = strDup(our_inet_ntoa(ourIPAddress));

    NetAddress serverAddress;
    portNumBits serverPortNum;
    if (!RTSPClient::parseRTSPURL(*env, destRTSPURL,
				  serverAddress, serverPortNum)) break;
    struct in_addr serverIPAddress;
    serverIPAddress.s_addr = *(unsigned*)(serverAddress.data());
    char* serverIPAddressStr = strDup(our_inet_ntoa(serverIPAddress));

    char const* destSDPFmt =
      "v=0\r\n"
      "o=- %u %u IN IP4 %s\r\n"
      "s=RTSP session, relayed through \"%s\"\n"
      "i=relayed RTSP session\n"
      "t=0 0\n"
      "c=IN IP4 %s\n"
      "a=control:*\n"
      "m=audio 0 RTP/AVP %u\n"
      "a=control:trackID=0\n";
    //#####LATER: Support video as well; multiple tracks; other codecs #####
    unsigned destSDPFmtSize = strlen(destSDPFmt)
      + 20 /* max int len */ + 20 + strlen(ourIPAddressStr)
      + strlen(progName)
      + strlen(serverIPAddressStr)
      + 3 /* max char len */;
    char* destSDPDescription = new char[destSDPFmtSize];
    sprintf(destSDPDescription, destSDPFmt,
	    our_random(), our_random(), ourIPAddressStr,
	    progName,
	    serverIPAddressStr,
	    desiredAudioRTPPayloadFormat);
    Boolean announceResult;
    if (username != NULL) {
      announceResult
	= rtspClientOutgoing->announceWithPassword(destRTSPURL,
						   destSDPDescription,
						   username, password);
    } else {
      announceResult
	= rtspClientOutgoing->announceSDPDescription(destRTSPURL,
						     destSDPDescription);
    }
    delete[] serverIPAddressStr; delete[] ourIPAddressStr;
    if (!announceResult) break;
    
    // Then, create a "MediaSession" object from this SDP description:
    MediaSession* destSession
      = MediaSession::createNew(*env, destSDPDescription);
    delete[] destSDPDescription;
    if (destSession == NULL) break;

    // Initiate, setup and play "destSession".
    // ##### TEMP HACK - take advantage of the fact that we have
    // ##### a single audio session only.
    MediaSubsession* destSubsession;
    PrioritizedRTPStreamSelector* multiSource;
    int multiSourceSessionId;
    char const* mimeType
      = desiredAudioRTPPayloadFormat == 0 ? "audio/PCMU" 
      : desiredAudioRTPPayloadFormat == 3 ? "audio/GSM"
      : "audio/???"; //##### FIX
    if (!destSession->initiateByMediaType(mimeType, destSubsession,
					  multiSource,
					  multiSourceSessionId)) break;
    if (!rtspClientOutgoing->setupMediaSubsession(*destSubsession,
						  True, True)) break;
    if (!rtspClientOutgoing->playMediaSubsession(*destSubsession,
						 0.0, -1.0, 1.0,
						 True/*hackForDSS*/)) break;

    // Next, set up "RTPSink"s for the outgoing packets:
    struct in_addr destAddr; destAddr.s_addr = 0; // because we're using TCP
    Groupsock* destGS = new Groupsock(*env, destAddr, 0/*aud*/, 255);
    if (destGS == NULL) break;
    RTPSink* destRTPSink = NULL;
    if (desiredAudioRTPPayloadFormat == 0) {
      destRTPSink = SimpleRTPSink::createNew(*env, destGS, 0, 8000,
					     "audio", "PCMU");
    } else if (desiredAudioRTPPayloadFormat == 3) {
      destRTPSink = GSMAudioRTPSink::createNew(*env, destGS);
    }
    if (destRTPSink == NULL) break;

    // Tell the sink to stream using TCP:
    destRTPSink->setStreamSocket(rtspClientOutgoing->socketNum(), 0/*aud*/);
    // LATER: set up RTCPInstance also #####

    // Next, set up RTPTranslator(s) between source(s) and destination(s),
    // and start playing them.
    MediaSubsessionIterator iter(*session);
    MediaSubsession *sourceSubsession = NULL;
    while ((sourceSubsession = iter.next()) != NULL) {
      if (strcmp(sourceSubsession->mediumName(), "audio") == 0) break;
    }
    if (sourceSubsession == NULL) break;
    RTPTranslator* rtpTranslator
      = RTPTranslator::createNew(*env, sourceSubsession->readSource());
    if (rtpTranslator == NULL) break;
    destRTPSink->startPlaying(*rtpTranslator,
			      subsessionAfterPlaying, sourceSubsession);
    
    // LATER: delete media on close #####

    return True;
  } while (0);

  return False;
}