
#include "MFSPublic.h"
#include "Rtp808Service.h"
#include "MediaRtmpProtocol.h"
#include "3rd/cJSON.h"
#include "AppSetting.h"

namespace mfs
{
	const int kChunkMessageHeaderLen[4] = { 11, 7, 3, 0 };

    static bool ClearRtpBuffer(std::list<BaseWriteBuffer>& buffers, int& bufferSize)
    {
        buffers.clear();
        bufferSize = 0;
        return false;

        bool needClear = false;
        bool hasVideo = false;
        int indexIFrame = -1;
        int icount = 0;
        for (auto iter = buffers.begin(); iter != buffers.end(); iter++,icount++)
        {
            Rtp808HeadBaseV* head = reinterpret_cast<Rtp808HeadBaseV*>((uint8_t*)(iter->data.data() + iter->dataOffset));
            if (head->dataType == RTP_808_DATATYPE_I || head->dataType == RTP_808_DATATYPE_P || head->dataType == RTP_808_DATATYPE_B)
            {
                hasVideo = true;
            }
            if (head->dataType == RTP_808_DATATYPE_I && icount>0)
            {
                indexIFrame = icount;
                break;
            }
        }
        if (!hasVideo)
        {
            // logd("Clear all audio");
            LOG_DEBUG << "Clear all audio" <<LOG_ENDL;
            buffers.clear();
        }
        else
        {
            if (indexIFrame == -1)
            {
                // 缺少I帧还要继续清除
                LOG_DEBUG << "Clear all p frame" <<LOG_ENDL;
                buffers.clear();
                needClear = true;
            }
            else
            {
                LOG_DEBUG << "Clear to i frame" <<LOG_ENDL;
                for (size_t i = 0; i < indexIFrame; i++)
                {
                    buffers.pop_front();
                }
                
            }
        }
        return needClear;
    }
    static bool IsKeyFrame(const uint8_t* data, uint32_t size)
    {
        int startCode = 0;
        if (data[0] == 0 && data[1] == 0 && data[2] == 1) {
            startCode = 3;
        }
        else if (data[0] == 0 && data[1] == 0 && data[2] == 0 && data[3] == 1) {
            startCode = 4;
        }

        int type = data[startCode] & 0x1f;
        if (type == 5 || type == 7)  {  // sps_pps_idr or idr 
            return true;
        }
        
        return false;
    }

    typedef std::pair<uint8_t*, uint8_t*> Nal; // <nal begin, nal end>
    Nal findNal(const uint8_t *data, uint32_t size)
    {
        Nal nal(nullptr, nullptr);

        if(size < 5)
        {
            return nal;
        }

        nal.second = const_cast<uint8_t*>(data) + (size-1);

        //uint32_t startCode = 0;
        uint32_t pos = 0;
        uint8_t prefix[3] = {0};
        memcpy(prefix, data, 3);
        size -= 3;
        data += 2;

        while(size--)
        {
            if ((prefix[pos % 3] == 0) && (prefix[(pos + 1) % 3] == 0) && (prefix[(pos + 2) % 3] == 1))
            {
                if(nal.first == nullptr) // 00 00 01 
                {
                    nal.first = const_cast<uint8_t*>(data) + 1;
                    //startCode = 3;
                }
                else if(data  > nal.first + 3)
                {
                    nal.second = const_cast<uint8_t*>(data) - 3;
                    break;
                }               
            }
            else if ((prefix[pos % 3] == 0) && (prefix[(pos + 1) % 3] == 0) && (prefix[(pos + 2) % 3] == 0))
            {
                if (*(data+1) == 0x01) // 00 00 00 01 
                {              
                    if(nal.first == nullptr)
                    {
                        if(size >= 1)
                        {
                            nal.first = const_cast<uint8_t*>(data) + 2;
                        }                       
                        else
                        {
                            break;  
                        }                  
                        //startCode = 4;
                    }
                    else //if(startCode == 4)
                    {
                        nal.second = const_cast<uint8_t*>(data) - 3;
                        break;
                    }                    
                }
            }

            prefix[(pos++) % 3] = *(++data);
        }
            
        if(nal.first == nullptr)
            nal.second = nullptr;

        return nal;
    }



    int MediaRtmpProtocol::GetAACSampleRateIndex(int sampleRate){
        for (int i = 0; i < 16; i++)
        {
            if (kSamplingFrequency[i] == sampleRate)
            {
                return i;
            }
            
        }
        return -1;
    }



