/*
 KX High Capacity High Performance Epoll Socket Client Daemon Class
 ---------------------------------------------------------------------------
 $Id: $ $Revision: $
 $HeadURL: $
 Copyright(c) 1990 - 2018 KRKIM(DURUMUL) Allrights reserved.
 */

/**
 * @file
 * @brief 
 * @detail 
 */

/*
 * History:
 * ===========================================================================
 * 2018.09.30  KRKIM   Initialized.
 *
 */

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <string.h>
#include <sys/epoll.h>
#include <arpa/inet.h>
#include <sys/socket.h>
#include <sys/types.h>
#include <netinet/in.h>
#include <netinet/tcp.h>
#include <fcntl.h>
#include <sys/ioctl.h>
#include <errno.h>
#include <signal.h>

#include <list>
#include <queue>
#include <vector>
#include <pthread.h>

#include "network.h"
#include "epollcomn.h"
#ifdef USE_INTERDBPOOL
#include "dbpool.h"
#endif

#include "kxHashMap.h"
#include "kxThread.h"
#include "iocq.h"
#include "tokenizer.h"

#include "kxClientSession.h"
#include "kxPollService.h"
#include "kxClientService.h"
#include "kxtracelog.h"

#define EPOLL_NEW
//클라이언트 소켓이므로 매번 ONESHOT 하지않는다.
//#define EPOLL_NOT_USE_ONESHOT_CLIENT

using namespace std;

CKxClientService::CKxClientService() : CKxService()
{
    m_pPollService = NULL;
    m_queuemode = QUEUE_MODE_WORKQUEUE | QUEUE_MODE_USE_CLIENT;
    m_bAutoReconnect = 1;
    m_ReconnectTimeval.tv_sec = 0;//1970.01.01 로부터 지나온 시간(초) TIMESTAMP 값이다.
    m_ReconnectTimeval.tv_usec = 0;
    m_nReconnectInterval = 30;//초단위: 로그가 너무많이 쌓여서 ...방지차원
    m_nTryConnectCount = 0;
}

CKxClientService::~CKxClientService()
{
    Destroy();
}

void CKxClientService::SetAutoReconnect(int bONOFF)
{
    m_bAutoReconnect = bONOFF;
}

void CKxClientService::SetMaxClient(int maxclient)
{
    m_nMaxClient = maxclient;
    int system_limit_fdcount = get_limit_fd();
    kxTraceLog(0, "SetMaxClient: Request MaxClient(%d),system limit max fd(%d)\n", maxclient, system_limit_fdcount);
    if (m_nMaxClient > system_limit_fdcount){
        m_nMaxClient = system_limit_fdcount;
        kxTraceLog(0, "SetMaxClient: over limit max fd(%d)\n", system_limit_fdcount);
    }
}

void CKxClientService::SetPollService(CKxPollService *pPollService)
{
    m_pPollService = pPollService;
}

bool CKxClientService::Init(int maxclient, int maxthreadpool, int epolltimeout,int conntimeout)
{
    SetMaxClient(maxclient);
    SetMaxThreadPool(maxthreadpool);
    m_epolltimeout = epolltimeout;
    m_conntimeout = conntimeout;
    kxTraceLog(0, "[KxSockClientService] Init:maxclient=%d maxthreadpool=%d epolltimeout=%d conntimeout=%d \n",
        maxclient,maxthreadpool, m_epolltimeout,m_conntimeout );
    if (m_nMaxThreadPool > 0){
        InitThreadPool(maxthreadpool);
    }
    return true;
}

bool CKxClientService::InitSockEvent()
{
    int epollsize = m_nMaxClient, poolsize = m_nMaxClient;
    m_eventfd = init_epoll(epollsize);
    if (m_eventfd < 0)
    {
        kxTraceLog(0, "InitSockClient:init_epoll error\n");
        return false;
    }

    // init pool
    m_pEpollEvents = new struct epoll_event[sizeof (*m_pEpollEvents) * poolsize];
    if (m_pEpollEvents == NULL)
    {
        kxTraceLog(0, "InitSockClient:memory alloc failed.\n");
        return false;
    }
    return true;
}

