/*
#  $Id$
# -----------------------------------------------------------------------------
#  The part of 'VideoNEXT MediaClient SDK'
# -----------------------------------------------------------------------------
#  Author: Petrov Maxim, 03/04/2009
#  Edited by:
#  QA by:
#  Copyright: videoNEXT LLC
# -----------------------------------------------------------------------------
*/

#include <stdio.h>
#include <stdlib.h>
#include <string.h> // memset
#include <signal.h>

#ifndef __MINGW32CE__
#include <locale.h>
#endif

#include "AudioVideoPlaybackController.h"
#include "BasicUsageEnvironment.hh"
#include "RTSPMediaStreamProducerImpl.h" 


#if defined(__WIN32__) || defined(_WIN32)
extern "C" int initializeWinsockIfNecessary();
#endif

namespace videonext { namespace media {

static Mutex envMutex;

const char *applicationName    = "VN_MediaPlayer";
const char *clientProtocolName = "RTSP";
const char *progName           = "VN_MediaPlayer";
    

class MyAudioVideoPlaybackController : public AudioVideoPlaybackController
{
public:
   MyAudioVideoPlaybackController(AdaptiveUsageEnvironment& env,
                                  MediaStreamProducerImpl *p,
                                  MediaStreamHandler *mediaStreamHandler,
                                  unsigned cacheSize,
                                  unsigned bufferLen)
       : AudioVideoPlaybackController(env, p, mediaStreamHandler,
                                      cacheSize, bufferLen)
   {}

   virtual ~MyAudioVideoPlaybackController() {}
   /*virtual*/ void setState(STREAM_STATE val, int errorCode = 0) {
       if(val == PLAY_FROM_SERVER)
           fMediaProducer->noteLiveness();
       AudioVideoPlaybackController::setState(val, errorCode);
   }
};

RTSPMediaStreamProducerImpl::RTSPMediaStreamProducerImpl(MediaStreamHandler *mediaStreamHandler, const std::string &url, 
                                                         unsigned cacheSize, unsigned streamId, bool streamOverTCP, 
                                                         unsigned bufferLen, float speed)
                                     