    bool MediaRtmpProtocol::sendAudioData(uint64_t timestamp, const std::vector<RtmpBuffer>& payload, uint32_t payload_size)
    {
        // logv("sendVideoData");
        if (payload_size == 0) {
            return false;
        }
        
        RtmpMessage rtmp_msg;
        rtmp_msg.type_id = RTMP_AUDIO;
        rtmp_msg._timestamp = timestamp;
        rtmp_msg.stream_id = stream_id_;
        rtmp_msg.payloadVec = payload;
        rtmp_msg.length = payload_size;
        sendRtmpChunks(RTMP_CHUNK_AUDIO_ID, rtmp_msg);

        return true;
    }
    bool MediaRtmpProtocol::sendAudioData(uint64_t timestamp, const RtmpBuffer& payload)
    {
        std::vector<RtmpBuffer> lst;
        lst.push_back(payload);
        return sendAudioData(timestamp, lst, payload.size);
    }
    bool MediaRtmpProtocol::sendVideoData(uint64_t timestamp, const std::vector<RtmpBuffer>& payload, uint32_t payload_size)
    {
        // logv("sendVideoData");
        if (payload_size == 0) {
            return false;
        }
        
        RtmpMessage rtmp_msg;
        rtmp_msg.type_id = RTMP_VIDEO;
        rtmp_msg._timestamp = timestamp;
        rtmp_msg.stream_id = stream_id_;
        rtmp_msg.payloadVec = payload;
        rtmp_msg.length = payload_size;
        sendRtmpChunks(RTMP_CHUNK_VIDEO_ID, rtmp_msg);

        return true;
    }
    bool MediaRtmpProtocol::sendVideoData(uint64_t timestamp, const RtmpBuffer& payload)
    {
        std::vector<RtmpBuffer> lst;
        lst.push_back(payload);
        return sendVideoData(timestamp, lst, payload.size);
    }
    int MediaRtmpProtocol::pushAudioFrame(const std::vector<RtmpBuffer>& data, int dataSize)
    {
        // logd("");
        uint64_t timestamp = timestamp_.elapsed();
        std::vector<char> payload;
        int index = 0;
        payload.resize(2);
        payload[index++] = audio_tag_;
        if (media_info_a.audio_codec_id == RTMP_CODEC_ID_AAC)
        {
            payload[index++] = 1; // 0: aac sequence header, 1: aac raw data
        }
        
        std::vector<RtmpBuffer> lst = data;
        lst.insert(lst.begin(), RtmpBuffer(payload.data(), index));
        sendAudioData(timestamp, lst, index + dataSize);
        return 0;
    }
    int MediaRtmpProtocol::pushAudioFrame(const RtmpBuffer& data)
    {
        std::vector<RtmpBuffer> lst;
        lst.push_back(data);
        return pushAudioFrame(lst, data.size);
    }
    int MediaRtmpProtocol::pushVideoFrame(const std::vector<RtmpBuffer>& data, int dataSize)
    {
        // logv("pushVideoFrame");
        auto isKeyFrame =IsKeyFrame(data[0].data, data[0].size);
        if (!has_key_frame_) {
            if (isKeyFrame) {
                logv("send key frame");
                has_key_frame_ = true;
                timestamp_.reset();
                sendVideoData(0, RtmpBuffer(avc_sequence_header_.data(), avc_sequence_header_size_));
            }
            else {
                return 0;
            }
        }

        uint64_t timestamp = timestamp_.elapsed();
        std::vector<char> payload;
        payload.resize(4096);

        uint8_t *buffer = (uint8_t *)payload.data();
        uint32_t index = 0;
        buffer[index++] = isKeyFrame ? 0x17: 0x27;
        buffer[index++] = 1;

        buffer[index++] = 0;
        buffer[index++] = 0;
        buffer[index++] = 0;

        buffer[index++] = (dataSize >> 24) & 0xff;
        buffer[index++] = (dataSize >> 16) & 0xff;
        buffer[index++] = (dataSize >> 8) & 0xff;
        buffer[index++] = (dataSize) & 0xff;

        std::vector<RtmpBuffer> lst = data;
        lst.insert(lst.begin(), RtmpBuffer(payload.data(), index));
        sendVideoData(timestamp, lst, index + dataSize);
        return 0;
    }
    int MediaRtmpProtocol::pushVideoFrame(const RtmpBuffer& data)
    {
        std::vector<RtmpBuffer> lst;
        lst.push_back(data);
        return pushVideoFrame(lst, data.size);
    }
    int MediaRtmpProtocol::setMediaInfoV(const MediaInfo& media_info)
    {
        media_info_v = media_info;

        if (media_info_v.video_codec_id == RTMP_CODEC_ID_H264) {
            if (media_info_v.sps_size > 0 && media_info_v.pps.size() > 0) {
                avc_sequence_header_.resize(4096);//.reset(new char[4096]);
                uint8_t *data = (uint8_t *)avc_sequence_header_.data();
                uint32_t index = 0;

                data[index++] = 0x17; // 1:keyframe  7:avc
                data[index++] = 0;    // 0: avc sequence header

                data[index++] = 0;
                data[index++] = 0;
                data[index++] = 0;

                // AVCDecoderConfigurationRecord
                data[index++] = 0x01; // configurationVersion
                data[index++] = media_info_v.sps.data()[1]; // AVCProfileIndication
                data[index++] = media_info_v.sps.data()[2]; // profile_compatibility
                data[index++] = media_info_v.sps.data()[3]; // AVCLevelIndication
                data[index++] = 0xff; // lengthSizeMinusOne

                // sps nums
                data[index++] = 0xE1; //&0x1f

                // sps data length
                data[index++] = media_info_v.sps_size >> 8;
                data[index++] = media_info_v.sps_size & 0xff;
                // sps data
                memcpy(data + index, media_info_v.sps.data(), media_info_v.sps_size);
                index += media_info_v.sps_size;

                // pps nums
                data[index++] = 0x01; //&0x1f
                // pps data length
                data[index++] = media_info_v.pps_size >> 8;
                data[index++] = media_info_v.pps_size & 0xff;
                // sps data
                memcpy(data + index, media_info_v.pps.data(), media_info_v.pps_size);
                index += media_info_v.pps_size;

                avc_sequence_header_size_ = index;
            }
            else {
                media_info_v.video_codec_id = 0;
            }
        }

        return 0;
    }
    