void CKxClientService::FreeSockEvent()
{
    close(m_eventfd);//모든 작업이 끝났다면 close()를 호출해서 닫아주어야 한다. 
    if (m_pEpollEvents)
        delete[]m_pEpollEvents;
    m_pEpollEvents = NULL;
}

bool CKxClientService::Connect(CKxClientSession *currentuser)
{
    int rc;
    int clientfd;
	/* set to reconnect to host */
    if (currentuser->GetSockFD() == INVALID_SOCKET){
        currentuser->InitSocket();
    }
    setnonblock(clientfd,false);

    /* connect to host */
    rc = currentuser->Connect();
    if(rc < 0){
        //윈도우랑 리눅스랑 #define ECONNREFUSED    111 /* Connection refused */ 이 다름.
        //linux     : EIDRM = 43 ECONNREFUSED = 111 EPIPE = 32
        //ms-windows: EIDRM = 111,ECONNREFUSED = 107 
        //http://www.virtsync.com/c-error-codes-include-errno
		kxTraceLog(0,"연결오류: host %s port %d rc = %d errno %d(%s)\n",
            currentuser->GetHost(),currentuser->GetPort(),rc,errno,strerror(errno));
        //파이프브로큰은 잘못된 소켓fd 접근
        if (errno == EPIPE || errno == ECONNREFUSED || errno == 111){//EIDRM
            return false;
        }
        //나머지는 논블로킹이라 기다리면 접속되는 상태를 의미.
	}

    clientfd = currentuser->GetSockFD();
	setnonblock(clientfd,true);
	settcpnodelay(clientfd,true);

#ifndef EPOLL_NEW
    del_epollfd(m_eventfd, clientfd);
#else
    Epoll_Del(m_eventfd, clientfd);
#endif
    /*no one shot */
#ifndef EPOLL_NEW
    rc = add_epollfd(m_eventfd, clientfd,EPEVT_READ|EPEVT_HANGUP|EPEVT_ERROR|EPEVT_ET);
#else
    //rc = Epoll_Add(m_eventfd, clientfd, currentuser, EPEVT_READ | EPEVT_HANGUP | EPEVT_ERROR | EPEVT_ET);
#ifdef EPOLL_NOT_USE_ONESHOT_CLIENT
    rc = Epoll_Add(m_eventfd, clientfd, currentuser, EPEVT_READ | EPEVT_WRITE | EPEVT_ERROR | EPEVT_ET);
#else
    rc = Epoll_Add(m_eventfd, clientfd, currentuser, EPEVT_READ | EPEVT_WRITE | EPEVT_ERROR | EPEVT_ET | EPEVT_ONESHOT);
#endif

#endif
    if (rc < 0)
    {
        kxTraceLog(0, "Connect:Epoll_Add sockfd(%d) error errno %d(%s)\n",clientfd,errno,strerror(errno));
        return false;
    }

    kxTraceLog(0, "[KxClientService] Connect:Host=%s port=%d sockfd(%d)\n", currentuser->GetHost(),currentuser->GetPort(),clientfd);
    return true;
}

