#include "AppSetting.h"
#include "BaseProtocol.h"
#include "MediaInfoService.h"

namespace mfs
{
    static uint64_t g_streamID = 0;
    BaseProtocol::BaseProtocol(std::shared_ptr<uvw::Loop> loop, std::shared_ptr<uvw::TCPHandle> client, const std::string& name):_loop(loop),_client(client),_serviceName(name)
    {
        _streamID = g_streamID++;
        _streamInfo.targetID = std::to_string(_streamID);
        _timeoutMilli = SOCKET_TIMEOUT_SECONDS*1000;
        _saveFilePathRecv = AppSetting::Instance()->streamDir + "/recv_"+ std::to_string(_streamID)+"_"+getTimeFileStr()+".log";
        _saveFilePathSend = AppSetting::Instance()->streamDir + "/send_"+ std::to_string(_streamID)+"_"+getTimeFileStr()+".log";
        createdirectories(AppSetting::Instance()->streamDir);
        UpdateActiveTime();
    }
    void BaseProtocol::SaveData(bool v)
    {
        if (_saveData != v)
        {
            _saveData = v;
            if (_saveData)
            {
                _fileRecv.open(_saveFilePathRecv, std::ios::app);
                if (!_fileRecv.is_open())
                {
                    loge("Failed to open data file:"+_saveFilePathRecv);
                }
                _fileSend.open(_saveFilePathSend, std::ios::app);
                if (!_fileSend.is_open())
                {
                    loge("Failed to open data file:"+_saveFilePathSend);
                }
                
            }
            else
            {
                _fileRecv.close();
                _fileSend.close();
            }
            
        }
        
    }
    void BaseProtocol::OnData(char* data, int dataLen)
    {
        if (_saveData && _fileRecv.is_open())
        {
            _fileRecv.write(data, dataLen);
            _fileRecv.flush();
        }
        
        if (auto ptrLoop = shareLoop())
        {
            _lastDataTime = ptrLoop->now();
        }
    }; 
    void BaseProtocol::OnAlloc(uv_buf_t *buf, int suggested)
    {
    }; 
    void BaseProtocol::OnClose()
    {
        logd(_serviceName + " client OnClose, write2client:"+std::to_string(_writeBytesAll)+", read4client:"+std::to_string(_readBytesAll));
        _streamInfo.state = "2";
        _streamInfo.sendBytes = std::to_string(_writeBytesAll);
        _streamInfo.recvBytes = std::to_string(_readBytesAll);
        _streamInfo.timestampEnd = std::to_string(std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::system_clock::now().time_since_epoch()).count());
        MediaInfoService::Instance()->sendFlowInfo(_streamInfo);
        if (_streamInfo.targetType!="0")
        {
            MediaInfoService::Instance()->sendUserEnd(_streamInfo);
        }
        else
        {
            MediaInfoService::Instance()->sendDeviceEnd(_streamInfo);
        }
    }; 
    void BaseProtocol::OnConnection()
    {
        logd(_serviceName + " client OnConnection.");
        _streamInfo.state = "0";
        _streamInfo.timestampStart = std::to_string(std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::system_clock::now().time_since_epoch()).count());
        MediaInfoService::Instance()->sendFlowInfo(_streamInfo);
    }; 
    void BaseProtocol::OnReady()
    {
        logd(_serviceName + " client OnReady.");
        _streamInfo.state = "1";
        MediaInfoService::Instance()->sendFlowInfo(_streamInfo);
    }
    void BaseProtocol::OnWrite(const uvw::WriteEvent &event)
    {
        // Write必须在loop所在线程才会触发onwrite
        // LOG_DEBUG << "client OnWrite:list="<<_listWriteBuffer.size() <<",buffer bytes="<< _writeBufferSize << LOG_ENDL;
        
        auto ptrBuffer =  _listWritedBuffer.front();
        _listWritedBuffer.pop_front();
        _writedBufferSize -= ptrBuffer.data.size();
        _writeBytesAll += ptrBuffer.data.size();
        if (_saveData && _fileSend.is_open())
        {
            _fileSend.write(ptrBuffer.data.data(), ptrBuffer.data.size());
            _fileSend.flush();
        }
        
        // LOG_DEBUG << "_writedBufferSize="<<_writedBufferSize <<";_writeBufferSize="<<_writeBufferSize <<LOG_ENDL;

        writeOneBuffer();
    }; 
    void BaseProtocol::OnError(const uvw::ErrorEvent &event)
    {
        if (ECONNRESET != event.code() &&  -ECONNRESET != event.code())
        {
            _streamInfo.message = "connect error:"+(std::to_string(event.code()))+" "+event.name()+ " " + event.what();
            std::stringstream ss;
            ss <<"client("<< _serviceName <<") error="<< event.code()<<",name="<<event.name()<<",waht="<<event.what();
            loge(ss.str());
        }
    }; 
    void BaseProtocol::OnEnd()
    {
        _streamInfo.message = "connect end!";
    }
    void BaseProtocol::PreAlloc(uv_buf_t *buf, int suggested)
    {
        if (!_hookfd)
        {
            OnAlloc(buf, suggested);
        }
        else
        {
            if (_onBeforeData)
            {
                // (_client.lock())->stop();
                _onBeforeData(this,buf);
                // (_client.lock())->read();
            }
        }
    }; 

    void BaseProtocol::writeOneBuffer()
    {
        if (_writeBufferSize>0)
        {
            auto ptrBuffer =  _listWriteBuffer.front();
            _listWriteBuffer.pop_front();

            _writeBufferSize -= ptrBuffer.data.size();
            _writedBufferSize += ptrBuffer.data.size();
            _listWritedBuffer.push_back(ptrBuffer);
            
            if (auto ptrClient = shareClient())
            {
                if (_cacheTestStream.is_open())
                {
                    _cacheTestStream.write(ptrBuffer.data.data(), ptrBuffer.data.size());
                    _cacheTestStream.flush();
                }
                ptrClient->write(ptrBuffer.data.data(), ptrBuffer.data.size());
            }
            else
            {
                loge("");
            }
            
        }
    }
    
    int BaseProtocol:: Write(char* data, int dataLen, bool forced)
    {
        if (!forced && _writeBufferSize + dataLen >= WRITE_BUFFER_MAX)
        {
            return -3;
        }
        BaseWriteBuffer buffer;
        buffer.data.resize(dataLen);
        memcpy(buffer.data.data(), data, dataLen);
        _listWriteBuffer.push_back(buffer);
        _writeBufferSize += dataLen;

        // LOG_DEBUG << "Write " << ((void*)this) << ", now buffer:" << _writeBufferSize << LOG_ENDL;
        if (auto ptrClient = shareClient())
        {
            if (ptrClient->writeQueueSize() <= 0)
            {
                writeOneBuffer();
            }
        }
        // if (_writedBufferSize <= WRITE_BUFFER_STEP)
        return dataLen;
    }
    int BaseProtocol:: Write2(char* data, int dataLen, bool forced)
    {
        // LOG_DEBUG << "Write " << ((void*)this) << ", now buffer:" << _writeBufferSize << LOG_ENDL;
        if (auto ptrClient = shareClient())
        {
            if (!forced && (ptrClient->writeQueueSize() > 0 || _listWritedBuffer.size() > 0))
            {
                return -3;
            }
            
            BaseWriteBuffer buffer;
            buffer.data.resize(dataLen);
            memcpy(buffer.data.data(), data, dataLen);
            _listWritedBuffer.push_back(buffer);
            _writedBufferSize += dataLen;
            
            ptrClient->write(buffer.data.data(), buffer.data.size());
        }
        return dataLen;
    }
    //TODO:待改造，若不拥塞直接写入socket缓存,缓存传递方式待优化，最好只在接收时拷贝一次
    int BaseProtocol::WriteV2(iovec* dataList, int size, bool forced)
    {
        int dataLen = 0;
        for(int i = 0; i != size; dataLen += dataList[i].iov_len, ++i);

        if (auto ptrClient = shareClient())
        {
            if (!forced && (ptrClient->writeQueueSize() > 0 || _listWritedBuffer.size() > 0))
            {
                return -3;
            }
            
            BaseWriteBuffer buffer;
            buffer.data.resize(dataLen);
            
            int offset = 0;
            for (int i = 0; i < size; i++)
            {
                memcpy(buffer.data.data()+offset, dataList[i].iov_base, dataList[i].iov_len);
                buffer.dataOffset = offset;
                offset+=dataList[i].iov_len;
            }

            _listWritedBuffer.push_back(buffer);
            _writedBufferSize += dataLen;
            
            ptrClient->write(buffer.data.data(), buffer.data.size());
        }
        
        return dataLen;
    }
    
    //TODO:待改造，若不拥塞直接写入socket缓存,缓存传递方式待优化，最好只在接收时拷贝一次
    int BaseProtocol::WriteV(iovec* dataList, int size, bool forced)
    {
        int dataLen = 0;
        for(int i = 0; i < size; dataLen += dataList[i].iov_len, ++i);

        if (!forced && _writeBufferSize + dataLen >= WRITE_BUFFER_MAX)
        {
            return -3;
        }
        
        // std::unique_ptr
        BaseWriteBuffer buffer;
        buffer.data.resize(dataLen);

        int offset = 0;
        for (int i = 0; i < size; i++)
        {
            memcpy(buffer.data.data()+offset, dataList[i].iov_base, dataList[i].iov_len);
            buffer.dataOffset = offset;
            offset+=dataList[i].iov_len;
        }

        _listWriteBuffer.push_back(buffer);
        _writeBufferSize += dataLen;

        // LOG_DEBUG << "Write " << ((void*)this) << ", now buffer:" << _writeBufferSize << LOG_ENDL;
        if (auto ptrClient = shareClient())
        {
            // if (_writedBufferSize <= WRITE_BUFFER_STEP)
            if (ptrClient->writeQueueSize() <= 0)
            {
                writeOneBuffer();
            }
            if (ptrClient->writeQueueSize()>0)
            {
                // LOG_DEBUG << "writeQueueSize="<<ptrClient->writeQueueSize() <<LOG_ENDL;
            }
        }
        
        return dataLen;
    }

    bool BaseProtocol::CheckTimeout(int timeoutMilli)
    {
        if (timeoutMilli<0)
        {
            timeoutMilli=_timeoutMilli;
        }
        if (timeoutMilli<0)
        {
            return false;
        }

        uvw::Loop::Time nowTime;
        if (auto ptrLoop = shareLoop())
        {
            nowTime = ptrLoop->now();
        }
        if ((nowTime - _lastActiveTime).count() > timeoutMilli)
        {
            _streamInfo.message = std::string("Timeout ") + (_lastDataUpdate?"self" : "other");
            std::stringstream ss;
            ss <<_serviceName <<">timeout last active="<<_lastActiveTime.count()<<";last data="<< _lastDataTime.count() <<";now="<<nowTime.count() <<", timeoutMilli="<<timeoutMilli<<", last connection1:"<<_lastDataUpdate<<", last connection1:"<<_lastDataUpdate2;
            loge(ss.str());
            return true;
        }
        return false;
    }
}