    int MediaRtmpProtocol::setMediaInfoA(const MediaInfo& media_info)
    {
        media_info_a = media_info;

        if (media_info_a.audio_codec_id == RTMP_CODEC_ID_AAC) {
                int index =0;
                aac_sequence_header_.resize(1024);// .reset(new char[aac_sequence_header_size_]);
                uint8_t *data = (uint8_t *)aac_sequence_header_.data();
                uint8_t sound_rate = 3; //for aac awlays 3
                uint8_t soundz_size = 1; //0:8bit , 1:16bit
                uint8_t sound_type = 1; //for aac awlays 1

                // audio tag data
                audio_tag_ = data[index++] = (((RTMP_CODEC_ID_AAC & 0xf) << 4) | ((sound_rate & 0x3) << 2) | ((soundz_size & 0x1) << 1) | (sound_type & 0x1));

                // aac packet type 
                data[index++] = 0; // 0: aac sequence header, 1: aac raw data
                int sampleRateIndex = GetAACSampleRateIndex(_audioSampleRate);
                data[index++] = 2<<3|sampleRateIndex>>1;
                data[index++] = (sampleRateIndex&1)<<7|(2<<3);
                data[index++] = 0x56;
                data[index++] = 0xe5;
                data[index++] = 0;
                // memcpy(data+2, media_info_a.audio_specific_config.data(), media_info_a.audio_specific_config_size);

                // 11 90 -- 48000 2, 12 10 -- 44100 2
                // uint32_t samplingFrequencyIndex = ((data[2] & 0x07) << 1) | ((data[3] & 0x80) >> 7);
                // uint32_t channel = (data[3] & 0x78) >> 3;
                // media_info_a.audio_channel = channel;
                // media_info_a.audio_samplerate = kSamplingFrequency[samplingFrequencyIndex];

                aac_sequence_header_size_ = index;
        } else {
            
// #define RTMP_CODEC_ID_AAC       10
// #define RTMP_CODEC_ID_G711A     7
// #define RTMP_CODEC_ID_G711U     8
// #define RTMP_CODEC_ID_ADPCM     1
            uint8_t sound_rate = 1; //5500
            uint8_t soundz_size = 1; //0:8bit , 1:16bit
            uint8_t sound_type = 0; //for aac awlays 1
            audio_tag_ = (((media_info_a.audio_codec_id & 0xf) << 4) | ((sound_rate & 0x3) << 2) | ((soundz_size & 0x1) << 1) | (sound_type & 0x1));
        }

        return 0;
    }
    int MediaRtmpProtocol::writeData(char* data, int dataLen)
    {
        logstream(_logWriteFlag, dataLen, _writeBytesAll, STREAM_LOG_RATE); 
        
        _wirtedData += dataLen;

        int iret = preWriteData((uint8_t*)data, dataLen);

        if (iret != 0)
        {
            return 0;
        }

        Rtp808HeadBaseV* headb = reinterpret_cast<Rtp808HeadBaseV*>(const_cast<char*>(data));

        RtmpMessage rtmp_msg;
        rtmp_msg.stream_id = stream_id_;

        if (headb->dataType == RTP_808_DATATYPE_A) {
            // logd("write audio frame");
            std::vector<RtmpBuffer> lstBuffer;
            int playloadLength = 0;
            uint64_t timestamp = 0;
            int offset = 0;
            while (offset < dataLen)
            {
                Rtp808HeadBaseA* head = reinterpret_cast<Rtp808HeadBaseA*>(const_cast<char*>(data+offset));
                timestamp = ReadUint64BE((char*)head->time);
                
                uint16_t dataLenPayLoad = swapBigEndian(head->dataLen);
                lstBuffer.push_back(RtmpBuffer(head->data, dataLenPayLoad));
                playloadLength += dataLenPayLoad;
                offset+=dataLenPayLoad+sizeof(Rtp808HeadBaseA);
                logd("loop audio dataLenPayLoad : " + std::to_string(dataLenPayLoad)+", playloadLength="+std::to_string(playloadLength)+",offset="+std::to_string(offset));
            }
	        uint8_t *frame_buf = (uint8_t*)lstBuffer[0].data;
            int frameSize = playloadLength;

            if (first_audio)
            {
                first_audio = false;
                MediaInfo media_info;
                switch (headb->headPT)
                {
                case PAYLOAD_TYPE_G711A:
                    media_info.audio_codec_id = RTMP_CODEC_ID_G711A;
                    break;
                case PAYLOAD_TYPE_G711U:
                    media_info.audio_codec_id = RTMP_CODEC_ID_G711U;
                    break;
                case PAYLOAD_TYPE_AAC:
                    media_info.audio_codec_id = RTMP_CODEC_ID_AAC;
                    break;
                case PAYLOAD_TYPE_ADPCM:
                    media_info.audio_codec_id = RTMP_CODEC_ID_ADPCM;
                    break;
                default:
                    loge("unknow codec");
                    return 0;
                }
                setMediaInfoA(media_info);

                if (media_info.audio_codec_id == RTMP_CODEC_ID_AAC)
                {
                    sendAudioData(0, RtmpBuffer(aac_sequence_header_.data(), aac_sequence_header_size_));
                }
            }

            logd("write audio : " + std::to_string(playloadLength));
            pushAudioFrame(lstBuffer, playloadLength);
        }
        else
        {
            std::vector<RtmpBuffer> lstBuffer;
            int playloadLength = 0;
            uint64_t timestamp = 0;
            int offset = 0;
            while (offset < dataLen)
            {
                Rtp808HeadBaseV* head = reinterpret_cast<Rtp808HeadBaseV*>(const_cast<char*>(data+offset));
                timestamp = ReadUint64BE((char*)head->time);
                

                uint16_t dataLenPayLoad = swapBigEndian(head->dataLen);
                lstBuffer.push_back(RtmpBuffer(head->data, dataLenPayLoad));
                playloadLength += dataLenPayLoad;
                offset+=dataLenPayLoad+sizeof(Rtp808HeadBaseV);
            }


	        uint8_t *frame_buf = (uint8_t*)lstBuffer[0].data;
            int frameSize = playloadLength;

			if (!has_sps_pps) {
				if (frame_buf[3] == 0x67 || frame_buf[4] == 0x67) {
					Nal sps = findNal(frame_buf, frameSize);
					if (sps.first != nullptr && sps.second != nullptr && *sps.first == 0x67) {
                        MediaInfo media_info;
						media_info.sps_size = (uint32_t)(sps.second - sps.first + 1);
						media_info.sps.resize(media_info.sps_size);// . reset(new uint8_t[media_info.sps_size]);
						memcpy(media_info.sps.data(), sps.first, media_info.sps_size);

						Nal pps = findNal(sps.second, frameSize - (int)(sps.second - frame_buf));
						if (pps.first != nullptr && pps.second != nullptr && *pps.first == 0x68) {
							media_info.pps_size = (uint32_t)(pps.second - pps.first + 1);
							media_info.pps.resize(media_info.pps_size);//reset(new uint8_t[media_info.pps_size]);
							memcpy(media_info.pps.data(), pps.first, media_info.pps_size);

							has_sps_pps = true;
                            logv("setMediaInfoV");
							setMediaInfoV(media_info); /* set sps pps */							
							// printf("Start rtmp pusher, rtmp url: %s , http-flv url: %s \n\n", RTMP_URL, HTTP_URL);
						}
					}
				}
			}
			
			if (has_sps_pps) {
				pushVideoFrame(lstBuffer, playloadLength); /* send h.264 frame */
			}
        }
        
            
        _packageNumber++;
        return 0;
    }
    