int CKxClientService::TryConnectSocketList()
{
    int conn = 0;
    int doit = 0;
    int retry = 0;
    //소켓연결 타임아웃을 줘도 즉시(1초내) 빠져나온다(논블로킹모드)
    //그래서 epollhup 이나 processpending ewouldblock 걸림.
    //그래서 loop를 엄청돌기때문에 최근재연결 시도시간으로 갭을 주기로 한다.
    //2018.10.05

    struct timeval now, start;
    gettimeofday(&now, NULL);

    if (m_ReconnectTimeval.tv_sec == 0){//제일 처음은 무조건 재연결
        gettimeofday(&start, NULL);
        m_ReconnectTimeval = start;
        kxTraceLog(0, "소켓끊김 감지/자동 재연결 기능시작: timestamp = %ld,체크간격=%d sec\n",
            m_ReconnectTimeval.tv_sec,m_nReconnectInterval);
    }
    else{
        if (now.tv_sec <= m_ReconnectTimeval.tv_sec + m_nReconnectInterval){
            return 0;
        }
    }

    CKxClientSession *currentuser = NULL;
    list < CKxSockSession* >::iterator iter;
    m_ClientSockList.Lock();
    for (iter = m_ClientSockList.GetClientSockListPtr()->begin();
        iter != m_ClientSockList.GetClientSockListPtr()->end(); ++iter)
    {
        currentuser = (CKxClientSession* )*iter;
        if(currentuser->IsConnected() == false){
            if (m_nTryConnectCount > 0){
                kxTraceLog(0, "끊긴소켓 재연결 시도중: host %s port %d\n",
                    currentuser->GetHost(), currentuser->GetPort());
            }
            else if (m_nTryConnectCount == 0){
                kxTraceLog(0, "연결시도: host %s port %d\n",
                    currentuser->GetHost(), currentuser->GetPort());
            }
            retry++;
            if(Connect(currentuser)== true){
                conn++;
			}
		}
    }
    if (retry > 0){
        m_nTryConnectCount++;
        if (m_nTryConnectCount % 10 == 0){
            //10 * m_nReconnectInterval 초가 지났을 경우는,계속 시도 하지말고 체크시간 간격을 늘려준다.
            //이유,아예 꺼진 서버로 접속시도 무한반복시 로그만 계속 쌓이므로 10분이넘으면 1분추가,
            //즉 10분 시점부터는 2분간격으로 폴링,20분부터는 3분간격,30분 부터는 4분간격
            //여기서 10분은 m_nReconnectInterval의 값이 60초일 경우를 뜻하고 시간간격은 배수가 된다.
            //죽었던 서버가 살아나면(장애나 운영상 점검시) 자동으로 클라이언트 소켓데몬은 자동으로 연결된다.
            m_nReconnectInterval = m_nReconnectInterval * 2;
            if (m_nReconnectInterval > 1800)//0.5시간을 넘으면
                m_nReconnectInterval = 1800;//0.5시간마다 체크
        }
    }

    gettimeofday(&now, NULL);
    m_ReconnectTimeval = now;
    m_ClientSockList.UnLock();
    return conn;
}

bool CKxClientService::DestroySockClient(CKxClientSession *currentuser)
{
    //클라이언트목록에서는 제외하지 않는다.
    //재연결을 위해서.
    currentuser->CloseClient();
    return true;
}

//메인쓰레드 시작직후 호출
//리턴: 0 성공 0이 아니면 종료
int CKxClientService::OnServiceBegin()
{
    
    return CKxService::OnServiceBegin();
}

//메인쓰레드 종료전 호출
int CKxClientService::OnServiceEnd()
{
    
    return CKxService::OnServiceEnd();
}

//메인쓰레드 시작직후 호출
//리턴: 0 성공 0이 아니면 종료
int CKxClientService::OnThreadBegin(int nID)
{
    return CKxService::OnThreadBegin(nID);
}

//메인쓰레드 종료전 호출
int CKxClientService::OnThreadEnd(int nID)
{
    
    return CKxService::OnThreadEnd(nID);
}

