// Copyright 2019 Fuzhou Rockchip Electronics Co., Ltd. All rights reserved.
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.

#include "live555_media_input.hh"

#include <assert.h>
#include <fcntl.h>
#include <unistd.h>

#include <algorithm>

#include "buffer.h"
#include "codec.h"
#include "utils.h"

namespace easymedia
{
    // A common "FramedSource" subclass, used for reading from a cached buffer list:

    Live555MediaInput::Live555MediaInput(UsageEnvironment& env)
        : Medium(env), connecting(false), video_callback(nullptr), audio_callback(nullptr), m_max_idr_size(0)
    {
    }

    Live555MediaInput::~Live555MediaInput()
    {
        LOG_FILE_FUNC_LINE();
        video_list.remove_if([](Source* s) {
            if (s->GetReadFdStatus()) {
                delete s;
                return true;
            } else {
                return false;
            }
        });

        audio_list.remove_if([](Source* s) {
            if (s->GetReadFdStatus()) {
                delete s;
                return true;
            } else {
                return false;
            }
        });

        muxer_list.remove_if([](Source* s) {
            if (s->GetReadFdStatus()) {
                delete s;
                return true;
            } else {
                return false;
            }
        });
        connecting = false;
    }

    void Live555MediaInput::Start(UsageEnvironment& env _UNUSED)
    {
        connecting = true;
        LOG_FILE_FUNC_LINE();
    }

    void Live555MediaInput::Stop(UsageEnvironment& env _UNUSED)
    {
        connecting = false;
        // TODO: tell main to stop all up flow resource
        LOG_FILE_FUNC_LINE();
    }

    Live555MediaInput* Live555MediaInput::createNew(UsageEnvironment& env)
    {
        return new Live555MediaInput(env);
    }

#define MAX_CACHE_NUMBER 60
    static void common_reduction(void* userdata, std::list<std::shared_ptr<MediaBuffer>>& mb_list)
    {
        Source* source = (Source*)userdata;
        if (mb_list.size() > source->GetCachedBufSize()) {
            for (unsigned i = 0; i < source->GetCachedBufSize() / 2; i++) {
                mb_list.pop_front();
                if (source) {
                    int j = 0;
                    ssize_t read_size = (ssize_t)sizeof(j);
                    ssize_t ret = read(source->GetReadFd(), &j, sizeof(j));
                    if (ret != read_size) {
                        LOG("%s:%d, read from pipe error, %m\n", __func__, __LINE__);
                    }
                }
            }
            LOG("call common_reduction.\n");
        }
    }

    static void h264_packet_reduction(void* userdata _UNUSED, std::list<std::shared_ptr<MediaBuffer>>& mb_list)
    {
        if (mb_list.size() < MAX_CACHE_NUMBER) {
            return;
        }
        // only remain one I frame
        auto i = mb_list.rbegin();
        for (; i != mb_list.rend(); ++i) {
            auto& b = *i;
            if (b->GetUserFlag() & MediaBuffer::kIntra) {
                break;
            }
        }
        if (i == mb_list.rend()) {
            return;
        }
        auto iter = mb_list.begin();
        for (; iter != mb_list.end(); ++iter) {
            auto& b = *iter;
            if (!(b->GetUserFlag() & MediaBuffer::kExtraIntra)) {
                break;
            }
        }
        LOG("h264 reduction before, num: %d \n", (int)mb_list.size());
        mb_list.erase(iter, (++i).base());
        LOG("h264 reduction after, num: %d \n", (int)mb_list.size());
    }

    FramedSource* Live555MediaInput::videoSource(CodecType c_type)
    {
        // if (!video_source)
        Source* source = new Source();
        if (!source) {
            return nullptr;
        }
        ListReductionPtr func;
        if (c_type == CODEC_TYPE_JPEG) {
            func = common_reduction;
        } else {
            func = h264_packet_reduction;
        }
        if (!source->Init(func)) {
            delete source;
            return nullptr;
        }
        video_list.push_back(source);
        if (c_type == CODEC_TYPE_JPEG) {
            return new CommonFramedSource(envir(), *source);
        } else {
            VideoFramedSource* video_source = new VideoFramedSource(envir(), *source);
            video_source->SetCodecType(c_type);
            return video_source;
        }
    }

