#include<sys/types.h>
#include<sys/socket.h>
#include<arpa/inet.h>
#include<unistd.h>
#include<fcntl.h>
#include<sys/epoll.h>
#include<poll.h>
#include<sys/timerfd.h>
#include<map>

#include "Socket.h"
#include "debug.h"
#include "MessageHandle.h"
#include "EpollThread.h"

#define TIMER_INTERVAL 5
#define TIMER_EXPIRE 60

////////////////////////////////////////////////////////////////////////////////
//HandlerThread
EpollHandler::HandlerThread::HandlerThread(EpollHandler* pHandler, CSema* pSema) 
: CThread(3000)
{
    m_pHandler = pHandler;
    m_pSema = pSema;
}

void EpollHandler::HandlerThread::Main()
{
    while(!m_bExit){
        RECVBUFF* pRB = m_pHandler->GetBuff();
        if(pRB){
            // send(pRB->socket, pRB->buffer, pRB->buffsize, 0);
            TNETMESSAGE msgRequest; 
            if( !m_bExit && ReadNetMessage( (unsigned char*)pRB->buffer, pRB->buffsize, &msgRequest )){
                TNETMESSAGE msgResponse;

                struct sockaddr_in peerAddr;
                UINT peerLen = sizeof(peerAddr);
                if( getpeername(pRB->socket, (struct sockaddr *)&peerAddr, &peerLen)<0 ){
                    DebugPrint("getpeername error: %d", errno);
                }

                USHORT peerPort = ntohs(peerAddr.sin_port);                
                if(msgRequest.iExtra < 0x0D00){
                    memcpy(msgRequest.idDevice, &peerAddr.sin_addr, 4);
                    memcpy(msgRequest.idDevice+4, &peerAddr.sin_port, 2);                    
                }
                else{
                    memcpy(msgResponse.idDevice, msgRequest.idDevice, 6);
                }
                // if( msgRequest.eMessageType==e_versionRequest ){
                //     DebugPrint( "CONNECT  %s:%d  type: TCP", inet_ntoa(peerAddr.sin_addr), peerPort );
                // }

                if( !m_bExit && HandleRequest( &msgRequest, &msgResponse, peerAddr.sin_addr.s_addr, peerPort ) ){
                    short nSize = DATA_BUFSIZE;
                    if( !m_bExit && WriteNetMessage( (unsigned char*)pRB->buffer, nSize, &msgResponse ) ){
                        if( send(pRB->socket, pRB->buffer, nSize, 0) == SOCKET_ERROR){
                            DebugPrint("send() failed with error %d", errno);			   			
                        }	
                    }
                }
            }

            delete pRB;
        }
        else{
            m_pSema->ActP();
        }
    }   

    DebugPrint("EpollHandlerThread exit"); 
}
///////////////////////////////////////////////////////////////////////////////
//EpollHandler
EpollHandler::EpollHandler()
: m_smProcess(NULL, 0), m_smBuffers(NULL, 1)
{
    for(int i=0; i<MAX_THREAD_NUM; i++){
        m_lstThreads[i] = new HandlerThread(this, &m_smProcess);
        m_lstThreads[i]->Resume();
    }
}

EpollHandler::~EpollHandler()
{
    for(int i=0; i<MAX_THREAD_NUM; i++){
        delete m_lstThreads[i];
        m_lstThreads[i] = NULL;
    }

    CMutexLock lockProcess( &m_smBuffers ); 
    while(m_lstBuff.size()){
        delete m_lstBuff.back();
        m_lstBuff.pop_back();
    }
}

void EpollHandler::AddBuff(RECVBUFF* pRB)
{
    CMutexLock lockProcess( &m_smBuffers );
    m_lstBuff.push_back(pRB);

    m_smProcess.ActV();
}

RECVBUFF* EpollHandler::GetBuff()
{
    CMutexLock lockProcess( &m_smBuffers ); 

    RECVBUFF* pRB = NULL;
    if(!m_lstBuff.empty()){
        pRB = m_lstBuff.front();
        m_lstBuff.pop_front();
    }
    return pRB;
}

void EpollHandler::Terminate()
{
    //Set Thread Terminated status
    for(int i=0; i<MAX_THREAD_NUM; i++){
        m_lstThreads[i]->Terminate();
    }

    //Release sema individully
    for(int i=0; i<MAX_THREAD_NUM; i++){
        m_smProcess.ActV();  
        usleep(10000);
    }
}
//////////////////////////////////////////////////////////////////////////////
//EpollThread
EpollThread::EpollThread()
: CThread(5000)
{
    m_pHandler = new EpollHandler();
    if(pipe(m_fdPipe)<0){
        DebugPrint("create pipe error: %d", errno);
        memset(m_fdPipe,0,sizeof(m_fdPipe));
    }
}