int CKxClientService::OnServiceLoop(int threadid)
{
    int	i;
    int	clientfd;
    int rc,start;
    int	poolsize = m_nMaxClient;
    int	max_got_events;
    //int reconnected = 1;
    int connected = 0;
    CKxQueue *pQueue = NULL;

    start = OnServiceBegin();
    while (start == 0 && m_bServiceRunning)
    {
        if (m_bAutoReconnect || connected == 0){
            connected = TryConnectSocketList();
            connected = 1;
        }
        //재연결 모드시에는 타임아웃을 준다.
        max_got_events = epoll_wait(m_eventfd, m_pEpollEvents, poolsize, m_epolltimeout/*-1*/);
        for (i = 0; i < max_got_events; i++)
        {
            //소유쓰레드의 이벤트큐에 입력한다.
            //int clientfd = ev.data.fd;
            CKxClientSession *currentuser = (CKxClientSession *)m_pEpollEvents[i].data.ptr;//m_ClientSockList.Find(clientfd);
            clientfd = currentuser->GetSockFD();
            if (currentuser == NULL)
                continue;
                //kxTraceLog(0, "[%d] currentuser->m_threadno=%d\n",
                //    i,currentuser->m_threadno);
            if (currentuser->m_threadno == -1){
                //char *remoteaddr = inet_ntoa(clientaddr.sin_addr);

                //가장 적은 소켓을 처리하는 쓰레드를 찾는다.					
                int thread_index = 0;
                thread_index = m_pThreadPool->FindMinClients();//m_ClientSockList.GetCount() % m_nMaxThreadPool;
                //kxTraceLog(0, "new client sock(%d) thread_index=%d\n",
                //    clientfd, thread_index);

                currentuser->m_threadno = thread_index;
                //currentuser->SetSockFD(clientfd,&clientaddr);
                //currentuser->InitContext(MAX_IOBUFFSIZE);
                //m_ClientSockList.Add(currentuser);
                //kxTraceLog(0, "new client sock(%d) GetThread\n", clientfd);

                CKxThreadCtx * lThread = m_pThreadPool->GetThread(thread_index);
                lThread->Lock();
                lThread->m_clientnum++;
                lThread->UnLock();
                //kxTraceLog(0, "new client sock(%d) currentuser=%x Thread[%d] clients=%d\n",
                //    clientfd, currentuser, thread_index, lThread->m_clientnum);

//#ifdef EPOLL_NEW
//                rc = Epoll_Add(m_eventfd, clientfd, currentuser, EPEVT_READ | EPEVT_WRITE | EPEVT_ET | EPEVT_ONESHOT);
//                if (rc < 0)
//                {
//                    kxTraceLog(0, "ERROR:add client to epool,so close client sock now.\n");
//                    currentuser->OnDisconnected();
//                    currentuser->CloseClient();
//                    break;
//                }
//#endif
                currentuser->OnConnected();
                kxTraceLog(0, "Connect OK:client sock(%d) currentuser=%x dispatch to Thread[%d] clients=%d\n",
                    clientfd, currentuser, thread_index, lThread->m_clientnum);
            }

            CKxThreadCtx *lThread = m_pThreadPool->GetThread(currentuser->m_threadno);
            kxTraceLog(0, "Client Epoll:sock(%d) dispatch to Thread[%d] clients=%d\n",
                clientfd, lThread->m_ord_id - 1, lThread->m_clientnum);

            KX_QUEUE_ITEM qitem;
            qitem.mask = KXQUEUE_MASK_USE_EVENTS;
            qitem.events.epollevent = m_pEpollEvents[i];

            //kxTraceLog(0, " PushSignal sock(%d) currentuser=%x\n", clientfd, currentuser);
            if (((m_queuemode & QUEUE_MODE_USE_CLIENT) & QUEUE_MODE_MASK) == QUEUE_MODE_USE_CLIENT){
                pQueue = (CKxQueue *)lThread;
            }
            else if (((m_queuemode & QUEUE_MODE_USE_SERVER) & QUEUE_MODE_MASK) == QUEUE_MODE_USE_SERVER){
                pQueue = this->GetQueue();
            }
            pQueue->PushSignal(qitem);
        }
        usleep(1);//FOR BIT CPU IDLE

        //TODO::Send pending data
    }
    kxTraceLog(0, "ServiceRunning Stopped.\n");
    rc = OnServiceEnd();
    return true;
}