    FramedSource* Live555MediaInput::audioSource()
    {
        Source* source = new Source();
        if (!source) {
            return nullptr;
        }
        if (!source->Init(common_reduction)) {
            delete source;
            return nullptr;
        }
        audio_list.push_back(source);
        ListSource* audio_source = new CommonFramedSource(envir(), *source);
        return audio_source;
    }

    FramedSource* Live555MediaInput::muxerSource()
    {
        Source* source = new Source();
        if (!source) {
            return nullptr;
        }
        if (!source->Init(common_reduction)) {
            delete source;
            return nullptr;
        }
        muxer_list.push_back(source);
        ListSource* muxer_source = new CommonFramedSource(envir(), *source);
        return muxer_source;
    }
#if 0
    static void printErr(UsageEnvironment& env, char const* str = NULL) {
        if(str != NULL) {
            env << str;
        }
        env << ": " << strerror(env.getErrno()) << "\n";
    }
#endif

    void Live555MediaInput::PushNewVideo(std::shared_ptr<MediaBuffer>& buffer)
    {
        if (!buffer) {
            return;
        }
        if ((buffer->GetUserFlag() & MediaBuffer::kIntra)) {
            if (m_max_idr_size < buffer->GetValidSize()) {
                m_max_idr_size = buffer->GetValidSize();
            }
        }
        video_list.remove_if([](Source* s) {
            if (s->GetReadFdStatus()) {
                delete s;
                return true;
            } else {
                return false;
            }
        });

        for (auto video : video_list) {
            if (video) {
                if (!video->GetReadFdStatus()) {
                    video->Push(buffer);
                }
            }
        }
    }

    void Live555MediaInput::PushNewAudio(std::shared_ptr<MediaBuffer>& buffer)
    {
        if (!buffer) {
            return;
        }
        audio_list.remove_if([](Source* s) {
            if (s->GetReadFdStatus()) {
                delete s;
                return true;
            } else {
                return false;
            }
        });
        for (auto audio : audio_list) {
            if (audio) {
                if (!audio->GetReadFdStatus()) {
                    audio->Push(buffer);
                }
            }
        }
    }

    void Live555MediaInput::PushNewMuxer(std::shared_ptr<MediaBuffer>& buffer)
    {
        if (!buffer) {
            return;
        }
        muxer_list.remove_if([](Source* s) {
            if (s->GetReadFdStatus()) {
                delete s;
                return true;
            } else {
                return false;
            }
        });
        for (auto muxer : muxer_list) {
            if (muxer) {
                if (!muxer->GetReadFdStatus()) {
                    muxer->SetCachedBufSize(buffer->GetValidSize());
                    muxer->Push(buffer);
                }
            }
        }
    }
    void Live555MediaInput::SetStartVideoStreamCallback(const StartStreamCallback& cb)
    {
        AutoLockMutex _alm(video_callback_mtx);
        video_callback = cb;
    }
    StartStreamCallback Live555MediaInput::GetStartVideoStreamCallback()
    {
        AutoLockMutex _alm(video_callback_mtx);
        return video_callback;
    }
    void Live555MediaInput::SetStartAudioStreamCallback(const StartStreamCallback& cb)
    {
        AutoLockMutex _alm(audio_callback_mtx);
        audio_callback = cb;
    }
    StartStreamCallback Live555MediaInput::GetStartAudioStreamCallback()
    {
        AutoLockMutex _alm(audio_callback_mtx);
        return audio_callback;
    }

    unsigned Live555MediaInput::getMaxIdrSize()
    {
        return (m_max_idr_size * 13 / 10) * 3 * 2 / 25;
    }
    Source::Source() : reduction(nullptr), m_cached_buffers_size(MAX_CACHE_NUMBER), m_read_fd_status(false)
    {
        wakeFds[0] = wakeFds[1] = -1;
        LOG("Source :: %p wakeFds[0] = %d, wakeFds[1]= %d.\n", this, wakeFds[0], wakeFds[1]);
    }

    void Source::CloseReadFd()
    {
        if (wakeFds[0] >= 0) {
            m_read_fd_status = true;
        }
    }

    Source::~Source()
    {
        if (wakeFds[0] >= 0) {
            ::close(wakeFds[0]);
            wakeFds[0] = -1;
        }
        if (wakeFds[1] >= 0) {
            ::close(wakeFds[1]);
            wakeFds[1] = -1;
        }
        LOG("~Source::%p remain %d buffers, will auto release\n", this, (int)cached_buffers.size());
    }