EpollThread::~EpollThread()
{
    if(!IsTerminated()){
        Terminate();
    }

    if(m_pHandler){
        delete m_pHandler;
        m_pHandler = NULL;
    }

    if(m_fdPipe[0]){
        close(m_fdPipe[0]);
        close(m_fdPipe[1]);
    }
}

bool EpollThread::SetNoBlock(int fd)
{
    int oldSocketFlag = fcntl(fd, F_GETFL, 0);
    int newSocketFlag = oldSocketFlag | O_NONBLOCK;
    if (fcntl(fd, F_SETFL, newSocketFlag) == -1){
        close(fd);
        return false;
    }   

    return true;
}

void EpollThread::Terminate()
{
    CThread::Terminate();

    if(m_fdPipe[1])
        write(m_fdPipe[1], "0", 1);
    
    if(m_pHandler)
        m_pHandler->Terminate();
}

void EpollThread::Main()
{
    //创建一个监听socket
    int listenfd = socket(AF_INET, SOCK_STREAM | SOCK_NONBLOCK | SOCK_CLOEXEC, 0);
    if (listenfd == -1){
        DebugPrint("create listen socket error");
        return;
    }

    //设置重用ip地址和端口号
    int on = 1;
    setsockopt(listenfd, SOL_SOCKET, SO_REUSEADDR, (char*)& on, sizeof(on));
    setsockopt(listenfd, SOL_SOCKET, SO_REUSEPORT, (char*)& on, sizeof(on));

    //初始化服务器地址
    struct sockaddr_in bindaddr;
    bindaddr.sin_family = AF_INET;
    bindaddr.sin_addr.s_addr = htonl(INADDR_ANY);
    bindaddr.sin_port = htons(LISTEN_PORT);

    if (bind(listenfd, (struct sockaddr*) & bindaddr, sizeof(bindaddr)) == -1)
    {
        close(listenfd);
        DebugPrint("bind listen socker error.");
        return;
    }

    //启动监听
    if (listen(listenfd, SOMAXCONN) == -1){
        close(listenfd);
        DebugPrint("listen error.");
        return;
    }

    //创建epollfd
    int epollfd = epoll_create1(EPOLL_CLOEXEC);
    if (epollfd == -1){
        DebugPrint("create epollfd error.");
        close(listenfd);
        return;
    }

    epoll_event listen_fd_event;
    listen_fd_event.data.fd = listenfd;
    listen_fd_event.events = EPOLLIN;

    //将监听sokcet绑定到epollfd上去
    if (epoll_ctl(epollfd, EPOLL_CTL_ADD, listenfd, &listen_fd_event) == -1){
        DebugPrint("epoll_ctl error");
        close(listenfd);
        return;
    }

    if(m_fdPipe[0]){
        struct epoll_event ev;
        ev.data.fd = m_fdPipe[0];
        ev.events = EPOLLIN; //水平触发方式，监听可读事件，即管道有数据可以读取了
        if(epoll_ctl(epollfd, EPOLL_CTL_ADD, m_fdPipe[0], &ev)<0){
            DebugPrint("epoll_ctl_add pipe error: %d", errno);    
        }
    }

    int timerfd = timerfd_create(CLOCK_MONOTONIC, TFD_NONBLOCK);
    if(timerfd!=-1 ){
        struct itimerspec newValue = {};
        newValue.it_interval.tv_sec = TIMER_INTERVAL;
        newValue.it_value.tv_sec = TIMER_INTERVAL;
        if(timerfd_settime(timerfd, 0, &newValue, NULL)==-1){
            DebugPrint("settime error");
            close(timerfd);
        }
        else{
            epoll_event timer_event;
            timer_event.data.fd = timerfd;
            timer_event.events = EPOLLIN;

            if(epoll_ctl(epollfd, EPOLL_CTL_ADD, timerfd, &timer_event)<0){
                DebugPrint("epoll_ctl_add timer error: %d", errno);   
                close(timerfd); 
            }
        }
        
    }

    map<int, int> mapFds; //tcp连接组
    int maxCount = 0;
    epoll_event epoll_events[1024];
    char dataBuff[DATA_BUFSIZE];
    while(!m_bExit){
        int n = epoll_wait(epollfd, epoll_events, 1024, -1);
        if(n < 0){
            //被信号中断
            if (errno == EINTR)
                continue;

            //出错,退出
            DebugPrint("epoll exit with error: %d", errno);
            break;
        }

        for (int i = 0; i < n && !m_bExit; ++i) {
            if (epoll_events[i].data.fd == m_fdPipe[0]){ //exit signal
                DebugPrint("Got epoll exit signal");
                break;
            }
            else if (epoll_events[i].data.fd == timerfd){
                // DebugPrint("epoll timer on");
                uint64_t exp=0;
                if(read(timerfd, &exp, sizeof(uint64_t))!=sizeof(uint64_t)){
                    close(timerfd);
                }

                int count = 0;
                for(auto it=mapFds.begin(); it!=mapFds.end();){
                    it->second += TIMER_INTERVAL;

                    if( it->second >= TIMER_EXPIRE ){//空闲1分钟关闭
                        close(it->first);
                        it = mapFds.erase(it);
                        // DebugPrint("close expired socket");
                    }
                    else{
                        count++;
                        it++;
                    }
                }

                if(count>maxCount){
                    DebugPrint("mapFds max size: %d", count);
                    maxCount = count;
                }
            }
            else if (epoll_events[i].data.fd == listenfd){
                if(epoll_events[i].events & EPOLLIN){//侦听socket,接受新连接
                    struct sockaddr_in clientaddr;
                    socklen_t clientaddrlen = sizeof(clientaddr);
                    int clientfd = accept4(listenfd, (struct sockaddr*)&clientaddr, &clientaddrlen, SOCK_NONBLOCK|SOCK_CLOEXEC);
                    if (clientfd != -1){
                        epoll_event client_event;
                        client_event.data.fd = clientfd;
                        client_event.events = EPOLLIN | EPOLLRDHUP;// | EPOLLET;
                        if (epoll_ctl(epollfd, EPOLL_CTL_ADD, clientfd, &client_event) == -1){
                            DebugPrint("add client fd to epollfd error");
                            close(clientfd);
                        }
                        // DebugPrint("add client fd %d to epollfd", clientfd);
                        // mapFds.insert(clientfd, 0);
                        mapFds[clientfd] = 0;
                    }
                }
            }  
            else if (epoll_events[i].events & EPOLLRDHUP ) {
                if (epoll_ctl(epollfd, EPOLL_CTL_DEL, epoll_events[i].data.fd, NULL) == -1){
                //     DebugPrint("EPOLL_CTL_DEL clientfd %d by event %d", epoll_events[i].data.fd, epoll_events[i].events);
                // }
                // else{
                    DebugPrint("EPOLL_CTL_DEL clientfd %d by event %d error", epoll_events[i].data.fd, epoll_events[i].events);
                }

                close(epoll_events[i].data.fd);

                mapFds.erase(mapFds.find(epoll_events[i].data.fd));
            }                   
            else if (epoll_events[i].events & EPOLLIN){  //事件可读
                // DebugPrint("client fd: %d recv data.", epoll_events[i].data.fd);

                int m = recv(epoll_events[i].data.fd, dataBuff, DATA_BUFSIZE, 0);

                // DebugPrint("read data size %d", m);

                if( m>0 && ValidNetMessage((UCHAR*)dataBuff, m) ){
                    RECVBUFF* pRB = new RECVBUFF;
                    memcpy(pRB->buffer, dataBuff, m);
                    pRB->socket = epoll_events[i].data.fd;
                    pRB->buffsize = m;

                    m_pHandler->AddBuff(pRB);
                }
                else if (m <= 0 && errno != EWOULDBLOCK && errno != EINTR){ //对端关闭了连接，从epollfd上移除clientfd
                    if (epoll_ctl(epollfd, EPOLL_CTL_DEL, epoll_events[i].data.fd, NULL) == -1){
                    //     DebugPrint("EPOLL_CTL_DEL clientfd %d normally", epoll_events[i].data.fd);
                    // }
                    // else{
                        DebugPrint("EPOLL_CTL_DEL clientfd %d by EPOLLIN error", epoll_events[i].data.fd);
                    }

                    close(epoll_events[i].data.fd);
                    mapFds.erase(mapFds.find(epoll_events[i].data.fd));
                }

                mapFds.find(epoll_events[i].data.fd)->second = 0;
            }
        }
    }

    close(epollfd);

    DebugPrint("EPOLL main exit");
}