int CKxClientService::OnThreadLoop(int threadid)
{
    int rc;
    int index,start;
    int count, count1;

    CKxThreadCtx * lThread;
    CKxService *pService;
    KX_QUEUE_ITEM qitem; //struct epoll_event ev;
    CKxQueue *pQueue = NULL;

    if (m_pThreadPool)
        lThread = m_pThreadPool->GetThreadByID(threadid);

    start = OnThreadBegin(lThread->m_ord_id);

    pService = (CKxService *)lThread->m_pService;//same as this
    kxTraceLog(TL_LOW, "[Thread %d] pthread_self() = %x thr_id = %x (%x) started\n",
        lThread->m_ord_id, pthread_self(), lThread->m_thr_id,threadid);

    //lThread = (CKxThreadCtx *)arg;
    //lThread->WatchJob();
    /*pthread_cond_wait()는 조건변수 cond로 시그널이 전달되기를 기다린다.
      이 함수는 호출되면 자동적으로 mutex잠금을 되돌려준다(UnLock m_mut internally)
      내부적으로 pthread_unlock_mutex()를 호출한다.
      이 함수를 호출한 쓰레드는 조건변수에 시그널이 전달될 때까지 기다리며,
      이때 CPU는 소비하지 않는다.
      시그널을 전달받아서 쓰레드가 깨어나면 자동적으로 mutex잠금을 얻는다(
      pthread_lock_mutex()를 사용)
      */
    //TODO::Check This!!
    //조건변수 m_cond가 모든쓰레드와 공용으로 사용하게 되면 
    //쓰레드 하나가 모든 이벤트를 모두 다 처리하는게 아닌가?
    //그렇다면 쓰레드풀의 나머지 쓰레드는 놀고있고 오직 한 쓰레드만 깨어나 작업하는거 아닌가??
    //따라서,쓰레드별로 조건변수를 두고 뮤텍스만 공용처리하여,이벤트큐에서 꺼내갈때,
    //충돌나지 않도록만 하게 한다?
    //그러면 먼저 시작된 쓰레드가 수신데이타 처리시간이 더 길면,나중에 깨어난 쓰레드의
    //처리시간이 더 짧으면 클라이언트로 응답하는 데이타가 순서가 바뀔 가능성이 있는것 같다.
    //이렇게 되면 쓰레드간 동기화를 별도로 처리해줘야 할 것같다.
    //즉,소켓이 현재 처리하는 것과 같은 경우만 처리토록 하고 이벤트큐를 끝까지 단독처리 하지 않게?
    //http://www.joinc.co.kr/modules/moniwiki/wiki.php/Site/Thread 참고해서 재분석 및 고려할것

    while (start == 0 && m_bServiceRunning)
    {
        //Pop Queue Data
        //Subscribe mode or workqueue mode
        //fanout 방식(각 쓰레드별 자체 queue 처리-같은데이터)
        //  => 모든 쓰레드가 각각 자신의 큐에 들어온 같은 데이터를 꺼내와서 동일한 데이터를 분배처리
        //work queue 방식(각 쓰레드별 공용 queue사용-각기다른 데이터처리)
        //  => 모든 쓰레드가 각각 서로다른 큐를 꺼내와서 처리
        
        //Subscribe mode or workqueue mode
        //workqueue:모든 쓰레드중 1개가 각각 서로다른 소켓데이터의 큐를 꺼내와서 처리
        if (((m_queuemode & QUEUE_MODE_USE_CLIENT) & QUEUE_MODE_MASK) == QUEUE_MODE_USE_CLIENT){
            pQueue = (CKxQueue *)lThread;
        }
        else if (((m_queuemode & QUEUE_MODE_USE_SERVER) & QUEUE_MODE_MASK) == QUEUE_MODE_USE_SERVER){
            pQueue = pService->GetQueue();
        }
        kxTraceLog(TL_LOW, "[Thread %d] WaitCondSignal() threadid=%x pQueue=%x(mode=%x)\n",lThread->m_ord_id,threadid,pQueue,m_queuemode);
        pQueue->WaitCondSignal();
        while (1){
            pQueue->Lock();
            count = pQueue->GetQueueSize();
            if (pQueue->IsEmpty() || count < 1){
                //kxTraceLog(TL_LOW,"[CKxThreadCtx %d] queue count=%d suspended to wait signal.\n",m_ord_id,count);
                pQueue->UnLock();
                break;
            }
			
			qitem = pQueue->PopData(0);
			count1 = pQueue->GetQueueSize();
			pQueue->UnLock();
            //kxTraceLog(TL_LOW, "[Thread %d] thr_id = %x popup ev,pService=%x\n", lThread->m_ord_id, lThread->m_thr_id, pService);
            pService->OnThreadProc(lThread->m_ord_id,(void*)&qitem);
            pQueue->Free(&qitem);//2018.09.30
        }
    }
    kxTraceLog(TL_LOW, "[Thread %d] pthread_self() = %x thr_id = %x stop\n", lThread->m_ord_id, pthread_self(), lThread->m_thr_id);
    rc = OnThreadEnd(lThread->m_ord_id);
    return 0;
}