    int MediaRtmpProtocol::parseChunkBody()
    {
        uint8_t* buf = _readParam.first.data();
        uint32_t bytes_used = 0;
        uint32_t buf_size = _readParam.second;

        if (_chunk_stream_id < 0) {
            return -1;
        }

        auto& rtmp_msg = _rtmp_messages[_chunk_stream_id];
        uint32_t chunk_size = rtmp_msg.length - rtmp_msg.index;

        if (chunk_size > _in_chunk_size) {
            chunk_size = _in_chunk_size;
        }

        if (buf_size < (chunk_size + bytes_used)) {
            return -2;
        }

        if (rtmp_msg.index + chunk_size > rtmp_msg.length) {
            return -1;
        }

        memcpy(rtmp_msg.payloadRecv.data() + rtmp_msg.index, buf + bytes_used, chunk_size);
        bytes_used += chunk_size;
        rtmp_msg.index += chunk_size;

        if (rtmp_msg.index >= rtmp_msg.length || 
            rtmp_msg.index%_in_chunk_size == 0) {
                _chunkState = PARSE_HEADER;
        }

        removeBuffer(_readParam.first.data(), _readParam.second, bytes_used);
        return 0;
    }
    int MediaRtmpProtocol::parseChunkHead()
    {
        if (_readParam.second<1)
        {
            return -2;
        }
        int buf_size = _readParam.second;
	    uint32_t bytes_used = 0;
        uint8_t* data = _readParam.first.data();
        uint8_t flags = *data;
	    bytes_used += 1;
        // RtmpChunkHeader* pChunkHeader = reinterpret_cast<RtmpChunkHeader*>(_readParam.first.data());
        int len = 0;
        uint8_t csid = flags & 0x3f; // chunk stream id
        if (csid == 0) { // csid [64, 319] 
            if (buf_size < (bytes_used + 2)) {
                return -2;
            }
            csid += data[bytes_used] + 64;
            bytes_used += 1;
        }
        else if (csid == 1) { // csid [64, 65599]
            if (buf_size < (3 + bytes_used)) {
                return -2;
            }
            csid += data[bytes_used + 1] * 255 + data[bytes_used] + 64;
            bytes_used += 2;
        }

        uint8_t fmt = (flags >> 6); 
        if (fmt >= 4) {
            return -1;
        }

	    uint32_t header_len = kChunkMessageHeaderLen[fmt]; 
        if (buf_size < (header_len + bytes_used)) {
            return -2;
        }

        RtmpMessageHeader header={0};
        memcpy(&header, data + bytes_used, header_len);
        bytes_used += header_len;

        auto& rtmp_msg = _rtmp_messages[csid];
        _chunk_stream_id = rtmp_msg.csid = csid;

        if (fmt == RTMP_CHUNK_TYPE_0 || fmt == RTMP_CHUNK_TYPE_1) {
            uint32_t length = ReadUint24BE((char*)header.length);
            if (rtmp_msg.length != length || rtmp_msg.payloadRecv.size()<=0) {
                rtmp_msg.length = length;
                rtmp_msg.payloadRecv.resize(rtmp_msg.length);// .reset(new char[rtmp_msg.length]);
            }
            rtmp_msg.index = 0;
            rtmp_msg.type_id = header.type_id;
        }

        if (fmt == RTMP_CHUNK_TYPE_0) {
            rtmp_msg.stream_id = ReadUint24LE((char*)header.stream_id);
        }

        uint32_t timestamp = ReadUint24BE((char*)header.timestamp);
        uint32_t extend_timestamp = 0;
        if (timestamp >= 0xffffff || rtmp_msg.timestamp >= 0xffffff) {
            if (buf_size < (4 + bytes_used)) {
                return 0;
            }
            extend_timestamp = ReadUint32BE((char*)data + bytes_used);
            bytes_used += 4;
        }

        if (rtmp_msg.index == 0) { // first chunk
            if (fmt == RTMP_CHUNK_TYPE_0) {
                // absolute timestamp 
                rtmp_msg._timestamp = 0;
                rtmp_msg.timestamp = timestamp;
                rtmp_msg.extend_timestamp = extend_timestamp;
            }
            else {
                // relative timestamp (timestamp delta)
                if (rtmp_msg.timestamp >= 0xffffff) {
                    rtmp_msg.extend_timestamp += extend_timestamp;
                }
                else {
                    rtmp_msg.timestamp += timestamp;
                }
            }
        }
        
        removeBuffer(_readParam.first.data(), _readParam.second, bytes_used);
        return 0;
    }
    int MediaRtmpProtocol::getMessage(RtmpMessage& out_rtmp_msg)
    {
        auto& rtmp_msg = _rtmp_messages[_chunk_stream_id];

        if (rtmp_msg.index == rtmp_msg.length) {
            if (rtmp_msg.timestamp >= 0xffffff) {
                rtmp_msg._timestamp += rtmp_msg.extend_timestamp;
            }
            else {
                rtmp_msg._timestamp += rtmp_msg.timestamp;
            }

            out_rtmp_msg = rtmp_msg;
            _chunk_stream_id = -1;
            rtmp_msg.Clear();
            return 0;
        }
        return -2;
    }
	int MediaRtmpProtocol::createBasicHeader(uint8_t fmt, uint32_t csid, char* buf)
    {
        int len = 0;

        if (csid >= 64 + 255) {
            buf[len++] = (fmt << 6) | 1;
            buf[len++] = (csid - 64) & 0xFF;
            buf[len++] = ((csid - 64) >> 8) & 0xFF;
        }
        else if (csid >= 64) {
            buf[len++] = (fmt << 6) | 0;
            buf[len++] = (csid - 64) & 0xFF;
        }
        else {
            buf[len++] = (fmt << 6) | csid;
        }
        return len;
    }
	int MediaRtmpProtocol::createMessageHeader(uint8_t fmt, RtmpMessage& rtmp_msg, char* buf)
    {
        int len = 0;

        if (fmt <= 2) {
            if (rtmp_msg._timestamp < 0xffffff) {
                WriteUint24BE((char*)buf, (uint32_t)rtmp_msg._timestamp);
            }
            else {
                WriteUint24BE((char*)buf, 0xffffff);
            }
            len += 3;
        }

        if (fmt <= 1) {
            WriteUint24BE((char*)buf + len, rtmp_msg.length);
            len += 3;
            buf[len++] = rtmp_msg.type_id;
        }

        if (fmt == 0) {
            WriteUint32LE((char*)buf + len, rtmp_msg.stream_id);
            len += 4;
        }

        return len;
    }