    : MediaStreamProducerImpl(mediaStreamHandler, url, cacheSize, streamId, streamOverTCP, bufferLen, speed),
      fEnv(0), fScheduler(0),
      fStopFlag(0), fDidShutdown(false), fIsSessionPaused(false),
      fFrameSinkBufferSize(1000000),
      fRTSPClient(0), fSession(0),
      fStepMode(false), fPlayingArchive(false), fVideoTranscoded(false), fPlayDirection(1), fLivenessCheckTask(0),
      fPlayTriggerId(0), fStepModeTriggerId(0), fPauseTriggerId(0)
{
    MutexGuard g(envMutex);
    
    if (url_.find("start") != std::string::npos) fPlayingArchive = true;               
    if (url_.find("transcode") != std::string::npos) fVideoTranscoded = true;
    
    // Begin by setting up our usage environment:
    fScheduler = BasicTaskScheduler::createNew();  
    fEnv = AdaptiveUsageEnvironment::createNew(*fScheduler, mediaStreamHandler_);
  
#if defined(__WIN32__) || defined(_WIN32)
    if (!initializeWinsockIfNecessary()) {
       fEnv->setResultErrMsg("Failed to initialize 'winsock': ");
       fEnv->reportBackgroundError();

       changeState(STOPPED);

       return;
  }
#else
    signal(SIGPIPE, SIG_IGN); 
#endif

    fPlaybackController = new MyAudioVideoPlaybackController(*fEnv, this, mediaStreamHandler, cacheSize, bufferLen);
}

/*virtual*/RTSPMediaStreamProducerImpl::~RTSPMediaStreamProducerImpl()
{
    fprintf(stderr, "RTSPMediaStreamProducerImpl::~RTSPMediaStreamProducerImpl: %p\n", this);

    if (fSession)
    {
        MediaSubsessionIterator iter(*fSession);
        MediaSubsession* subsession;
        while ((subsession = iter.next()) != NULL) {
            fPlaybackController->removeSink((FrameConsumer*)subsession->sink);
        }
    }
    if (fPlaybackController)
    {
        Medium::close(fPlaybackController);
        fPlaybackController = NULL;
    }
    
    shutdown();      
        
    std::list<ClientData*>::iterator pos;
    for (pos = fClientDataList.begin(); pos != fClientDataList.end(); ++pos)
    {
        delete *pos;
    }

    delete fScheduler;
    fEnv->reclaim2();
}

/*virtual*/ void RTSPMediaStreamProducerImpl::setPlayDirection(int value)
{
   fPlayDirection = value;
}

/*virtual*/void RTSPMediaStreamProducerImpl::changeState(STREAM_STATE state, int errorCode)
{
   fPlaybackController->setState(state, errorCode);
}

/*virtual*/void RTSPMediaStreamProducerImpl::open()
{
    gettimeofday(&fStartTime, NULL);
          
    // Create our client object:
    fRTSPClient = RTSPClient::createNew(*fEnv, 0 /*verbosity level*/, applicationName, 0);
    
    if (fRTSPClient == NULL) 
    {
        *fEnv << "Failed to create " << clientProtocolName << " client: " << fEnv->getResultMsg() << "\n";

        changeState(STOPPED);

        return;
    }
    
    changeState(OPENING);
    
    // Open the URL, to get a SDP description:
    char* sdpDescription
       = getSDPDescriptionFromURL(url_.c_str(), 0 /*username*/, 0 /*password*/);
    if (sdpDescription == NULL) 
    {
        fEnv->setResultErr(EINTERNALERROR);
        changeState(STOPPED, EINTERNALERROR);

        *fEnv << "Failed to get a SDP description from URL " << "\"" << url_.c_str() << "\": " << fEnv->getResultMsg() << "\n";

        return;
    }


    // Get server version
    char *tool = strstr(sdpDescription, "a=tool:VideoNEXT streamer");
    if (tool)
    {
       int number[3];
       if (sscanf(tool, "a=tool:VideoNEXT streamer v%d.%d.%d", &number[0], &number[1], &number[2]) == 3)
       {
          serverVersion_ = number[0]*100 + number[1]*10 + number[2];
       }
    }


    *fEnv << "Got SDP description: " << sdpDescription << "\n";
    
    // Create a media session object from this SDP description:
    fSession = MediaSession::createNew(*fEnv, sdpDescription);
    delete[] sdpDescription;
    if (fSession == NULL) 
    {
	fEnv->setResultErr(EINTERNALERROR);
        *fEnv << "Failed to create a MediaSession object from the SDP description: "
            << fEnv->getResultMsg() << "\n";


        changeState(STOPPED, EINTERNALERROR);

        return;
    } 
    else if (!fSession->hasSubsessions()) 
    {
        *fEnv << "This session has no media subsessions (i.e., \"m=\" lines)\n";

        changeState(STOPPED, EINTERNALERROR);
        
        return;
    }
    
    // Then, setup the "RTPSource"s for the session:
    MediaSubsessionIterator iter(*fSession);
    int socketInputBufferSize = 0;
    Boolean madeProgress = False;
    MediaSubsession *subsession;
    while ((subsession = iter.next()) != NULL) 
    {
         /* Value taken from mplayer */
        if( !strcmp( subsession->mediumName(), "audio" ) )
            socketInputBufferSize = 100000;
        else if( !strcmp( subsession->mediumName(), "video" ) )
            socketInputBufferSize = 2000000;
        else
            continue;

        // Creating receivers    
        if (!subsession->initiate(-1/*simpleRTPoffsetArg*/)) 
        {
            *fEnv << "Unable to create receiver for \"" << subsession->mediumName()
                << "/" << subsession->codecName() << "\" subsession: " << fEnv->getResultMsg() << "\n";
        } 
        else 
        {
            *fEnv << "Created receiver for \"" << subsession->mediumName()
                << "/" << subsession->codecName() << "\" subsession: (client ports "
                << subsession->clientPortNum() << "-" << subsession->clientPortNum()+1 << ")\n";
              
            madeProgress = True;

            if (subsession->rtpSource() != NULL) 
            {
                subsession->rtpSource()->setPacketReorderingThresholdTime(30000);

                if (socketInputBufferSize > 0) 
                {
                    // Set the RTP source's input buffer size as specified:
                    int socketNum
                        = subsession->rtpSource()->RTPgs()->socketNum();
                    unsigned curBufferSize
                        = getReceiveBufferSize(*fEnv, socketNum);
                    /*
                    unsigned newBufferSize
                        = setReceiveBufferTo(*env, socketNum, socketInputBufferSize);
                    */
                    
                    /* Increase the buffer size */
                    unsigned newBufferSize = increaseReceiveBufferTo( *fEnv, socketNum, socketInputBufferSize );
                    
                    *fEnv << "Changed socket receive buffer size for the \"" << subsession->mediumName()
                        << "/" << subsession->codecName() << "\" subsession from "
                        << curBufferSize << " to " << newBufferSize << " bytes\n";
                }
            }
        }
    } 

  
    if (!madeProgress) 
    {
       fEnv->setResultErrMsg("Could not initialize session. No sink created");
       fEnv->setResultErr(EINTERNALERROR);

       changeState(STOPPED);
       
       return;
    }

    // Perform additional 'setup' on each subsession, before playing them:
    if (!setupStreams())
    {
       char msg[512];
       snprintf(msg, sizeof(msg), "Setup Streams Failed: %s.",  fEnv->getResultMsg());
       fEnv->setResultErrMsg(msg);
       fEnv->setResultErr(EINTERNALERROR);
       
       changeState(STOPPED, EINTERNALERROR);

       return;
    }
    
    // Create and start Frame Consumers for each subsession:
    madeProgress = False;
    iter.reset();
    while ((subsession = iter.next()) != NULL) 
    {
        if (subsession->readSource() == NULL) continue; // was not initiated

        FrameConsumer *sink = NULL;

        StreamInfo *streamInfo = new StreamInfo;
        streamInfo->streamId = streamId_;
        streamInfo->RTPCodecStr = subsession->codecName();
        streamInfo->configData = 0;
        streamInfo->configDataSize = 0;
        streamInfo->duration = 0;

        if (subsession->fmtp_config() != 0)
           streamInfo->configData = parseGeneralConfigStr(subsession->fmtp_config(), streamInfo->configDataSize);

        // Creating consumer
        if( !strcmp( subsession->mediumName(), "audio" ) )
        {
           streamInfo->type = AUDIO;

           // Getting audio rate, channels, bits...
           streamInfo->numChannels   = subsession->numChannels();
           streamInfo->samplingFreq  = subsession->rtpTimestampFrequency();
           streamInfo->bitsPerSample = (streamInfo->samplingFreq == 8000 && streamInfo->numChannels == 1) ? 8 : 16;            

           sink  = AudioFrameConsumer::createNew(*fEnv, fPlaybackController,
                                                 fFrameSinkBufferSize,
                                                 decoderType_,
                                                 streamInfo);
           fPlaybackController->addSink(sink, AudioVideoPlaybackController::AUDIO);
           
        }
        else if( !strcmp( subsession->mediumName(), "video" ) )
        {
           streamInfo->type = VIDEO;

           sink  = VideoFrameConsumer::createNew(*fEnv, fPlaybackController, 
                                                 fFrameSinkBufferSize,
                                                 decoderType_, preferedPixelFormat_,
                                                 streamInfo, fVideoTranscoded);

           fPlaybackController->addSink(sink, AudioVideoPlaybackController::VIDEO);

           if (strcmp(subsession->codecName(), "H264") == 0 
               && subsession->fmtp_spropparametersets() != NULL
               && !fVideoTranscoded)
           {
               // For H264 video RTP streams, the 'sprop-parameter-sets' information
               // from the SDP description used to convey any sequence and picture
               // parameter set NAL units that MUST precede any other NAL units in
               // decoding order.  The value of the parameter is base64 representation
               // of the initial parameter set NAL units.

               unsigned arrayLen;
               SPropRecord* sprop_array = parseSPropParameterSets(subsession->fmtp_spropparametersets(), arrayLen);
               unsigned initialNALULen = 0;

               for(size_t i=0; i<arrayLen; i++) {
                   initialNALULen += sprop_array[i].sPropLength;
               }
		    
               unsigned char* initialNALU = new unsigned char[initialNALULen + 4*2/*startcodes length just for case*/];
               unsigned char* pos = initialNALU;
               for(size_t i=0; i<arrayLen; i++) {
                   if (!(sprop_array[i].sPropBytes[0] == 0 && sprop_array[i].sPropBytes[1] == 0 
                         && sprop_array[i].sPropBytes[2] == 0 && sprop_array[i].sPropBytes[3] == 1))
                   {
                       *pos++ = 0x00; *pos++ = 0x00; *pos++ = 0x00; *pos++ = 0x01; // start code
                       initialNALULen+=4; // start code length
                   }
                   memcpy(pos, sprop_array[i].sPropBytes, sprop_array[i].sPropLength);
                   pos += sprop_array[i].sPropLength;
               }

               // Insert decoded data at the front of the output file:
//               struct timeval timeNow;
//             gettimeofday(&timeNow, NULL);
//               sink->addData(initialNALU+4, initialNALULen-4, timeNow);
               sink->addExtraData(initialNALU+4, initialNALULen-4);
               delete[] sprop_array;
               delete[] initialNALU;
           }

           if (!sink->init())
           {
               delete sink;
               sink = NULL;
           }
        }
        else
        {
           fEnv->setResultErr(EUNKNOWNSTREAMTYPE);
           changeState(STOPPED);

           if (streamInfo->configData) delete[] streamInfo->configData;
           delete streamInfo;

           return;
        }

        subsession->sink = sink;
        
        if (subsession->sink == NULL) 
        {             
           break;
        } 

        ClientData *pClientData = new ClientData(this, subsession);
        
        subsession->sink->startPlaying(*(subsession->readSource()),
                    RTSPMediaStreamProducerImpl::subsessionAfterPlaying,
                    pClientData);

        // Also set a handler to be called if a RTCP "BYE" arrives
        // for this subsession:
        if (subsession->rtcpInstance() != NULL) 
        {
           subsession->rtcpInstance()->setByeHandler(subsessionByeHandler, pClientData);
            
           subsession->rtcpInstance()->setSRHandler(noteServerLiveness, this);
           subsession->rtcpInstance()->setAPPHandler(rtcpAPPHandler, this);

        }
        
        fClientDataList.push_back(pClientData);
        
        madeProgress = True;
    }
       
    if (!madeProgress) 
    {
       fEnv->setResultErrMsg("Could not initialize session. No sink created");
       fEnv->setResultErr(EINTERNALERROR);

       changeState(STOPPED, EINTERNALERROR);

       return;
    }
    
    // Create trigger for future play request
    fStepModeTriggerId = fEnv->taskScheduler().createEventTrigger(stepModeHandlerTriggerProc);
    fPlayTriggerId     = fEnv->taskScheduler().createEventTrigger(playHandlerTriggerProc);
    fPauseTriggerId    = fEnv->taskScheduler().createEventTrigger(pauseHandlerTriggerProc);
    
    changeState(OPENED);

    fEnv->taskScheduler().doEventLoop(&fStopFlag); // does not return
}

/*static*/ void RTSPMediaStreamProducerImpl::playHandlerTriggerProc(void *data)
{
    RTSPMediaStreamProducerImpl *impl = static_cast<RTSPMediaStreamProducerImpl *>(data);

    impl->noteLiveness();

    if (!impl->startPlayingStreams())
    {
       *(impl->fEnv) << "Failed to start playing session: " << impl->fEnv->getResultMsg() << "\n";

       char msg[512];
       snprintf(msg, sizeof(msg), "Failed to start playing session: %s.",  impl->fEnv->getResultMsg());
       impl->fEnv->setResultErrMsg(msg);
       impl->fEnv->setResultErr(EINTERNALERROR);

       impl->changeState(STOPPED, EINTERNALERROR);
    }
}

/*static*/ void RTSPMediaStreamProducerImpl::pauseHandlerTriggerProc(void *data)
{
    RTSPMediaStreamProducerImpl *impl = static_cast<RTSPMediaStreamProducerImpl *>(data);

    impl->clientPauseSession(impl->fSession);
}

/*static*/ void RTSPMediaStreamProducerImpl::stepModeHandlerTriggerProc(void *data)
{
    RTSPMediaStreamProducerImpl *impl = static_cast<RTSPMediaStreamProducerImpl *>(data);
    MutexGuard mg(impl->fMutex);

    
    char str[20] = {};
    snprintf(str, sizeof str, "%d", impl->fStepMode);
    impl->fRTSPClient->setMediaSessionParameter(*impl->fSession, "stepMode", str);
}


// called from another thread!
/*virtual*/void RTSPMediaStreamProducerImpl::play()
{
    if (fStopFlag == 1) 
       return;

    if (fPlaybackController->hasSinks() && fPlayTriggerId == 0)
    {
        resume();
        return;
    }

    if (fPlaybackController->getState() != OPENED)
        return;

    // called from different thread
    int id = fPlayTriggerId;
    fPlayTriggerId = 0;
    fEnv->taskScheduler().triggerEvent(id, this);
}
        
// called from another thread!
void RTSPMediaStreamProducerImpl::stop()
{
   if (fStopFlag == 1) 
       return;
   
   if (fPlaybackController && fPlaybackController->hasSinks()) 
   {
      fPlaybackController->setPlayFramesFromServer(false);
      fPlaybackController->setPaused(true);         
   }

   printf("RTSPMediaStreamProducerImpl::stop()\n");
   fStopFlag = 1;
}

// can be called from another thread!
void RTSPMediaStreamProducerImpl::pause()
{
   if (!fPlaybackController->hasSinks()) return;
  
   if (fPlayingArchive && !fStepMode)  
   {
      fPlaybackController->setServerPaused(true);

      // called from different thread
      // fEnv->taskScheduler().triggerEvent(fPauseTriggerId, this);
      MutexGuard mg(fMutex);
      clientPauseSession(fSession);
   }

   fPlaybackController->setPlayFramesFromServer(false);
   fPlaybackController->setPaused(true);    
}

// can be called from different thread
void RTSPMediaStreamProducerImpl::setStepMode(bool value)
{
   if (!fPlaybackController->hasSinks()) return;
    
    if (fPlayingArchive) 
    {
        // if step mode was disabled, sending RTSP "PAUSE"
        if (fStepMode && value == 0) 
        {
            pause();
        }
        
        if (fStepMode != value)
        {
            fStepMode = value;
            fEnv->taskScheduler().triggerEvent(fStepModeTriggerId, this);
        }
    }
    
    fStepMode = value;  
}

bool RTSPMediaStreamProducerImpl::isStepMode()
{
   return fStepMode == 1;
}

/*virtual*/ void RTSPMediaStreamProducerImpl::setSpeed(float speed)
{
   if (!fPlaybackController->hasSinks()) return;
    
   fPlaybackController->setSpeed(speed);
    
   if (fPlaybackController->isPlayFramesFromServer() || 1 /* Change server speed always*/)
   {
        char speed_string[20];
#ifndef __MINGW32CE__
        char *curLocale = setlocale(LC_NUMERIC, "C");
#endif
        snprintf(speed_string, sizeof speed_string, "%f", speed);       
        MutexGuard mg(fMutex);
        fRTSPClient->setMediaSessionParameter(*fSession, "videoSpeed", speed_string);
#ifndef __MINGW32CE__
        setlocale(LC_NUMERIC, curLocale);
#endif 
    } 
    
}

/*virtual*/ void RTSPMediaStreamProducerImpl::setJitterBufferLen(unsigned value)
{
   fPlaybackController->setJitterBufferLen(value);
}
        
void RTSPMediaStreamProducerImpl::jumpToTimestamp(int timestamp, bool mark)
{
   if (!fPlaybackController->hasSinks()) return;
    
   if (fPlayingArchive) 
   {
      if (mark && fPlaybackController->getDirection() > 0) fPlaybackController->markJump();
   }    
   // looking in the cache first
    if (mark && fPlaybackController->jumpToTimestamp(timestamp, fPlayingArchive))
    {
        if (fPlaybackController->isPlayFramesFromServer() && !fPlaybackController->isPaused())
        {
            pause();
           
            changeState(IDLE);
        }            
    }
    else if (fPlaybackController->getDirection() > 0 && fPlayingArchive)
    {        
        char str[20];
        snprintf(str, sizeof str, "%d", timestamp);
        MutexGuard mg(fMutex);
        fRTSPClient->setMediaSessionParameter(*fSession, "jump2timestamp", str);
    }

    
}

void RTSPMediaStreamProducerImpl::setBandwidth(int value)
{
    char str[20];
    if (value == 0)
    {
        strcpy(str, "auto");
    }
    else if (value == -1)
    {
        strcpy(str, "full");
    }
    else
    {
        snprintf(str, sizeof str, "%d", value);
    }
    MutexGuard mg(fMutex);
    fRTSPClient->setMediaSessionParameter(*fSession, "bandwidth", str);

}

void RTSPMediaStreamProducerImpl::resume()
{
   MutexGuard g(fCacheMutex);

   if (!fPlaybackController->hasSinks()) return;

    fPlaybackController->setPaused(false);
    if (fPlayDirection < 0)
    {
       if (fPlaybackController->playFromCache(-1))
       {
          changeState(IDLE);
       }
    }
    else
    {
        if (fPlaybackController->playFromCache(1))
        {
//            playbackController->clear();
            fPlaybackController->setPlayFramesFromServer(true);
            if (fPlayingArchive)
            {
               fPlaybackController->setServerPaused(false);
               MutexGuard mg(fMutex);
               clientStartPlayingSession(fSession);
            }
            else
            {
               changeState(PLAY_FROM_SERVER);
            }           
        }
    }
    
//    fPlaybackController->clear();

}


// from openRTSP:

char* RTSPMediaStreamProducerImpl::getOptionsResponse(char const* url) {
  
   return fRTSPClient->sendOptionsCmd(url);
  
}

char* RTSPMediaStreamProducerImpl::getSDPDescriptionFromURL(char const* url,
                   char const* username, char const* password) {
  char* result;
  if (username != NULL && password != NULL) {  
     result = fRTSPClient->describeWithPassword(url, username, password);
  } else {
     result = fRTSPClient->describeURL(url);
  }
  
  /*int statusCode = */fRTSPClient->describeStatus();
    
  return result;
}

Boolean RTSPMediaStreamProducerImpl::clientSetupSubsession(MediaSubsession* subsession,
                                                           Boolean streamUsingTCP) {
  if (fRTSPClient == NULL || subsession == NULL) return False;

  return fRTSPClient->setupMediaSubsession(*subsession, False, streamUsingTCP);
}

Boolean RTSPMediaStreamProducerImpl::clientStartPlayingSession(MediaSession* session) {
    
  if (fRTSPClient == NULL || session == NULL) return False;
  fIsSessionPaused = false;
  
  return fRTSPClient->playMediaSession(*session);
}

Boolean RTSPMediaStreamProducerImpl::clientTearDownSession(MediaSession* session) {
   if (fRTSPClient == NULL || session == NULL) return False;
  
   return fRTSPClient->teardownMediaSession(*session);
}

Boolean RTSPMediaStreamProducerImpl::clientPauseSession(MediaSession* session) {
  
   if (fIsSessionPaused) return True;
  
  if (fRTSPClient == NULL || fSession == NULL) return False;
  
  Boolean result;
  
  fIsSessionPaused = true;
  
  result = fRTSPClient->pauseMediaSession(*session);
    
  /* if we stream over TCP we do not receive response from server.
   * Therefore do little delay to prevent crossing next RTSP command
   */
   if (streamOverTCP_)
   {
#ifdef WIN32
        Sleep(100);
#else
        usleep(100000);
#endif   
   }  
    
    return result;
}


// from live
bool RTSPMediaStreamProducerImpl::setupStreams() {
  MediaSubsessionIterator iter(*fSession);
  MediaSubsession *subsession;
  Boolean madeProgress = False;

  while ((subsession = iter.next()) != NULL) {
    if (subsession->clientPortNum() == 0) 
    {
       fprintf(stderr, "!!! Port number was not set\n");
       continue; // port # was not set
    }

    if (!clientSetupSubsession(subsession, streamOverTCP_)) {
        *fEnv << "Failed to setup \"" <<  subsession->mediumName()
            << "/" << subsession->codecName() << "\" subsession: " << fEnv->getResultMsg() << "\n";
    } else {
        *fEnv << "Setup \"" <<  subsession->mediumName()
            << "/" << subsession->codecName() << "\"subsession (client ports "
            << subsession->clientPortNum() << "-" << subsession->clientPortNum()+1
            << ")\n";   
      madeProgress = True;
    }
  }

  return madeProgress;
}

bool RTSPMediaStreamProducerImpl::startPlayingStreams() {
  if (!clientStartPlayingSession(fSession)) {
     return false;
  } else {
     *fEnv << "Entering into loop to receive video stream\n";
  }

//  *fEnv << "Receiving streamed data\n";

  return true;
}

void RTSPMediaStreamProducerImpl::tearDownStreams() {
  if (fSession == NULL) return;

  clientTearDownSession(fSession);
}

void RTSPMediaStreamProducerImpl::closeMediaSinks() {
  if (fSession == NULL) return;
  MediaSubsessionIterator iter(*fSession);
  MediaSubsession* subsession;
  while ((subsession = iter.next()) != NULL) {
    Medium::close(subsession->sink);
    subsession->sink = NULL;
  }
}

void RTSPMediaStreamProducerImpl::subsessionAfterPlaying(void* clientData) {
  // Begin by closing this media subsession:
 
  MediaSubsession* subsession   = ((ClientData*)clientData)->mediaSubsession;   
  RTSPMediaStreamProducerImpl* mp   = ((ClientData*)clientData)->mp;
  
  mp->fPlaybackController->removeSink((FrameConsumer*)subsession->sink);
 
  Medium::close(subsession->sink);
  
  subsession->sink = NULL;

  // Next, check whether *all* subsessions have now been closed:
  MediaSession& session = subsession->parentSession();
  MediaSubsessionIterator iter(session);
  while ((subsession = iter.next()) != NULL) {
    if (subsession->sink != NULL) return; // this subsession is still active
  }

  // All subsessions have now been closed
  sessionAfterPlaying(clientData); 
    
}

void RTSPMediaStreamProducerImpl::subsessionByeHandler(void* clientData) {
  
    RTSPMediaStreamProducerImpl *mp         = ((ClientData*)clientData)->mp;
    MediaSubsession* subsession   = ((ClientData*)clientData)->mediaSubsession;  
    
    struct timeval timeNow;
    gettimeofday(&timeNow, NULL);
  
    unsigned secsDiff = timeNow.tv_sec - mp->fStartTime.tv_sec;
    
   *mp->fEnv << "Received RTCP \"BYE\" on \"" << subsession->mediumName()
    << "/" << subsession->codecName()
        << "\" subsession (after " << secsDiff
    << " seconds)\n";
    
  // Act now as if the subsession had closed:
  subsessionAfterPlaying(clientData);
}

void RTSPMediaStreamProducerImpl::sessionAfterPlaying(void* clientData) {
  
    RTSPMediaStreamProducerImpl *mp         = ((ClientData*)clientData)->mp;
    
    mp->changeState(STOPPED);

    mp->stop();
     
}

void RTSPMediaStreamProducerImpl::shutdown() {
  
   MutexGuard g(fCacheMutex);

  if (fDidShutdown) 
  {
     return;
  }
    
  fDidShutdown = true;      
  
  // Close our output files:
  closeMediaSinks();

  // Teardown, then shutdown, any outstanding RTP/RTCP subsessions
  tearDownStreams();
  
  // shutdown our session and client:  
  Medium::close(fRTSPClient);
  Medium::close(fSession);

  fEnv->taskScheduler().unscheduleDelayedTask(fLivenessCheckTask);   
}

#define LIVENESS_TIMEOUT 10 // sec
/*
unsigned RTSPMediaProducer::livenessTimeout(){
    return fLivenessTimeout;
}
*/
void RTSPMediaStreamProducerImpl::noteLiveness() {  
   fEnv->taskScheduler().unscheduleDelayedTask(fLivenessCheckTask);   
   fLivenessCheckTask = fEnv->taskScheduler()
      .scheduleDelayedTask(LIVENESS_TIMEOUT*1000000,
                 (TaskFunc*)livenessTimeoutTask, this);  
}

void RTSPMediaStreamProducerImpl::noteServerLiveness(void* clientData) {
    RTSPMediaStreamProducerImpl* mp = (RTSPMediaStreamProducerImpl*)clientData;
    
    mp->noteLiveness();    
}

void RTSPMediaStreamProducerImpl::livenessTimeoutTask(void* clientData) {
  // If this gets called, the client session is assumed to have timed out 
       
  RTSPMediaStreamProducerImpl* mp = (RTSPMediaStreamProducerImpl*)clientData;  
    
  *mp->fEnv << "RTCP Timeout after " << LIVENESS_TIMEOUT << " secs.\n";
  


  STREAM_STATE state = IDLE;
  if (mp->fPlaybackController->getState() == IDLE && mp->url_.find("mcast=1") != std::string::npos)
  {
     state = STOPPED;
  }

  mp->fEnv->setResultErr(ERTCPTIMEOUT);
  
  mp->changeState(state, ERTCPTIMEOUT);
  
}

void RTSPMediaStreamProducerImpl::rtcpAPPHandler(void *clientData, unsigned char* name, unsigned char* data, unsigned dataSize)
{
    RTSPMediaStreamProducerImpl* mp = (RTSPMediaStreamProducerImpl*)clientData;
    if (strcmp((char*)name, "EOS") == 0)
    {
        *mp->fEnv << "End of stream\n";

        mp->fEnv->setResultErr(EENDOFSTREAM);

        mp->changeState(IDLE, EENDOFSTREAM);
    }
    else if (strcmp((char*)name, "EOA") == 0)
    {
        *mp->fEnv << "End of archive\n";

        mp->fEnv->setResultErr(EENDOFARCHIVE);

        mp->changeState(IDLE, EENDOFARCHIVE);
    }
    else if (strcmp((char*)name, "MFC") == 0)
    {
        *mp->fEnv << "Media format changed\n";

        mp->fEnv->setResultErr(EMEDIAFORMATCHANGED);

        mp->changeState(IDLE, EMEDIAFORMATCHANGED);
    }
}

bool RTSPMediaStreamProducerImpl::startRecording(const char* file_name, const std::map<std::string, std::string>& metadata)
{
	return fPlaybackController->startRecording(file_name, metadata);
}

bool RTSPMediaStreamProducerImpl::endRecording(bool stop)
{
	return fPlaybackController->endRecording(stop);
}

}}