bool CKxClientService::StartService()
{
    int connected;
    InitSockEvent();

    return CKxService::StartService();
}

bool CKxClientService::StopService()
{
    SetServiceRun(false);
    FreeSockEvent();
    return CKxService::StopService();
}

//접속할 클라이언트 소켓,관리대상 클라이언트를 등록한다.
CKxClientSession * CKxClientService::AddClientSock(char *hostname,int port,int conntimeout)
{
    bool rc = false;
    CKxClientSession *currentuser = new CKxClientSession();
   
    currentuser->Init(hostname, port, conntimeout);
    if (currentuser != NULL)
    {
        currentuser->InitContext(MAX_IOBUFFSIZE);
        m_ClientSockList.Add(currentuser);
        return currentuser;
    }
    return NULL;
}

//이미 연결된 리슨서버의 accept된 소켓
//CKxClientSession * CKxClientService::AddClientSock(int sockfd, sockaddr_in *pClientAddr)
//{
//    bool rc = false;
//    CKxClientSession *currentuser = new CKxClientSession();
//    if (currentuser != NULL)
//    {
//        currentuser->SetSockFD(sockfd, pClientAddr);
//        currentuser->InitContext(MAX_IOBUFFSIZE);
//        m_ClientSockList.Add(currentuser);
//        return currentuser;
//    }
//    return NULL;
//}

bool CKxClientService::RemoveClientSock(int sockfd)
{
    return m_ClientSockList.Remove(sockfd);
}

bool CKxClientService::RemoveClientSock(CKxClientSession *pUser)
{
    return m_ClientSockList.Remove(pUser);
}

int CKxClientService::OnServicePolling(TIPARAM *polltime)
{
	return CKxService::OnServicePolling(polltime);
}

int CKxClientService::OnThreadQueue(int nID/*orderid*/,void *param/*popped queue data*/)
{
    return CKxService::OnThreadQueue(nID,param);
}