	int MediaRtmpProtocol::createChunk(uint32_t csid, RtmpMessage& rtmp_msg, char* buf, uint32_t buf_size, std::vector<RtmpBuffer>& vecData)
    {
        uint32_t buf_offset = 0, payload_offset = 0;
        uint32_t vecIndex =0, vecOffset = 0;
        
        uint32_t capacity = (rtmp_msg.length / _out_chunk_size + 1) * 5 + 20;
        if (buf_size < capacity) {
            return -1;
        }

        buf_offset += createBasicHeader(0, csid, buf + buf_offset); //first chunk
        buf_offset += createMessageHeader(0, rtmp_msg, buf + buf_offset);
        if (rtmp_msg._timestamp >= 0xffffff) {
            WriteUint32BE((char*)buf + buf_offset, (uint32_t)rtmp_msg._timestamp);
            buf_offset += 4;
        }

        // push header
        vecData.push_back(RtmpBuffer(buf, buf_offset));
        auto lastBufOffset = buf_offset;
        while (rtmp_msg.length > 0)
        {
            int chunkSize = _out_chunk_size;
            // logd("create chunk ");
            while (chunkSize > 0)
            {
                int validSize = rtmp_msg.payloadVec[vecIndex].size - vecOffset;
                // logd("chunkSize="+std::to_string(chunkSize)+", vecIndex="+std::to_string(vecIndex)+", vecOffset="+std::to_string(vecOffset)+", validSize="+std::to_string(validSize)+", rtmp_msg.length="+std::to_string(rtmp_msg.length));
                if (validSize > chunkSize)
                {
                    vecData.push_back(RtmpBuffer(rtmp_msg.payloadVec[vecIndex].data + vecOffset, chunkSize));
                    vecOffset += chunkSize;
                    rtmp_msg.length -= chunkSize;
                    chunkSize -= chunkSize;
                }
                else
                {
                    vecData.push_back(RtmpBuffer(rtmp_msg.payloadVec[vecIndex].data + vecOffset, validSize));
                    chunkSize -= validSize;
                    
                    vecIndex++;
                    vecOffset = 0;
                    rtmp_msg.length -= validSize;
                }
                
                if (rtmp_msg.length==0)
                {
                    break;
                }
                if (rtmp_msg.length<0)
                {
                    loge("error length:"+std::to_string(rtmp_msg.length));
                    break;
                }
                if (vecIndex>=rtmp_msg.payloadVec.size())
                {
                    loge("error idnex:"+std::to_string(vecIndex));
                    break;
                }
                
                
            }
            if (rtmp_msg.length>0)
            {
                buf_offset += createBasicHeader(3, csid, buf + buf_offset);
                if (rtmp_msg._timestamp >= 0xffffff) {
                    WriteUint32BE(buf + buf_offset, (uint32_t)rtmp_msg._timestamp);
                    buf_offset += 4;
                }
                vecData.push_back(RtmpBuffer(buf+lastBufOffset, buf_offset-lastBufOffset));
                lastBufOffset = buf_offset;
            }
        }

        return buf_offset;
    }

