#pragma once 
#include <string>
#include "hiredis.h"
#include "async.h" 
#include <functional>
#include <memory>
#include <iostream>

#include <thread>
#include <hiredis.h>
#include <async.h>
#include <adapters/libev.h>
#include "logger.hpp"
#include "sync_event_queue.hpp"
#include "event_queue.hpp"

enum ConnStatus{

    CONN_INIT,
    CONN_AUTH, 
    CONN_OPEN,
    CONN_CLOSE
}; 
enum RedisEvent{
    EVT_CONNECT,
    EVT_AUTH,
    EVT_DISCONNECTED,
    EVT_RESULT, 
}; 

typedef std::function<void(RedisEvent)> RedisEvtHandle; 

typedef std::function<void(redisReply * )> RedisHandle; 

struct RedisCommand{
    RedisCommand(int t):type(t)
    {
    }
    RedisCommand(const std::string &cmd  = "",RedisHandle h = nullptr,
	    redisReply *r= nullptr):command(cmd),handle(h),reply(r)
    { 
    }
    int type  = EVT_RESULT; 
    std::string command; 
    RedisHandle handle = nullptr; 
    redisReply * reply = nullptr; 
    void * data; 
    ~RedisCommand()
    {
	if (reply != nullptr)
	{
	    freeReplyObject(reply); 
	    reply = nullptr; 
	}
    }
}; 

typedef std::function<void(RedisCommand* )> ResultHandler; 

typedef EventQueue<RedisCommand*>  RedisResultQueue; 
class RedisConn
{
    public: 
	RedisConn(int idx = 0)
	{
	    m_index = idx; 
	    m_status = CONN_INIT; 
	}

	int init(const std::string & host  = "127.0.0.1",
		int port = 6379,const std::string & passwd  = "")
	{
	    m_host = host; 
	    m_port = port; 
	    m_passwd = passwd ; 
	    if (connect() != 0) {return -1;} 
	    m_thread = std::thread(&RedisConn::run,this); 
	    return 0; 
	}
	int connect()
	{
	    //DLog("connect to server %s:%d",m_host.c_str(),m_port); 
	    struct timeval timeout = { 1, 500000 }; // 1.5 seconds
	    m_ctx = redisConnectWithTimeout(m_host.c_str() , m_port, timeout);
	    if (m_ctx->err) {
		DLog("Error: %s", m_ctx->errstr);
		return -1; 
	    }

	    if (this->auth() != 0) {
		DLog("Error auth failed.");
		return -1;
	    } 

	    m_isRunning = true; 
	    return 0; 
	}
	
	void disconnect() {
	    if (m_ctx) {
		redisFree(m_ctx);
		m_ctx = nullptr;
	    }
	    m_isRunning = false;
	    m_inner_queue.notify();
	    if (m_thread.joinable()) {
		m_thread.join();
	    }
	}
	
	int auth()
	{
	    if (!m_passwd.empty())
	    {
		redisReply* reply = (redisReply*)redisCommand(m_ctx,  "AUTH %s",m_passwd.c_str() );
		if (!reply) { return -1; }
		m_status = CONN_OPEN; 
		freeReplyObject(reply);
		return 0;
	    }
	    else 
	    {
		m_status = CONN_OPEN;
		return 0; 
	    }
	}

	void execute(const std::string & query,RedisCommand * pResult )
	{
	    pResult->command = query; 
	    m_inner_queue.push(pResult); 
	}

	void run()
	{

		std::chrono::duration<double> sync_escaped_time;
		int count =0; 
	    
	    while (m_isRunning) {
		m_inner_queue.process([&](RedisCommand* pRst ){

			auto start = std::chrono::system_clock::now();
			redisReply *reply = (redisReply* ) redisCommand(m_ctx,pRst->command.c_str() );  
			auto end = std::chrono::system_clock::now();

			sync_escaped_time += (end - start);
			if (count++  % 10000 == 9999 )
			{
			    //std::cout << "Connection " << m_index << ": per 10000 escaped " << sync_escaped_time.count() << std::endl; 
			    count  = 0; 
			    sync_escaped_time = std::chrono::duration<double>(0); 
			}

			pRst->reply = reply ; 
			this->result_queue.push(pRst); 
			}); 
	    }

	    //DLog("############quit run ################"); 
	}

	bool is_connected()
	{
	    return m_status == CONN_OPEN || m_status == CONN_AUTH;  
	}

	RedisResultQueue  result_queue; 
    private:

	SyncEventQueue<RedisCommand* >  m_inner_queue; 

	int m_index; 
	std::thread m_thread; 
	std::string m_host; 
	int m_port; 
	std::string m_passwd; 
	redisContext * m_ctx ; 
	bool m_isRunning = false; 
	ConnStatus m_status; 

};