//
//  ZWRedisTool.cpp
//
//  Created by 杨净 on 2018/7/18.
//  Copyright © 2018年 杨净. All rights reserved.
//
//  email: 1026397500@qq.com
//

#include <hiredis/adapters/libuv.h>
#include "ZWRedisTool.h"
//#include "../MediaForwardService/src/WebSocket.h"

namespace mfs
{
    static std::shared_ptr<ZWRedisTool> _sharedInstance = std::make_shared<ZWRedisTool>();

    std::shared_ptr<ZWRedisTool> ZWRedisTool::RedisTool()
    {
        return _sharedInstance;
    }

    uint32_t ZWRedisTool::TaskData::NOW_ID = 1;


    int ZWRedisTool::connectRedis()
    {
        LOG_DEBUG <<"connectRedis,par1="<<redisParam.first.c_str()<<",par2="<<redisParam.second <<LOG_ENDL;
        FORWORT_ASSERT_RETURN_LOG_MSG(_loop, "uv loop uninitialized");
        
        auto strip = redisParam.first.c_str();
        auto port = redisParam.second;
        _ac = redisAsyncConnect(strip, port);
        if(_ac&&_ac -> err)
        {
            cleanTask();
            LOG_ERROR << _ac -> errstr << LOG_ENDL;
            return -1;
        }
        _initTime = std::chrono::system_clock::now();
        _invalid = 2;

        _ac -> data = this;
        if (_ac->c.timeout)
        {
            LOG_DEBUG << "connectRedis, "<< std::to_string(_ac->c.timeout->tv_sec) << "." << std::to_string(_ac->c.timeout->tv_usec) <<LOG_ENDL;
        }
        
        redisLibuvAttach(_ac, _loop);
        redisAsyncSetConnectCallback(_ac, connectCallback);
        redisAsyncSetDisconnectCallback(_ac, disconnectCallback);
        return 0;
    }

    void ZWRedisTool::commandCallback(redisAsyncContext* ac, void* r, void* data)
    {
        ZWRedisTool* redisTool = static_cast<ZWRedisTool*>(ac->data);

        redisReply* reply = reinterpret_cast<redisReply*>(r);
        auto taskid = (long)data;
        auto iter = redisTool->taskList.find(taskid);
        if(iter!=redisTool->taskList.end())
        {
            auto task = iter->second;
            if(task -> _type != TaskType_Client)
            {
                if((!reply || reply -> type == REDIS_REPLY_ERROR || !reply -> str ||
                    (reply -> type == REDIS_REPLY_STRING && strcmp(reply -> str, "OK") != 0)))
                {
                    redisAsyncDisconnect(ac);
                }
                else if(task -> _type == TaskType_Select)
                {
                    LOG_DEBUG <<"commandCallback Select"<<LOG_ENDL;
                    redisTool -> _invalid = 0;  // 数据库切换成功
                }
                else if (task -> _type == TaskType_Verify)
                {
                    LOG_DEBUG <<"commandCallback Verify"<<LOG_ENDL;
                }
            }
            else if(task -> _type == TaskType_Client)
            {
                if(std::shared_ptr<ITaskClient> pclient = task -> _client.lock())
                {
                    pclient->RedisCommandCallback(task->_key, reply);
                }
            }
            redisTool->taskList.erase(iter);
        }
    }

    void ZWRedisTool::cleanTask()
    {
        for(auto& task: taskList)
        {

            if(std::shared_ptr<ITaskClient> pclient = task.second -> _client.lock())
            {
                pclient->RedisCommandCallback("", nullptr);
            }
        }
        taskList.clear();
        // memset(&_epollEvent, 0, sizeof(epoll_event));
        // _epollInit = 0;
    }