    bool MediaRtmpProtocol::handleMessage(RtmpMessage& rtmp_msg)
    {
        bool ret = true;  
        switch(rtmp_msg.type_id)
        {        
            case RTMP_VIDEO:
                ret = handleVideo(rtmp_msg);
                break;
            case RTMP_AUDIO:
                ret = handleAudio(rtmp_msg);
                break;
            case RTMP_INVOKE:
                ret = handleInvoke(rtmp_msg);
                break;
            case RTMP_NOTIFY:
                ret = handleNotify(rtmp_msg);
                break;
            case RTMP_FLEX_MESSAGE:
                logd("unsupported rtmp flex message.");
                ret = false;
                break;
            case RTMP_SET_CHUNK_SIZE:           
                _in_chunk_size = (ReadUint32BE(rtmp_msg.payloadRecv.data()));
                break;
            case RTMP_BANDWIDTH_SIZE:
                break;
            case RTMP_FLASH_VIDEO:
                loge("unsupported rtmp flash video.");
                ret = false;
                break;    
            case RTMP_ACK:
                break;            
            case RTMP_ACK_SIZE:
                break;
            case RTMP_USER_EVENT:
                break;
            default:
                logd("unkonw message type : "+ std::to_string(rtmp_msg.type_id));
                break;
        }

        if (!ret) {
            //printf("rtmp_msg.type_id:%x\n", rtmp_msg.type_id);
        }
            
        return ret?0:-1;
    }
    bool MediaRtmpProtocol::handleInvoke(RtmpMessage& rtmp_msg)
    {
        bool ret  = true;
        amf_decoder_.reset();
    
        int bytes_used = amf_decoder_.decode((const char *)rtmp_msg.payloadRecv.data(), rtmp_msg.length, 1);
        if (bytes_used < 0) {
            return false;
        }

        std::string method = amf_decoder_.getString();
        //LOG_INFO("[Method] %s\n", method.c_str());
         {
            if(rtmp_msg.stream_id == 0) {
                bytes_used += amf_decoder_.decode(rtmp_msg.payloadRecv.data()+bytes_used, rtmp_msg.length-bytes_used);
                if(method == "connect") {            
                    ret = handleConnect();
                }
                else if(method == "createStream") {      
                    ret = handleCreateStream();
                }
            }
            else if(rtmp_msg.stream_id == stream_id_) {
                bytes_used += amf_decoder_.decode((const char *)rtmp_msg.payloadRecv.data()+bytes_used, rtmp_msg.length-bytes_used, 3);
                stream_name_ = amf_decoder_.getString();
                stream_path_ = "/" + app_ + "/" + stream_name_;
            
                if((int)rtmp_msg.length > bytes_used) {
                    bytes_used += amf_decoder_.decode((const char *)rtmp_msg.payloadRecv.data()+bytes_used, rtmp_msg.length-bytes_used);                      
                }
                
                if(method == "publish") {            
                    ret = handlePublish();
                }
                else if(method == "play") {          
                    ret = handlePlay();
                }
                else if(method == "play2") {         
                    ret = handlePlay2();
                }
                else if(method == "deleteStream") {
                    ret = handleDeleteStream();
                }
                else if (method == "releaseStream") {
                    loge("releaseStream Not implemented");
                }
            }
        }
        return ret;
    }
    bool MediaRtmpProtocol::handleNotify(RtmpMessage& rtmp_msg)
    {
        loge("Not implemented");
        return false;
    }
    bool MediaRtmpProtocol::handleVideo(RtmpMessage& rtmp_msg)
    {
        loge("Not implemented");
        return false;
    }
    bool MediaRtmpProtocol::handleAudio(RtmpMessage& rtmp_msg)
    {
        loge("Not implemented");
        return false;
    }
    void MediaRtmpProtocol::setPeerBandwidth()
    {
        std::vector<char> data(5);
        WriteUint32BE(data.data(), peer_bandwidth_);
        data.data()[4] = 2;
        RtmpMessage rtmp_msg;
        rtmp_msg.type_id = RTMP_BANDWIDTH_SIZE;
        rtmp_msg.payloadVec.push_back(RtmpBuffer(data.data(), data.size()));
        rtmp_msg.length = 5;
        sendRtmpChunks(RTMP_CHUNK_CONTROL_ID, rtmp_msg);
    }
    void MediaRtmpProtocol::sendAcknowledgement()
    {
        std::vector<char> data(4);
        WriteUint32BE(data.data(), acknowledgement_size_);

        RtmpMessage rtmp_msg;
        rtmp_msg.type_id = RTMP_ACK_SIZE;
        rtmp_msg.payloadVec.push_back(RtmpBuffer(data.data(), data.size()));
        rtmp_msg.length = 4;
        sendRtmpChunks(RTMP_CHUNK_CONTROL_ID, rtmp_msg);
    }
    void MediaRtmpProtocol::setChunkSize()
    {
        _out_chunk_size = (max_chunk_size_);
        std::vector<char> data(4);
        WriteUint32BE((char*)data.data(), max_chunk_size_);

        RtmpMessage rtmp_msg;
        rtmp_msg.type_id = RTMP_SET_CHUNK_SIZE;
        rtmp_msg.payloadVec.push_back(RtmpBuffer(data.data(), data.size()));
        rtmp_msg.length = 4;
        sendRtmpChunks(RTMP_CHUNK_CONTROL_ID, rtmp_msg);
    }
    
