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

#include <assert.h>

#include "PlayoutBuffer.h"
#include "AudioVideoPlaybackController.h"
#include "MediaStreamProducerImpl.h"

namespace videonext { namespace media {

//#define DEBUG_PLAYBACK

// Jitter buffer will be cleared for live streams if it's more than maxLatency value
const ACE_Time_Value maxLatency(2, 0);

PlayoutList::PlayoutList(AdaptiveUsageEnvironment &env)
    : fEnv(env)
{
}

/*virtual*/ PlayoutList::~PlayoutList()
{
}

void PlayoutList::insert(const CPMediaFrame &frame)
{
    ACE_Guard<ACE_Recursive_Thread_Mutex> g(lock_);

    fFrameList[frame->mediaType].push_back(frame);
}

ACE_Time_Value PlayoutList::length(MediaFrame::MEDIA_TYPE media_type) const
{
    ACE_Guard<ACE_Recursive_Thread_Mutex> g((ACE_Recursive_Thread_Mutex&)lock_);

    if (fFrameList[media_type].empty())
        return ACE_Time_Value(0,0);

    ACE_Time_Value vLength(0,0);

    if (fFrameList[media_type].size() > 1)
    {
        vLength = ACE_Time_Value((*fFrameList[media_type].rbegin())->tv) - ACE_Time_Value((*fFrameList[media_type].begin())->tv);
    }

    return vLength;
}

size_t PlayoutList::size(MediaFrame::MEDIA_TYPE media_type) const
{
    ACE_Guard<ACE_Recursive_Thread_Mutex> g((ACE_Recursive_Thread_Mutex&)lock_);

    return fFrameList[media_type].size();
}

void PlayoutList::clear()
{
    ACE_Guard<ACE_Recursive_Thread_Mutex> g((ACE_Recursive_Thread_Mutex&)lock_);

    fFrameList[MediaFrame::AUDIO].clear();
    fFrameList[MediaFrame::VIDEO].clear();
}

CPMediaFrame PlayoutList::getNextFrame(int type, ACE_Time_Value *delayToNextFrame)
{
    ACE_Guard<ACE_Recursive_Thread_Mutex> g(lock_);

    int winner = type;
    if (winner == -1)
    {
        // If type of frame has not defined, try to start from audio.
        if (fFrameList[MediaFrame::AUDIO].size() && fFrameList[MediaFrame::VIDEO].size())
        {
            CPMediaFrame nextVFrame = fFrameList[MediaFrame::VIDEO].front();
            for (std::list<CPMediaFrame>::iterator it = fFrameList[MediaFrame::AUDIO].begin(); it != fFrameList[MediaFrame::AUDIO].end();)
            {
                if (ACE_Time_Value((*it)->tv) < ACE_Time_Value(nextVFrame->tv))
                {
#ifdef DEBUG_PLAYBACK
                    fprintf(stderr, "Removed outdated audio frame (%ld.%06d)\n", (*it)->tv.tv_sec, (int)(*it)->tv.tv_usec);
#endif
                    fFrameList[MediaFrame::AUDIO].erase(it++);
                }
                else
                {
                    ++it;
                }
            }
        }

        // Select the winner
        if (!(fFrameList[MediaFrame::AUDIO].size() + fFrameList[MediaFrame::VIDEO].size()))
            return CPMediaFrame(0);

        // by default
        winner = fFrameList[MediaFrame::VIDEO].size() ? MediaFrame::VIDEO : MediaFrame::AUDIO;
        struct timeval winner_pts = fFrameList[winner].front()->tv;

        if (fFrameList[MediaFrame::AUDIO].size() && fFrameList[MediaFrame::VIDEO].size())
        {
            CPMediaFrame nextAFrame = fFrameList[MediaFrame::AUDIO].front();
            CPMediaFrame nextVFrame = fFrameList[MediaFrame::VIDEO].front();

            if (ACE_Time_Value(nextAFrame->tv) > ACE_Time_Value(nextVFrame->tv))
            {
                winner = MediaFrame::AUDIO;
                winner_pts = nextAFrame->tv;
            }
            else
            {
                winner = MediaFrame::VIDEO;
                winner_pts = nextVFrame->tv;
            }
        }
        fEnv.log("PlayoutBuffer: Winner frame is %c (%ld.%06d)\n", "AV"[winner], winner_pts.tv_sec, (int)winner_pts.tv_usec);
#ifdef DEBUG_PLAYBACK
        fEnv.log("PlayoutBuffer: BEGIN\n");
        this->print();
        fEnv.log("PlayoutBuffer: END\n");
#endif
    }

    if (fFrameList[winner].empty())
    {
        return CPMediaFrame(0);
    }

    CPMediaFrame frame = fFrameList[winner].front();
    fFrameList[winner].pop_front();

    if (delayToNextFrame)
    {
        ACE_Time_Value nextFrameTime;
        if (fFrameList[winner].empty())
        {
            *delayToNextFrame = ACE_Time_Value::zero;
        }
        else
        {
            CPMediaFrame nextFrame = fFrameList[winner].front();
            *delayToNextFrame = ACE_Time_Value(nextFrame->tv) - ACE_Time_Value(frame->tv);

            if (delayToNextFrame->sec() < 0 || delayToNextFrame->sec() > 2)
            {
                fprintf(stderr, "OUPS. Something wrong in the frame ordering\n");
                *delayToNextFrame = ACE_Time_Value(0,0);
            }
        }
    }

    return frame;
}

void PlayoutList::print()
{
    ACE_Guard<ACE_Recursive_Thread_Mutex> g(lock_);

    std::list<CPMediaFrame>::iterator aIter = fFrameList[MediaFrame::AUDIO].begin();
    std::list<CPMediaFrame>::iterator vIter = fFrameList[MediaFrame::VIDEO].begin();

    for (; aIter != fFrameList[MediaFrame::AUDIO].end();
         ++aIter)
    {
        fprintf(stderr, "[A] %ld.%06d\n", (*aIter)->tv.tv_sec, (int)(*aIter)->tv.tv_usec);
    }

    for (; vIter != fFrameList[MediaFrame::VIDEO].end();
         ++vIter)
    {
        fprintf(stderr, "[V] %ld.%06d\n", (*vIter)->tv.tv_sec, (int)(*vIter)->tv.tv_usec);
    }

    return;


  for (; !(aIter == fFrameList[MediaFrame::AUDIO].end() && vIter == fFrameList[MediaFrame::VIDEO].end());
         ++aIter, ++vIter)
    {

        if (aIter != fFrameList[MediaFrame::AUDIO].end())
        {
            fprintf(stderr, "[A] %ld.%d\n", (*aIter)->tv.tv_sec, (int)(*aIter)->tv.tv_usec);
        }
        else if (vIter != fFrameList[MediaFrame::AUDIO].end())
        {
            fprintf(stderr, "[V] %ld.%d\n", (*vIter)->tv.tv_sec, (int)(*vIter)->tv.tv_usec);
        }
    }
}


////////////////////////////////////////////////
/// PlayoutBuffer impl
////////////////////////////////////////////////
static ACE_THR_FUNC_RETURN event_loop(void *arg)
{
    ACE_Reactor *reactor = static_cast<ACE_Reactor *>(arg);
    reactor->owner(ACE_OS::thr_self());
    reactor->run_reactor_event_loop();
    return 0;
}

PlayoutBuffer::PlayoutBuffer(AdaptiveUsageEnvironment& env,
                             AudioVideoPlaybackController *playbackController,
                             unsigned bufferLen)
    : fEnv(env), fPlaybackController(playbackController), fBufferLen(0,0), fPlayoutList(env)
    , fPaused(false), fBuffering(true)
{
    fNextPlayTime[0] = fNextPlayTime[1] = ACE_Time_Value(0);

    if (bufferLen > 1000)
    {
        fBufferLen = ACE_Time_Value(bufferLen / 1000, (bufferLen % 1000) * 1000);
    }
    else
    {
        fBufferLen = ACE_Time_Value(0, bufferLen * 1000);
    }

    fprintf(stderr, "%p Created jitter buffer %ld.%06ds\n", this, (long)fBufferLen.sec(), (int)fBufferLen.usec());
    fReactor = new ACE_Reactor(new ACE_TP_Reactor, true);

    fReactorThreadId = ACE_Thread_Manager::instance()->spawn_n
        (1, event_loop, fReactor/*, THR_NEW_LWP|THR_DETACHED|THR_INHERIT_SCHED*/);
}

/*virtual*/ PlayoutBuffer::~PlayoutBuffer()
{
    fprintf(stderr, "%p Destroyed jitter buffer %ld.%06ds\n", this, (long)fBufferLen.sec(), (int)fBufferLen.usec());

    delete fReactor;
}

void PlayoutBuffer::close()
{
    fReactor->cancel_timer(this);

    fReactor->end_reactor_event_loop();

    ACE_Thread_Manager::instance()->wait_grp(fReactorThreadId);

    delete this;
}

void PlayoutBuffer::scheduleFrame(CPMediaFrame &cpMediaFrame)
{
    ACE_Guard<ACE_Recursive_Thread_Mutex> g(fTimerLock);

    fPlayoutList.insert(cpMediaFrame);

    if (fBuffering)
    {
        fReactor->cancel_timer(this);

        if (fPlaybackController->getState() != BUFFERING)
        {
            fEnv.log("PlayoutBuffer: Start buffering %ld.%ds\n", (long)fBufferLen.sec(), (int)fBufferLen.usec());
            if (fBufferLen.sec() >= 3)
            {
                fBufferLen = ACE_Time_Value(3,0);
            }
            fPlaybackController->setState(BUFFERING);
        }

        if (fPlayoutList.length() >= fBufferLen)
        {
            // Buffer is filled, start playback
            fPlaybackController->setState(PLAY_FROM_SERVER);
            fBuffering = false;
            fEnv.log("PlayoutBuffer: Buffer filled: %ld.%06ds, required %ld.%06ds. Audio %zu, Video %zu frames\n", (long)fPlayoutList.length().sec(), (int)fPlayoutList.length().usec(), (long)fBufferLen.sec(), (int)fBufferLen.usec(), fPlayoutList.size(MediaFrame::AUDIO), fPlayoutList.size(MediaFrame::VIDEO));
            fReactor->schedule_timer(this, 0, ACE_Time_Value(0, 0));

#ifdef DEBUG_PLAYBACK
            fSumDuration[0] = fSumDuration[1] = fSumRealDuration[0] = fSumRealDuration[1] =  fPrevPlayTime[0] =  fPrevPlayTime[1] = ACE_Time_Value(0);
#endif
        }
    }
}

void PlayoutBuffer::setPaused(bool value)
{
    ACE_Guard<ACE_Recursive_Thread_Mutex> g(fTimerLock);

    if (fPaused == value)
        return;

    fPaused = value;

    if (fPaused)
    {
        fReactor->cancel_timer(this);
    }
    else
    {
        fReactor->schedule_timer(this, 0, ACE_Time_Value(0, 0));
    }
}

void PlayoutBuffer::flush()
{
    ACE_Guard<ACE_Recursive_Thread_Mutex> g(fTimerLock);

    fPlayoutList.clear();
}

void PlayoutBuffer::soft_flush()
{
    ACE_Guard<ACE_Recursive_Thread_Mutex> g(fTimerLock);

    fReactor->cancel_timer(this);

    while (1)
    {
        CPMediaFrame frame = fPlayoutList.getNextFrame(MediaFrame::VIDEO, NULL);

        if (!frame.get())
            break;

        fPlaybackController->playFrame(frame, false);
    }
    while (1)
    {
        CPMediaFrame frame = fPlayoutList.getNextFrame(MediaFrame::AUDIO, NULL);

        if (!frame.get())
            break;

        fPlaybackController->playFrame(frame, false);
    }

    fPlayoutList.clear();
}

/*virtual*/ int PlayoutBuffer::handle_timeout(const ACE_Time_Value &current_time, const void *act)
{
    ACE_Guard<ACE_Recursive_Thread_Mutex> g(fTimerLock);

    int frame_type = -1;

    if (act)
    {
        frame_type = (long)act - 1;
    }

    playFrame(frame_type);

    return 0;
}

void PlayoutBuffer::playFrame(int frame_type)
{
    ACE_Guard<ACE_Recursive_Thread_Mutex> g(fTimerLock);

    if (fPlayoutList.size() < 2)
    {
        fEnv.log("PlayoutBuffer: BUFFER NOT ENOUGH\n");

        // Actually that's bad sign. That means that jitter buffer is small and should be increased.
        fBufferLen += ACE_Time_Value(0, 256000);
        fEnv.log("PlayoutBuffer: New playout buffer len: %ld.%06d\n", (long)fBufferLen.sec(), (int)fBufferLen.usec());
        fPlaybackController->setState(BUFFERING);
        fReactor->cancel_timer(this);
        fBuffering = true;
        fNextPlayTime[0] = fNextPlayTime[1] = ACE_Time_Value(0);
        return;
    }

    ACE_Time_Value now = ACE_OS::gettimeofday();

    // Get next frame and its playback time from list
    ACE_Time_Value delayToNextFrame(0,0);
    CPMediaFrame frame = fPlayoutList.getNextFrame(frame_type, &delayToNextFrame);
    ACE_Time_Value duration = delayToNextFrame;

    if (!frame.get())
    {
        // Something went wrong
        fEnv.log("PlayoutBuffer: Could not get frame %d from list (NULL) Video: %zu, %ld.%d, Audio: %zu, %ld.%d (allowed: %ld.%d)\n", frame_type,
                 fPlayoutList.size(MediaFrame::VIDEO), (long)fPlayoutList.length(MediaFrame::VIDEO).sec(), (int)fPlayoutList.length(MediaFrame::VIDEO).usec(),
                 fPlayoutList.size(MediaFrame::AUDIO), (long)fPlayoutList.length(MediaFrame::AUDIO).sec(), (int)fPlayoutList.length(MediaFrame::AUDIO).usec(),
                 (long)fBufferLen.sec(), (int)fBufferLen.usec());
        if (frame_type == -1 || frame_type == MediaFrame::VIDEO)
        {
            this->soft_flush();
            fReactor->cancel_timer(this);
            if (fNextPlayTime[MediaFrame::AUDIO] != ACE_Time_Value::zero)
            {
                fBufferLen += ACE_Time_Value(0, 128000);
            }
            fBuffering = true;
            fNextPlayTime[0] = fNextPlayTime[1] = ACE_Time_Value(0);
        }
        else if (frame_type != -1 && fNextPlayTime[frame_type] != ACE_Time_Value::zero)
        {
            this->soft_flush();
            fReactor->cancel_timer(this);
            if (fNextPlayTime[MediaFrame::AUDIO] != ACE_Time_Value::zero)
            {
                fBufferLen += ACE_Time_Value(0, 128000);
            }
            fBuffering = true;
            fNextPlayTime[0] = fNextPlayTime[1] = ACE_Time_Value(0);
        }

        return;
    }

    if (!fPlaybackController->mediaProducer()->isPlayingArchive())
    {
        // Check for buffer overflow
        ACE_Time_Value currentBufferLength = fPlayoutList.length();
        if (currentBufferLength != ACE_Time_Value::zero && currentBufferLength > (fBufferLen + maxLatency))
        {
            fEnv.log("PlayoutBuffer: Buffer overflown! Size: %zu, Len: %ld.%06d (allowed: %ld.%06d)\n", fPlayoutList.size(), (long)fPlayoutList.length().sec(), (int)fPlayoutList.length().usec(), (long)fBufferLen.sec(), (int)fBufferLen.usec());
            this->soft_flush();
            fReactor->cancel_timer(this);
            fBuffering = true;
            fNextPlayTime[0] = fNextPlayTime[1] = ACE_Time_Value(0);
            return;
        }
    }

#ifdef DEBUG_PLAYBACK
    fprintf(stderr, "Size: %zu, Len: %ld.%d (allowed: %ld.%d)\n", fPlayoutList.size(), fPlayoutList.length().sec(), (int)fPlayoutList.length().usec(), fBufferLen.sec(), (int)fBufferLen.usec());
    fprintf(stderr, "[%c] playing %ld.%d. (duration: %ld.%d (sum:%ld.%d), realduration: %ld.%d (sum: %ld.%d)) (now: %ld.%d)\n",
            "AV"[frame->mediaType], frame->tv.tv_sec,  frame->tv.tv_usec, delayToNextFrame.sec(),  delayToNextFrame.usec(),
            fSumDuration[frame->mediaType].sec(), fSumDuration[frame->mediaType].usec(),
            (now - fPrevPlayTime[frame->mediaType]).sec(), (now - fPrevPlayTime[frame->mediaType]).usec(),
            fSumRealDuration[frame->mediaType].sec(), fSumRealDuration[frame->mediaType].usec(),
            now.sec(), now.usec());
    fSumDuration[frame->mediaType] += delayToNextFrame;
    if (fPrevPlayTime[frame->mediaType] != ACE_Time_Value::zero)
        fSumRealDuration[frame->mediaType] += (now - fPrevPlayTime[frame->mediaType]);
    fPrevPlayTime[frame->mediaType] = now;
#endif

    // Schedule next frame for playback.
    if (!fPaused)
    {
        if (fNextPlayTime[frame->mediaType] == ACE_Time_Value::zero)
            fNextPlayTime[frame->mediaType] = now;

        // Scheduler latency is time between we scheduled playback and time when our callback is actually called.
        ACE_Time_Value schedulerLatency(0);
        if (now > fNextPlayTime[frame->mediaType])
            schedulerLatency = now - fNextPlayTime[frame->mediaType];

        fNextPlayTime[frame->mediaType] += delayToNextFrame;

        if (schedulerLatency < delayToNextFrame)
            delayToNextFrame -= schedulerLatency;
        else
            delayToNextFrame = ACE_Time_Value(0);

        if (fReactor->schedule_timer(this, (void*)((long)frame->mediaType+1), delayToNextFrame) == -1)
        {
            fprintf(stderr, "schedule_timer() failed: %s\n", ACE_OS::strerror(ACE_OS::last_error()));
        }

        if (frame_type == -1)
        {
            if (frame->mediaType == MediaFrame::AUDIO)
                fReactor->schedule_timer(this, (void*)(2), duration);
            else
                fReactor->schedule_timer(this, (void*)(1), ACE_Time_Value(0));
        }
    }

    fTimerLock.release();

    // Playing back current frame
    fPlaybackController->playFrame(frame);

    fTimerLock.acquire();

}


}}