//http://vpj.posterous.com/tcp-echo-server-example-in-c-using-epoll
//int CKxClientService::OnThreadProc(struct epoll_event ev)
int CKxClientService::OnThreadProc(int nID/*orderid*/,void *param/*popped queue data*/)
{
    // read from the cfd
    int rc = 0;
    int rc2 = 0;
    KX_QUEUE_ITEM *qitem = (KX_QUEUE_ITEM*) param;
    struct epoll_event ev;
    ev = qitem->events.epollevent;
    int clientfd;
    CKxClientSession *currentuser = (CKxClientSession *)ev.data.ptr;//m_ClientSockList.Find(clientfd);
    CKxThreadCtx * lThread;
    CKxService *pService;

    /*lThread = m_pThreadPool->GetThreadByOrderID(nID);*/
    lThread = m_pThreadPool->GetThread(currentuser->m_threadno);
    pService = (CKxService *)lThread->m_pService;//same as this

    clientfd = currentuser->GetSockFD();
    //kxTraceLog(0, "[KxClientService] OnThreadProc:client sock(%d) currentuser=%x ev.data.fd=%x\n", clientfd, currentuser,ev.data.fd);

    if (currentuser == NULL)
    {
        kxTraceLog(0, "error,client sock_context is null\n");
        return -1000;
    }

    if (ev.events & EPOLLHUP){
        CloseClient(currentuser);

        kxTraceLog(0, "client(%d) EPOLLHUP detected.close socket.Thread[%d] clients= %d\n",
            clientfd, currentuser->m_threadno, lThread->m_clientnum);
        return -1002;
    }

    if (ev.events & EPOLLERR){
        kxTraceLog(0, "client(%d) EPOLLERR detected.so skip.\n", clientfd);
        return -1003;
    }

    if (ev.events & EPOLLIN){
        kxTraceLog(0, "client(%d) EPOLLIN status detected.available to recv incoming data now.\n", clientfd);
        currentuser->Lock();
        rc = currentuser->OnRecvPacket();
        currentuser->UnLock();
        if (rc == SOCK_TRY_CLOSE){
            CloseClient(currentuser);
            //int trid = 123;
            //int duringsec = 60;
            //char pkstr[1031];
            //system hangup
            //sprintf(pkstr, "HUP %d %d\n", cfd,trid,duringsec);
            //senddatatoall(pkstr);
            kxTraceLog(0, "client(%d) SOCK_TRY_CLOSE\n",clientfd);
            return -1004;
        }
        // re add to epoll on ET mode (in oneshot)
#ifndef EPOLL_NOT_USE_ONESHOT_CLIENT
#ifndef EPOLL_NEW
        rc2 = modify_epollfd(m_eventfd,clientfd,EPEVT_READ|EPEVT_WRITE|EPEVT_ET|EPEVT_ONESHOT);
#else
        rc2 = Epoll_Modify(m_eventfd, clientfd, currentuser, EPEVT_READ | EPEVT_WRITE | EPEVT_ET | EPEVT_ONESHOT);
#endif
#endif

        if (rc2 < 0)
        {
            kxTraceLog(0, "client(%d) ERROR:modify_epollfd to recv.\n", clientfd);
        }
    }
    if (ev.events & EPOLLOUT){
        kxTraceLog(0, "client(%d) EPOLLOUT status detected.available to send now.\n", clientfd);
        currentuser->Lock();
        rc = currentuser->OnSendPacket();
        currentuser->UnLock();
        if (rc == SOCK_TRY_CLOSE){
            CloseClient(currentuser);
            kxTraceLog(0, "client(%d) EPOLLOUT finish DATA,close socket.Thread[%d] clients= %d\n",
                clientfd, currentuser->m_threadno, lThread->m_clientnum);
        }
        else if (rc == SOCK_DATA_PENDING_SHIFT_SENDBUFFER){
            kxTraceLog(0, "client(%d) send buffer is remain,so modify_epollfd to send at nexttime.\n", clientfd);
#ifndef EPOLL_NOT_USE_ONESHOT_CLIENT
#ifndef EPOLL_NEW
            rc2 = modify_epollfd(m_eventfd,clientfd,EPEVT_READ|EPEVT_WRITE|EPEVT_ET|EPEVT_ONESHOT);
#else			
            rc2 = Epoll_Modify(m_eventfd, clientfd, currentuser, EPEVT_READ | EPEVT_WRITE | EPEVT_ET | EPEVT_ONESHOT);
#endif
#endif
            if (rc2 < 0)
            {
                kxTraceLog(0, "client(%d) ERROR:modify_epollfd to send.\n", clientfd);
            }
        }
        else{//to read event only
            kxTraceLog(0, "client(%d) send buffer is flush,so modify_epollfd to recv at nexttime.\n", clientfd);
#ifndef EPOLL_NOT_USE_ONESHOT_CLIENT
#ifndef EPOLL_NEW
            rc2 = modify_epollfd(m_eventfd,clientfd,EPEVT_READ|EPEVT_ET|EPEVT_ONESHOT);
#else
            rc2 = Epoll_Modify(m_eventfd, clientfd, currentuser, EPEVT_READ | EPEVT_ET | EPEVT_ONESHOT);
            //rc2 = Epoll_Modify(m_eventfd, clientfd, currentuser, EPEVT_READ | EPEVT_WRITE | EPEVT_ET | EPEVT_ONESHOT);
#endif
#endif
            if (rc2 < 0)
            {
                kxTraceLog(0, "client(%d) ERROR:modify_epollfd to recv after send.\n", clientfd);
            }
        }
    }
    return rc;
}