    bool MediaRtmpProtocol::sendInvokeMessage(uint32_t csid, const std::vector<RtmpBuffer>& payload, uint32_t payload_size)
    {
        RtmpMessage rtmp_msg;
        rtmp_msg.type_id = RTMP_INVOKE;
        rtmp_msg.timestamp = 0;
        rtmp_msg.stream_id = stream_id_;
        rtmp_msg.payloadVec = payload;
        rtmp_msg.length = payload_size; 
        sendRtmpChunks(csid, rtmp_msg);  
        return true;
    }
    bool MediaRtmpProtocol::sendNotifyMessage(uint32_t csid, const std::vector<RtmpBuffer>& payload, uint32_t payload_size)
    {
        RtmpMessage rtmp_msg;
        rtmp_msg.type_id = RTMP_NOTIFY;
        rtmp_msg.timestamp = 0;
        rtmp_msg.stream_id = stream_id_;
        rtmp_msg.payloadVec = payload;
        rtmp_msg.length = payload_size; 
        sendRtmpChunks(csid, rtmp_msg);  
        return true;
    }
    bool MediaRtmpProtocol::sendInvokeMessage(uint32_t csid, const RtmpBuffer& payload)
    {
        std::vector<RtmpBuffer> lst;
        lst.push_back(payload);
        return sendInvokeMessage(csid, lst, payload.size);
    }
    bool MediaRtmpProtocol::sendNotifyMessage(uint32_t csid, const RtmpBuffer& payload)
    {
        std::vector<RtmpBuffer> lst;
        lst.push_back(payload);
        return sendNotifyMessage(csid, lst, payload.size);
    }
    int MediaRtmpProtocol::sendRtmpChunks(uint32_t csid, RtmpMessage& rtmp_msg)
    {
        // logv("sendRtmpChunks");
        uint32_t capacity = rtmp_msg.length/ max_chunk_size_ *5 + 1024;
        std::shared_ptr<char> buffer(new char[capacity]);
        std::vector<RtmpBuffer> vecBuffer;
        int size = createChunk(csid, rtmp_msg, buffer.get(), capacity, vecBuffer);
        if (size > 0) {

            iovec* pdataList = new iovec[vecBuffer.size()];
            for (size_t i = 0; i < vecBuffer.size(); i++)
            {
                pdataList[i].iov_base = vecBuffer[i].data;
                pdataList[i].iov_len = vecBuffer[i].size;
            }
            int result = BaseProtocol::WriteV(pdataList, vecBuffer.size());
            if (result == -3)
            {
                // 不是回放就跳帧
                if (!_isPlayback)
                {
                    std::stringstream ss;
                    ss << _serviceName <<"> WriteV buffer is full, data len: " << (size) << ", buffer size: " << _writeBufferSize;
                    logd(ss.str());
                    ClearRtpBuffer(_listWriteBuffer, _writeBufferSize);
                }
            }
            else
            {
                UpdateActiveTime();
            }
            
            return result > 0 ? size : result;
        } else {
            return -1;
        }
    }