    bool Source::Init(ListReductionPtr func)
    {
        // create pipe fds
        int ret = pipe2(wakeFds, O_CLOEXEC);
        if (ret) {
            LOG("pipe2 failed: %m\n");
            return false;
        }
        assert(wakeFds[0] >= 0 && wakeFds[1] >= 0);
        reduction = func;
        return true;
    }

    void Source::Push(std::shared_ptr<MediaBuffer>& buffer)
    {
        AutoLockMutex _alm(mtx);
        if (reduction) {
            reduction(this, cached_buffers);
        }
        cached_buffers.push_back(buffer);
        // mtx.notify();
        int i = 0;
        ssize_t count = write(wakeFds[1], &i, sizeof(i));
        if (count < 0) {
            LOG("write failed: %s, %p, fd = %d\n", strerror(errno), this, wakeFds[1]);
        }
    }

    std::shared_ptr<MediaBuffer> Source::Pop()
    {
        AutoLockMutex _alm(mtx);
        if (cached_buffers.empty()) {
            return nullptr;
        }
        auto buffer = cached_buffers.front();
        cached_buffers.pop_front();
        return std::move(buffer);
    }

    void Source::SetCachedBufSize(size_t one_buf_size)
    {
        // max: 5 M/s
        if (one_buf_size > 0) {
            m_cached_buffers_size = 1024 * 1024 * 5 / one_buf_size;
        }
    }
    void ListSource::doGetNextFrame()
    {
        assert(fSource.GetReadFd() >= 0);
        // Await the next incoming data on our FID:
        envir().taskScheduler().turnOnBackgroundReadHandling(
            fSource.GetReadFd(), (TaskScheduler::BackgroundHandlerProc*)&incomingDataHandler, this);
    }

    void ListSource::doStopGettingFrames()
    {
        LOG_FILE_FUNC_LINE();
        FramedSource::doStopGettingFrames();
    }

    void ListSource::incomingDataHandler(ListSource* source, int /*mask*/)
    {
        source->incomingDataHandler1();
    }

    void ListSource::incomingDataHandler1()
    {
        // Read the data from our file into the client's buffer:
        readFromList();

        assert(fSource.GetReadFd() >= 0);
        // Stop handling any more input, until we're ready again:
        envir().taskScheduler().turnOffBackgroundReadHandling(fSource.GetReadFd());

        // Tell our client that we have new data:
        afterGetting(this);
    }

    bool ListSource::readFromList(bool flush _UNUSED)
    {
        return false;
    }

    void ListSource::flush()
    {
        readFromList(true);
        fFrameSize = 0;
        fNumTruncatedBytes = 0;
    }

    VideoFramedSource::VideoFramedSource(UsageEnvironment& env, Source& source)
        : ListSource(env, source), got_iframe(false)
    {
        // fReadFd = input.vs->GetReadFd();
    }

    VideoFramedSource::~VideoFramedSource()
    {
        LOG_FILE_FUNC_LINE();
        // fInput.video_source = NULL;
    }