bool CKxClientService::senddatatoall(const char *chatmsg, int len)
{
    int sendflags = 0;
    CKxClientSession *currentuser = NULL;
    list < CKxSockSession* >::iterator iter;
    m_ClientSockList.Lock();
    for (iter = m_ClientSockList.GetClientSockListPtr()->begin();
        iter != m_ClientSockList.GetClientSockListPtr()->end(); ++iter)
    {
        currentuser = (CKxClientSession* )*iter;
        // sending only / no need to lock
        send((*iter)->GetSockFD(), chatmsg, len, sendflags);
    }
    m_ClientSockList.UnLock();
    return true;
}

int CKxClientService::PostSendToAll(char *pData, int nLen,int bNow)
{
    int sendflags = 0;
    CKxClientSession *currentuser = NULL;
    list < CKxSockSession* >::iterator iter;
    m_ClientSockList.Lock();
    for (iter = m_ClientSockList.GetClientSockListPtr()->begin();
        iter != m_ClientSockList.GetClientSockListPtr()->end(); ++iter)
    {
        currentuser = (CKxClientSession* )*iter;
        currentuser->PostSend(pData, nLen);
        if (bNow)
            currentuser->SendPacket();
        //kxTraceLog(0, "PostSendToAll\n");
    }
    m_ClientSockList.UnLock();
    return true;
}

int CKxClientService::PostSendToDataChannel(char *pData, int nLen,int bNow)
{
    return 0;
}

/**
 *
 * @brief   해당 클라이언트소켓을 닫는다.
 * @details
 * @param   CKxClientSession pclient 클라이언트 소켓 클래스
 * @date    2018-09-09
 * @version 0.0.1
 * @return  0 -> 성공, -1 -> 실패
 * @todo
 */
int CKxClientService::CloseClient(CKxClientSession *pclient)
{
    int clientfd;
    clientfd = pclient->GetSockFD();
    pclient->OnDisconnected();
#ifndef EPOLL_NEW
    del_epollfd(m_eventfd, clientfd);
#else
    Epoll_Del(m_eventfd, clientfd);
    CKxThreadCtx * lThread = m_pThreadPool->GetThread(pclient->m_threadno);
    lThread->Lock();
    lThread->m_clientnum--;
    lThread->UnLock();
#endif
	kxTraceLog(0,"[CloseClient] 소켓끊어짐: host %s port %d\n",pclient->GetHost(),pclient->GetPort());

    if (m_bAutoReconnect){
        pclient->CloseClient();
    }
    else{
        RemoveClientSock(clientfd);
        pclient->CloseClient();
        delete pclient;
    }
    return 0;
}