    int ZWRedisTool::getValueAsync(const std::string& key, time_t currentTime, std::weak_ptr<ITaskClient> client)
    {
        if(_invalid == 0)
        {
            LOG_DEBUG <<"getValueAsync, _invalid="<<_invalid<<LOG_ENDL;
            auto ptask = std::make_shared<TaskData>(key, TaskType_Client, client);
            taskList.insert(make_pair(ptask->id(), ptask));
#define DO_REDIS_COMMAND(command, context, retCode) \
            do \
            { \
                if(command != REDIS_OK) \
                { \
                    redisAsyncDisconnect(context); \
                    retCode; \
                } \
            } while(0)

            DO_REDIS_COMMAND(redisAsyncCommand(_ac, commandCallback, reinterpret_cast<void*>(ptask->id()), "HGETALL %s", key.c_str()), _ac, return -1);
            return 0;
        }
        else
        {
            LOG_ERROR <<"getValueAsync, _invalid="<<_invalid<<LOG_ENDL;
        }
        return -1;
    }

    void ZWRedisTool::connectCallback(const redisAsyncContext* ac, int status)
    {
        // LOG_DEBUG <<"connect Callback,err= "<< ac->err << ",err str="<<ac->errstr<<",status="<<status <<LOG_ENDL;
        ZWRedisTool* redisTool = reinterpret_cast<ZWRedisTool*>(ac -> data);
        if (status!=0)
        {
            LOG_ERROR << "connect redis error:"<< ac->errstr<<",status="<<status << LOG_ENDL; 
            redisTool->_invalid = 1;
            return;
        }
        
        // 检测密码
        auto ptask = std::make_shared<TaskData>("", TaskType_Verify);
        redisTool ->taskList.insert(make_pair(ptask->id(), ptask));
        DO_REDIS_COMMAND(redisAsyncCommand(redisTool -> _ac, commandCallback, reinterpret_cast<void*>(ptask->id()), "AUTH ZwLBs_Redis@8F3!"), redisTool -> _ac, return);

        // 切换数据库 5
        ptask = std::make_shared<TaskData>("", TaskType_Select);
        redisTool ->taskList.insert(make_pair(ptask->id(), ptask));
        DO_REDIS_COMMAND(redisAsyncCommand(redisTool -> _ac, commandCallback, reinterpret_cast<void*>(ptask->id()), "SELECT 5"), redisTool -> _ac, return);
    }
    
        void ZWRedisTool::checkConnect()
        {
            // LOG_DEBUG << "1 checkConnect _ac="<<(void*)_ac<<",_invalid="<<_invalid<<LOG_ENDL;
            if (_invalid == 1)
            { 
                LOG_DEBUG << "Reconnection redis" << LOG_ENDL;
                connectRedis();
            }
            else if (_invalid == 2)
            {
                auto now = std::chrono::system_clock::now();
                auto milliseconds = std::chrono::duration_cast<std::chrono::milliseconds>(now - _initTime).count();
                if (milliseconds>5000)
                {
                    LOG_ERROR <<"Time out redis connection!"<<LOG_ENDL;
                    if (_ac)
                    {
                        redisAsyncContext* ac = _ac;
                        _ac = nullptr;
                        redisAsyncDisconnect(ac);
                        _invalid = 1;
                    }
                }
            }
            else if (_invalid == 0)
            {
                auto now = std::chrono::system_clock::now();
                for (auto iter = taskList.begin(); iter != taskList.end(); iter++)
                {
                    auto milliseconds = std::chrono::duration_cast<std::chrono::milliseconds>(now - iter->second->_requestTime).count();
                    if (milliseconds>5000)
                    {
                        LOG_ERROR <<"Time out redis task!"<<LOG_ENDL;
                        if (_ac)
                        {
                            redisAsyncContext* ac = _ac;
                            _ac = nullptr;
                            redisAsyncDisconnect(ac);
                            _invalid = 1;
                        }
                        break;
                    }
                }
            }
            // LOG_DEBUG << "2 checkConnect _ac="<<(void*)_ac<<LOG_ENDL;
        }
}