    bool VideoFramedSource::readFromList(bool flush _UNUSED)
    {
#ifdef DEBUG_SEND
        fprintf(stderr, "$$$$ %s, %d\n", __func__, __LINE__);
#endif
        std::shared_ptr<MediaBuffer> buffer;
        int i = 0;
        ssize_t read_size = (ssize_t)sizeof(i);
        ssize_t ret = read(fSource.GetReadFd(), &i, sizeof(i));
        if (ret != read_size) {
            LOG("video %s:%d, fd = %d. read from pipe error, %m\n", __func__, __LINE__, fSource.GetReadFd());
            envir() << __LINE__ << " read from pipe error: " << errno << "\n";
            goto err;
        }

        buffer = fSource.Pop();
        if (buffer) {
            if (!got_iframe) {
                got_iframe = buffer->GetUserFlag() & MediaBuffer::kIntra;
                if (!got_iframe && !(buffer->GetUserFlag() & MediaBuffer::kExtraIntra)) {
                    goto err;
                }
            }
            fPresentationTime = buffer->GetTimeVal();
            fPresentationTime.tv_sec += 1;
// gettimeofday(&fPresentationTime, NULL);
#ifdef DEBUG_SEND
            fprintf(stderr, "video frame time: %ld, %ld.\n", fPresentationTime.tv_sec, fPresentationTime.tv_usec);
#endif
            fFrameSize = buffer->GetValidSize();
#ifdef DEBUG_SEND
            envir() << "video frame size: " << fFrameSize << "\n";
#endif
            assert(fFrameSize > 0);
            uint8_t* p = (uint8_t*)buffer->GetPtr();
            if (buffer->GetUserFlag() & MediaBuffer::kIntra) {
                int intra_size = 0;
                uint8_t* intra_ptr = (uint8_t*)GetIntraFromBuffer(buffer, intra_size, codec_type);
                assert(intra_ptr);
                assert(intra_size > 0);
                p = intra_ptr;
                fFrameSize = intra_size;
            }
            assert(p[0] == 0);
            assert(p[1] == 0);
            if (p[2] == 0) {
                assert(p[3] == 1);
                read_size = 4;
            } else {
                assert(p[2] == 1);
                read_size = 3;
            }
            fFrameSize -= read_size;
            if (fFrameSize > fMaxSize) {
                LOG("%s : %d, fFrameSize(%d) > fMaxSize(%d)\n", __func__, __LINE__, fFrameSize, fMaxSize);
                fNumTruncatedBytes = fFrameSize - fMaxSize;
                fFrameSize = fMaxSize;
            } else {
                fNumTruncatedBytes = 0;
            }
            memcpy(fTo, p + read_size, fFrameSize);
            return true;
        }

    err:
        fFrameSize = 0;
        fNumTruncatedBytes = 0;
        return false;
    }

    CommonFramedSource::CommonFramedSource(UsageEnvironment& env, Source& source) : ListSource(env, source)
    {
        // fReadFd = input.as->GetReadFd();
    }

    CommonFramedSource::~CommonFramedSource()
    {
        LOG_FILE_FUNC_LINE();
        // fInput.audio_source = NULL;
    }

#ifdef DEBUG_FRAME
    static unsigned sg_fFramesize = 0;
    static long sg_lastTvsec = 0;
#endif
    bool CommonFramedSource::readFromList(bool flush _UNUSED)
    {
#ifdef DEBUG_SEND
        fprintf(stderr, "$$$$ %s, %d\n", __func__, __LINE__);
#endif
        std::shared_ptr<MediaBuffer> buffer;
        uint8_t* p;

        int i = 0;
        ssize_t read_size = (ssize_t)sizeof(i);
        ssize_t ret = read(fSource.GetReadFd(), &i, sizeof(i));
        if (ret != read_size) {
            LOG("common %s:%d, fd = %d. read from pipe error, %m\n", __func__, __LINE__, fSource.GetReadFd());
            envir() << __LINE__ << " read from pipe error: " << errno << "\n";
            goto err;
        }

        buffer = fSource.Pop();
        if (buffer) {
            p = (uint8_t*)buffer->GetPtr();
            fPresentationTime = buffer->GetTimeVal();
            fPresentationTime.tv_sec += 1;
#ifdef DEBUG_SEND
            fprintf(stderr, "common frame time: %ld, %ld.\n", fPresentationTime.tv_sec, fPresentationTime.tv_usec);
#endif
            fFrameSize = buffer->GetValidSize();
#ifdef DEBUG_FRAME
            sg_fFramesize += fFrameSize;
            struct timeval t_now;
            gettimeofday(&t_now, NULL);
            if (sg_lastTvsec != t_now.tv_sec) {
                sg_lastTvsec = t_now.tv_sec;
                envir() << "RTSP::audio frame in one sec is: " << sg_fFramesize << "\n";
                sg_fFramesize = 0;
            }
#endif
#ifdef DEBUG_SEND
            envir() << "audio frame size: " << fFrameSize << "\n";
#endif
            assert(fFrameSize > 0);
            if (fFrameSize > fMaxSize) {
                LOG("%s : %d, fFrameSize(%d) > fMaxSize(%d)\n", __func__, __LINE__, fFrameSize, fMaxSize);
                fNumTruncatedBytes = fFrameSize - fMaxSize;
                fFrameSize = fMaxSize;
            } else {
                fNumTruncatedBytes = 0;
            }
            memcpy(fTo, p, fFrameSize);
            return true;
        }

    err:
        fFrameSize = 0;
        fNumTruncatedBytes = 0;
        return false;
    }

} // namespace easymedia