/*
 *******************************************************************************
 *                                   INCLUDE
 *******************************************************************************
 */
#include "RTMPDemuxHandler.h"
#include "Logger.h"
#include "../core/System.h"
#include "util/Math.h"
#include <list>

namespace Video
{
namespace MediaServer
{
/*
 *******************************************************************************
 *                                CLASS DEFINE
 *******************************************************************************
 */
RTMPDemuxHandler::RTMPDemuxHandler(const std::string& uri)
    : Core::DemuxHandler(Poco::format("MediaServer::DemuxHandler(%s)", uri), uri)
{
}

RTMPDemuxHandler::~RTMPDemuxHandler()
{
}

void RTMPDemuxHandler::run(Core::VideoEncoder::Ptr pVideoEncoder, Core::AudioEncoder::Ptr pAudioEncoder,
                           const Post& post, std::atomic<bool>& isCancelled)
{
    _pVideoEncoder = pVideoEncoder;
    _pAudioEncoder = pAudioEncoder;
    _post          = post;

    Poco::URI   uriParser(uri());
    std::string scheme = uriParser.getScheme();
    std::string host   = uriParser.getHost();
    short       port   = uriParser.getPort();
    port               = port <= 0 ? 1935 : port;
    std::string path   = uriParser.getPath();

    if (!path.empty() && path[0] == '/')
        path = path.substr(1);

    size_t pos = path.find_first_of('/');
    if (pos == std::string::npos)
    {
        logger().error("%s: Invalid URL: %s", name(), uri());
        Poco::Thread::sleep(2000);
        return;
    }

    std::string turl     = scheme + "://" + host + ":" + std::to_string(port) + "/" + path.substr(0, pos);
    std::string appname  = path.substr(0, pos);
    std::string playpath = path.substr(pos + 1);

    struct rtmp_client_handler_t handler;
    handler.send     = RTMPDemuxHandler::clientSend;
    handler.onaudio  = RTMPDemuxHandler::clientOnAudio;
    handler.onvideo  = RTMPDemuxHandler::clientOnVideo;
    handler.onscript = RTMPDemuxHandler::clientOnScript;

    const size_t             buffsz  = 2 * 1024 * 1024;
    Poco::SharedPtr<uint8_t> pBuffer = new uint8_t[buffsz];
    int                      rc;
    while (!isCancelled)
    {
        _audioInfoParsed = false;
        _videoInfoParsed = false;

        while (!isCancelled)
        {
            _socket = socket_connect_host(host.c_str(), port, 2000);
            if (_socket != socket_invalid)
            {
                socket_setnonblock(_socket, 0);
                break;
            }

            Poco::Thread::sleep(2000);
            continue;
        }

        _pCient = rtmp_client_create(appname.c_str(), playpath.c_str(), turl.c_str(), this, &handler);
        if (!_pCient)
        {
            logger().error("%s: Create RTMP-Client FAIL!(URL: %s)", name(), uri());
            Poco::Thread::sleep(2000);
            goto rtmp_create_fail;
        }

        _pDemuxer = flv_demuxer_create(RTMPDemuxHandler::flvDemuxerHandler, this);
        if (!_pDemuxer)
        {
            logger().error("%s: Create FLV Debuxer FAIL!(URL: %s)", name(), uri());
            Poco::Thread::sleep(2000);
            goto demuxer_fail;
        }

        rc = rtmp_client_start(_pCient, 1);
        if (rc != 0)
        {
            logger().error("%s: Start RTMP-Client FAIL!(URL: %s)", name(), uri());
            Poco::Thread::sleep(2000);
            goto rmtp_start_fail;
        }

        while (!isCancelled)
        {
            rc = socket_recv(_socket, pBuffer.get(), buffsz, 0);
            if (rc <= 0)
            {
                logger().error("%s: Socket recv FAIL: %d(URL: %s)", name(), rc, uri());
                break;
            }

            rc = rtmp_client_input(_pCient, pBuffer.get(), rc);
            if (rc != 0)
            {
                logger().error("%s: RTMP client input FAIL: %d(URL: %s)", name(), rc, uri());
                break;
            }
        }

        rtmp_client_stop(_pCient);

    rmtp_start_fail:
        flv_demuxer_destroy(_pDemuxer);

    demuxer_fail:
        rtmp_client_destroy(_pCient);

    rtmp_create_fail:
        socket_close(_socket);
        _socket = socket_invalid;
        Poco::Thread::sleep(2000);
    }
}

int RTMPDemuxHandler::parseFLVMetaData(const uint8_t* data, size_t bytes)
{
    double audiocodecid    = 0;
    double audiodatarate   = 0;
    double audiodelay      = 0;
    double audiosamplerate = 0;
    double audiosamplesize = 0;
    int    stereo          = 0;
    double videocodecid    = 0;
    double videodatarate   = 0;
    double framerate       = 0;
    double height          = 0;
    double width           = 0;

    struct amf_object_item_t prop[16];
    struct amf_object_item_t items[1];
    char                     buffer[64];

#define AMF_OBJECT_ITEM_VALUE(v, amf_type, amf_name, amf_value, amf_size) \
    { \
        v.type  = amf_type; \
        v.name  = amf_name; \
        v.value = amf_value; \
        v.size  = amf_size; \
    }
    int i = 0;
    AMF_OBJECT_ITEM_VALUE(prop[i], AMF_NUMBER, "audiocodecid", &audiocodecid, sizeof(audiocodecid));
    i++;
    AMF_OBJECT_ITEM_VALUE(prop[i], AMF_NUMBER, "audiodatarate", &audiodatarate, sizeof(audiodatarate));
    i++;
    AMF_OBJECT_ITEM_VALUE(prop[i], AMF_NUMBER, "audiodelay", &audiodelay, sizeof(audiodelay));
    i++;
    AMF_OBJECT_ITEM_VALUE(prop[i], AMF_NUMBER, "audiosamplerate", &audiosamplerate, sizeof(audiosamplerate));
    i++;
    AMF_OBJECT_ITEM_VALUE(prop[i], AMF_NUMBER, "audiosamplesize", &audiosamplesize, sizeof(audiosamplesize));
    i++;
    AMF_OBJECT_ITEM_VALUE(prop[i], AMF_BOOLEAN, "stereo", &stereo, sizeof(stereo));
    i++;
    AMF_OBJECT_ITEM_VALUE(prop[i], AMF_NUMBER, "videocodecid", &videocodecid, sizeof(videocodecid));
    i++;
    AMF_OBJECT_ITEM_VALUE(prop[i], AMF_NUMBER, "videodatarate", &videodatarate, sizeof(videodatarate));
    i++;
    AMF_OBJECT_ITEM_VALUE(prop[i], AMF_NUMBER, "framerate", &framerate, sizeof(framerate));
    i++;
    AMF_OBJECT_ITEM_VALUE(prop[i], AMF_NUMBER, "height", &height, sizeof(height));
    i++;
    AMF_OBJECT_ITEM_VALUE(prop[i], AMF_NUMBER, "width", &width, sizeof(width));
    i++;

    AMF_OBJECT_ITEM_VALUE(items[0], AMF_OBJECT, "onMetaData", prop, i);

    const uint8_t* end = data + bytes;
    if (AMF_STRING != data[0] || NULL == (data = AMFReadString(data + 1, end, 0, buffer, sizeof(buffer) - 1)))
        return -1;

    if (0 == strcmp(buffer, "@setDataFrame"))
    {
        if (AMF_STRING != data[0] || NULL == (data = AMFReadString(data + 1, end, 0, buffer, sizeof(buffer) - 1)))
            return -1;
    }

    if (0 != strcmp(buffer, "onMetaData"))
        return 0;

    if (amf_read_items(data, end, items, sizeof(items) / sizeof(items[0])))
    {
        if (framerate > 0)
            _pVideoEncoder->setFrameRate(framerate > 60 ? 60 : framerate);

        if (videodatarate > 0)
            _pVideoEncoder->setBitrate((int64_t)videodatarate);

        if (audiodatarate > 0)
            _pAudioEncoder->setBitrate((int64_t)audiodatarate);
    }

    return 0;
}

inline const uint8_t* RTMPDemuxHandler::annexBStartcode(const uint8_t* data, size_t bytes)
{
    size_t i;
    for (i = 2; i + 1 < bytes; i++)
    {
        if (0x01 == data[i] && 0x00 == data[i - 1] && 0x00 == data[i - 2])
            return data + i + 1;
    }

    return nullptr;
}

bool RTMPDemuxHandler::isH264BFrame(const void* h264, size_t bytes)
{
    ptrdiff_t      n;
    const uint8_t *p, *next, *end;

    end = (const uint8_t*)h264 + bytes;
    p   = annexBStartcode((const uint8_t*)h264, bytes);

    while (p)
    {
        next = annexBStartcode(p, end - p);
        if (next)
            n = next - p - 3;
        else
            n = end - p;

        while (n > 0 && 0 == p[n - 1])
            n--;

        assert(n > 0);
        if (n > 0)
        {
            if ((p[0] & 0x1f) == 1)
            {
                bitstream_t stream;
                bitstream_init(&stream, (const uint8_t*)(p + 1), n - 1);
                bitstream_read_ue(&stream);
                int32_t sliceType = bitstream_read_ue(&stream);
                if (sliceType == 1)
                    return true;
            }
        }
        p = next;
    }
    return false;
}

int RTMPDemuxHandler::processH264Data(const void* data, size_t bytes, uint32_t pts, uint32_t dts, int flags)
{
    if (!_videoInfoParsed)
        mpeg4_h264_annexb_nalu(data, bytes, RTMPDemuxHandler::h264NALhandler, this);

    enqueueVideoPacket(data, bytes, pts, dts, flags);
    return 0;
}

int RTMPDemuxHandler::processH265Data(const void* data, size_t bytes, uint32_t pts, uint32_t dts, int flags)
{
    if (!_videoInfoParsed)
        mpeg4_h264_annexb_nalu(data, bytes, RTMPDemuxHandler::h265NALhandler, this);

    enqueueVideoPacket(data, bytes, pts, dts, flags);
    return 0;
}

int RTMPDemuxHandler::processASCData(const void* data, size_t bytes, uint32_t pts, uint32_t dts, int flags)
{
    struct mpeg4_aac_t aac;
    if (mpeg4_aac_audio_specific_config_load((const uint8_t*)data, bytes, &aac) >= 0)
    {
        _pAudioEncoder->setCodecID(Core::CodecID::AAC);
        _pAudioEncoder->setFrameSize(1024);
        _pAudioEncoder->setSampleFormat(Core::SampleFormat::FLTP);
        _pAudioEncoder->setSampleRate((int)aac.sampling_frequency);
        _pAudioEncoder->setChannelLayout(aac.channels == 2 ? Core::ChannelLayout::STEREO : Core::ChannelLayout::MONO);
        _pAudioEncoder->commit();
        logger().information("%s: Audio found: AAC %dHz %s", name(), (int)aac.sampling_frequency,
                             Core::channelLayoutToString(_pAudioEncoder->getChannelLayout()));
        ;
    }

    return 0;
}

int RTMPDemuxHandler::processAACData(const void* data, size_t bytes, uint32_t pts, uint32_t dts, int flags)
{
    enqueueAudioPacket(data, bytes, pts, dts, flags);
    return 0;
}

int RTMPDemuxHandler::processOpusHeadData(const void* data, size_t bytes, uint32_t pts, uint32_t dts, int flags)
{
    struct opus_head_t opus;
    if (opus_head_load((const uint8_t*)data, bytes, &opus) >= 0)
    {
        _pAudioEncoder->setCodecID(Core::CodecID::OPUS);
        _pAudioEncoder->setFrameSize(960);
        _pAudioEncoder->setSampleFormat(Core::SampleFormat::S16);
        _pAudioEncoder->setSampleRate((int)opus.input_sample_rate);
        _pAudioEncoder->setChannelLayout(opus_head_channels(&opus) == 2 ? Core::ChannelLayout::STEREO
                                                                        : Core::ChannelLayout::MONO);
        _pAudioEncoder->commit();
        logger().information("%s: Audio found: OPUS %dHz %s", name(), (int)opus.input_sample_rate,
                             Core::channelLayoutToString(_pAudioEncoder->getChannelLayout()));
    }

    return 0;
}

int RTMPDemuxHandler::processOpusData(const void* data, size_t bytes, uint32_t pts, uint32_t dts, int flags)
{
    enqueueAudioPacket(data, bytes, pts, dts, flags);
    return 0;
}

int RTMPDemuxHandler::processG711AData(const void* data, size_t bytes, uint32_t pts, uint32_t dts, int flags)
{
    if (!_audioInfoParsed)
    {
        _pAudioEncoder->setCodecID(Core::CodecID::G711A);
        _pAudioEncoder->setFrameSize(1024);
        _pAudioEncoder->setSampleFormat(Core::SampleFormat::S16P);
        _pAudioEncoder->setSampleRate((int)8000);
        _pAudioEncoder->setChannelLayout(Core::ChannelLayout::MONO);
        _pAudioEncoder->commit();
        logger().information("%s: Audio found: G711A %dHz %s", name(), 8000, std::string("MONO"));

        _audioInfoParsed = true;
    }

    enqueueAudioPacket(data, bytes, pts, dts, flags);
    return 0;
}

int RTMPDemuxHandler::processG711UData(const void* data, size_t bytes, uint32_t pts, uint32_t dts, int flags)
{
    if (!_audioInfoParsed)
    {
        _pAudioEncoder->setCodecID(Core::CodecID::G711U);
        _pAudioEncoder->setFrameSize(1024);
        _pAudioEncoder->setSampleFormat(Core::SampleFormat::S16P);
        _pAudioEncoder->setSampleRate((int)8000);
        _pAudioEncoder->setChannelLayout(Core::ChannelLayout::MONO);
        _pAudioEncoder->commit();
        logger().information("%s: Audio found: G711U %dHz %s", name(), 8000, std::string("MONO"));

        _audioInfoParsed = true;
    }

    enqueueAudioPacket(data, bytes, pts, dts, flags);
    return 0;
}

void RTMPDemuxHandler::enqueueVideoPacket(const void* data, size_t bytes, uint32_t pts, uint32_t dts, int flags)
{
    uint8_t* pBuffer = new uint8_t[bytes];
    memcpy(pBuffer, data, bytes);

    uint64_t localPTS = pts;
    uint64_t localDTS = dts;

    adjustTimestamp(localPTS, localDTS);
    updateTimeline(localPTS);

    _post(new Core::Packet(_pVideoEncoder->code(),
                           Core::Packet::FLAG_IS_VIDEO | (flags ? Core::Packet::FLAG_IS_KEY : 0), pBuffer, bytes,
                           localPTS, localDTS),
          Core::MediaType::VIDEO);
}

void RTMPDemuxHandler::enqueueAudioPacket(const void* data, size_t bytes, uint32_t pts, uint32_t dts, int flags)
{
    uint8_t* pBuffer = new uint8_t[bytes];
    memcpy(pBuffer, data, bytes);

    uint64_t localPTS = pts;
    uint64_t localDTS = dts;

    adjustTimestamp(localPTS, localDTS);
    updateTimeline(localPTS);
    _post(new Core::Packet(_pAudioEncoder->code(), flags, pBuffer, bytes, localPTS, localDTS), Core::MediaType::AUDIO);
}

int RTMPDemuxHandler::clientSend(void* param, const void* header, size_t len, const void* data, size_t bytes)
{
    socket_bufvec_t vec[2];
    socket_setbufvec(vec, 0, (void*)header, len);
    socket_setbufvec(vec, 1, (void*)data, bytes);
    return socket_send_v_all_by_time(((RTMPDemuxHandler*)param)->_socket, vec, bytes ? 2 : 1, 0, 2000);
}

int RTMPDemuxHandler::clientOnAudio(void* param, const void* data, size_t bytes, uint32_t timestamp)
{
    return flv_demuxer_input(((RTMPDemuxHandler*)param)->_pDemuxer, FLV_TYPE_AUDIO, data, bytes, timestamp);
}

int RTMPDemuxHandler::clientOnVideo(void* param, const void* data, size_t bytes, uint32_t timestamp)
{
    return flv_demuxer_input(((RTMPDemuxHandler*)param)->_pDemuxer, FLV_TYPE_VIDEO, data, bytes, timestamp);
}

int RTMPDemuxHandler::clientOnScript(void* param, const void* data, size_t bytes, uint32_t timestamp)
{
    return flv_demuxer_input(((RTMPDemuxHandler*)param)->_pDemuxer, FLV_TYPE_SCRIPT, data, bytes, timestamp);
}

int RTMPDemuxHandler::flvDemuxerHandler(void* param, int codec, const void* data, size_t bytes, uint32_t pts,
                                        uint32_t dts, int flags)
{
    int               rc;
    RTMPDemuxHandler* pHandler = (RTMPDemuxHandler*)param;

    switch (codec)
    {
    case FLV_SCRIPT_METADATA:
        rc = pHandler->parseFLVMetaData((const uint8_t*)data, bytes);
        break;

    case FLV_VIDEO_AVCC:
        break;

    case FLV_VIDEO_H264:
        rc = pHandler->processH264Data(data, bytes, pts, dts, flags);
        break;

    case FLV_VIDEO_HVCC:
        break;

    case FLV_VIDEO_H265:
        rc = pHandler->processH265Data(data, bytes, pts, dts, flags);
        break;

    case FLV_AUDIO_ASC:
        rc = pHandler->processASCData(data, bytes, pts, dts, flags);
        break;

    case FLV_AUDIO_AAC:
        rc = pHandler->processAACData(data, bytes, pts, dts, flags);
        break;

    case FLV_AUDIO_OPUS_HEAD:
        rc = pHandler->processOpusHeadData(data, bytes, pts, dts, flags);
        break;

    case FLV_AUDIO_OPUS:
        rc = pHandler->processOpusData(data, bytes, pts, dts, flags);
        break;

    case FLV_AUDIO_G711A:
        rc = pHandler->processG711AData(data, bytes, pts, dts, flags);
        break;

    case FLV_AUDIO_G711U:
        rc = pHandler->processG711UData(data, bytes, pts, dts, flags);
        break;

    default:
        logger().warning("%s: UNIMPLEMENTED FLV-CODEC type: %d(%#x)", pHandler->name(), codec, (uint32_t)codec);
        break;
    }

    (void)rc;
    return 0;
}

void RTMPDemuxHandler::h264NALhandler(void* param, const uint8_t* nalu, size_t bytes)
{
    RTMPDemuxHandler* pHandler = (RTMPDemuxHandler*)param;

    struct h264_sps_t sps;
    if (h264_sps_parse(nalu, bytes, &sps) == 0)
    {
        int x, y, width, height;

        h264_display_rect(&sps, &x, &y, &width, &height);

        auto pEncoder = pHandler->_pVideoEncoder;
        pEncoder->setCodecID(Core::CodecID::H264);
        pEncoder->setWidth(width);
        pEncoder->setHeight(height);
        pEncoder->commit();

        pHandler->_videoInfoParsed = true;

        logger().information("%s: Video found: H.264 %dx%d", pHandler->name(), width, height);
    }
}

void RTMPDemuxHandler::h265NALhandler(void* param, const uint8_t* nalu, size_t bytes)
{
    RTMPDemuxHandler* pHandler = (RTMPDemuxHandler*)param;

    struct h265_sps_t sps;
    if (h265_sps_parse(nalu, bytes, &sps) == 0)
    {
        int x, y, width, height;

        h265_display_rect(&sps, &x, &y, &width, &height);

        auto pEncoder = pHandler->_pVideoEncoder;
        pEncoder->setCodecID(Core::CodecID::H265);
        pEncoder->setWidth(width);
        pEncoder->setHeight(height);
        pEncoder->commit();

        pHandler->_videoInfoParsed = true;

        logger().information("%s: Video found: H.265 %dx%d", pHandler->name(), width, height);
    }
}

void RTMPDemuxHandler::adjustTimestamp(uint64_t& pts, uint64_t& dts)
{
    pts *= 1000ull;
    dts *= 1000ull;

    auto drift = getTimeDrift(pts);
    pts += drift;
    dts += drift;
}

void RTMPDemuxHandler::updateTimeline(uint64_t ts)
{
    if (getTimeline() < ts)
        setTimeline(ts);
}

void RTMPDemuxHandler::seek(uint64_t ts)
{
}
}
}