    bool MediaRtmpProtocol::handleConnect()
    {
        logd("");
        if(!amf_decoder_.hasObject("app")) {
            return false;
        }

        AmfObject amfObj = amf_decoder_.getObject("app");
        app_ = amfObj.amf_string;
        if(app_ == "") {
            return false;
        }

        sendAcknowledgement();
        setPeerBandwidth();   
        setChunkSize();

        AmfObjects objects;    
        amf_encoder_.reset();
        amf_encoder_.encodeString("_result", 7);
        amf_encoder_.encodeNumber(amf_decoder_.getNumber());

        objects["fmsVer"] = AmfObject(std::string("FMS/4,5,0,297"));
        objects["capabilities"] = AmfObject(255.0);
        objects["mode"] = AmfObject(1.0);
        amf_encoder_.encodeObjects(objects);
        objects.clear();
        objects["level"] = AmfObject(std::string("status"));
        objects["code"] = AmfObject(std::string("NetConnection.connect.Success"));
        objects["description"] = AmfObject(std::string("Connection succeeded."));
        objects["objectEncoding"] = AmfObject(0.0);
        amf_encoder_.encodeObjects(objects);  

        sendInvokeMessage(RTMP_CHUNK_INVOKE_ID, RtmpBuffer(amf_encoder_.data().data(), amf_encoder_.size()));
        return true;
    }
    bool MediaRtmpProtocol::handleCreateStream()
    {
        logd("");
        int stream_id = _default_stream_id;

        AmfObjects objects;
        amf_encoder_.reset();
        amf_encoder_.encodeString("_result", 7);
        amf_encoder_.encodeNumber(amf_decoder_.getNumber());
        amf_encoder_.encodeObjects(objects);
        amf_encoder_.encodeNumber(stream_id);

        sendInvokeMessage(RTMP_CHUNK_INVOKE_ID, RtmpBuffer(amf_encoder_.data().data(), amf_encoder_.size()));
        stream_id_ = stream_id;
        return true;
    }
    bool MediaRtmpProtocol::handlePublish()
    {
        loge("Not implemented");
        return false;
    }
    bool MediaRtmpProtocol::handlePlay()
    {
        logd("");
        //LOG_INFO("[Play] app: %s, stream name: %s, stream path: %s\n", app_.c_str(), stream_name_.c_str(), stream_path_.c_str());

        AmfObjects objects; 
        amf_encoder_.reset(); 
        amf_encoder_.encodeString("onStatus", 8);
        amf_encoder_.encodeNumber(0);
        amf_encoder_.encodeObjects(objects);
        objects["level"] = AmfObject(std::string("status"));
        objects["code"] = AmfObject(std::string("NetStream.Play.Reset"));
        objects["description"] = AmfObject(std::string("Resetting and playing stream."));
        amf_encoder_.encodeObjects(objects);   
        if(!sendInvokeMessage(RTMP_CHUNK_INVOKE_ID, RtmpBuffer(amf_encoder_.data().data(), amf_encoder_.size()))) {
            return false;
        }

        objects.clear(); 
        amf_encoder_.reset(); 
        amf_encoder_.encodeString("onStatus", 8);
        amf_encoder_.encodeNumber(0);    
        amf_encoder_.encodeObjects(objects);
        objects["level"] = AmfObject(std::string("status"));
        objects["code"] = AmfObject(std::string("NetStream.Play.Start"));
        objects["description"] = AmfObject(std::string("Started playing."));   
        amf_encoder_.encodeObjects(objects);
        if(!sendInvokeMessage(RTMP_CHUNK_INVOKE_ID, RtmpBuffer(amf_encoder_.data().data(), amf_encoder_.size()))) {
            return false;
        }

        amf_encoder_.reset(); 
        amf_encoder_.encodeString("|RtmpSampleAccess", 17);
        amf_encoder_.encodeBoolean(true);
        amf_encoder_.encodeBoolean(true);
        if(!this->sendNotifyMessage(RTMP_CHUNK_DATA_ID, RtmpBuffer(amf_encoder_.data().data(), amf_encoder_.size()))) {
            return false;
        }
                
        _connectionState = START_PLAY; 

        _url = stream_path_;
        OnReadyWebSocket();

        // onMessage("");
        return true;
    }
    bool MediaRtmpProtocol::handlePlay2()
    {
        loge("Not implemented");
        return false;
    }
    bool MediaRtmpProtocol::handleDeleteStream()
    {
        loge("Not implemented");
        return false;
    }
	bool MediaRtmpProtocol::handleResult(RtmpMessage& rtmp_msg)
    {
        loge("Not implemented");
        return false;
    }
	bool MediaRtmpProtocol::handleOnStatus(RtmpMessage& rtmp_msg)
    {
        loge("Not implemented");
        return false;
    }

    int MediaRtmpProtocol::handshake()
    {
        int result = 0;
        do
        {
            if (_handshakeState == HANDSHAKE_WAIT_C0C1)
            {
                // 接收C0+C1
                if ((result = recvHandShakeData())  != 0)
                {
                    break;
                }
                _handshakeState = HANDSHAKE_WAIT_S0S1S2;
            }
            if (_handshakeState == HANDSHAKE_WAIT_S0S1S2)
            {
                // 发送S0+S1+S2
                if ((result = sendHandShakeData())  != 0)
                {
                    break;
                }
                _handshakeState = HANDSHAKE_WAIT_C2;
            }
            if (_handshakeState == HANDSHAKE_WAIT_C2)
            {
                // 接收C2
                if ((result = recvHandShakeData2()) != 0)
                {
                    break;
                }
            }

        } while (false);
        

        return result;
    }
    int MediaRtmpProtocol::sendHandShakeData()
    {
        _rtmpDataS1.resize(1536);
        char version = _rtmpVersion;
        iovec data[3] = {{&version, 1}, {_rtmpDataS1.data(), _rtmpDataS1.size()}, {_rtmpDataC1.data(), _rtmpDataC1.size()}};
        int result = WriteV(data, sizeof(data) / sizeof(iovec));
        if(result > 0)
        {
            return 0;
        }
        return result;
    }
    int MediaRtmpProtocol::recvHandShakeData2()
    {
        if (_readParam.second<1536)
        {
            return -2;
        }
        
        std::vector<uint8_t> dataC2(1536);
        memcpy(dataC2.data(), _readParam.first.data(), 1536);


        removeBuffer(_readParam.first.data(), _readParam.second, 1536);
        
        return 0;
    }
    int MediaRtmpProtocol::recvHandShakeData()
    {
        if (_readParam.second<1537)
        {
            return -2;
        }
        
        uint8_t version = *_readParam.first.data();
        _rtmpVersionClient = version;

        _rtmpDataC1.resize(1536);
        memcpy(_rtmpDataC1.data(), _readParam.first.data()+1, 1536);

        _readParam.second = 0;
        return 0;
    }

}