/*
 client socket context
 ---------------------------------------------------------------------------
 $Id: $ $Revision: $
 $HeadURL: $
 durumul@gmail.com
*/

/**
 * @file
 * @brief client socket context
 * @detail 
 */

/*
 * History:
 * ===========================================================================
 * 2007.07.11  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 <vector>
#include <queue>
#include <pthread.h>

#include "tokenizer.h"
#include "iocq.h"
#include "kxThread.h"
/*#include "dbpool.h"*/
#include "epollcomn.h"
#include "kxSockSession.h"
#include "kxtracelog.h"
#include "network.h"
using namespace std;

CKxSockSession::CKxSockSession()
{
	m_sockfd = INVALID_SOCKET;
	m_threadno = -1;
	Init();
}

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

int CKxSockSession::getNextSendTRID()
{
	m_nSendTRID++;
	if (m_nSendTRID > MAX_TRID_LIMIT)
		m_nSendTRID = 1;
	return m_nSendTRID;
}

void CKxSockSession::SetSockFD(int sockfd,sockaddr_in *pClientAddr)
{ 
	m_sockfd = sockfd;

	if(pClientAddr)
		memcpy(&m_sockaddr,pClientAddr,sizeof(m_sockaddr));
	else{
		union {
			struct sockaddr sa;
			struct sockaddr_in in;
		} sin;

		socklen_t l = sizeof(sockaddr);
		int res = getpeername(m_sockfd, (sockaddr*)&sin.sa, &l);
		memcpy(&m_sockaddr,(sockaddr_in*)&sin.in,sizeof(m_sockaddr));
	}
	char *remoteaddr = inet_ntoa(m_sockaddr.sin_addr);
	strncpy(m_ip,remoteaddr,sizeof(m_ip));
	//kxTraceLog(0,"client ip=[%s]\n",m_ip);
}

//연결후 epoll 이벤트에 넣기전에 데이터가 올라옴
//빠른 소켓 epoll 등록처리를 위해 최소화 실행
void CKxSockSession::SetSockFD(int sockfd)
{ 
	m_sockfd = sockfd;
}

char *CKxSockSession::GetRemoteIP()
{
    return m_ip;
}

bool CKxSockSession::Init()
{
	pthread_mutex_init(&m_mut, NULL);
	m_pSockCtx = NULL;
    m_nSendTRID = 0;
    m_sockfd = INVALID_SOCKET;
	return true;
}

bool CKxSockSession::Destroy()
{
	FreeContext();
	pthread_mutex_destroy(&m_mut);
	kxTraceLog(0,"CKxSockSession::Destroy\n");
	return true;
}

int CKxSockSession::GetAvailIOBuffQueueSize()
{
	IOBUFF *pIOBuff = m_pSockCtx->iobuff->buff;
	int nAvailBuffSize = CIOCQ::GetAvailSize(pIOBuff);
    return nAvailBuffSize;
}

int CKxSockSession::Enqueue(char *buf, int len)
{
	int i;
	int rc = 0;
	IOBUFF *pIOBuff = m_pSockCtx->iobuff->buff;
	int nAvailBuffSize1 = CIOCQ::GetAvailSize(pIOBuff);
	int addlen = CIOCQ::Enqueue(pIOBuff,buf,len);
    int nAvailBuffSize2 = CIOCQ::GetAvailSize(pIOBuff);
	kxTraceLog(TL_HIGH,"Enqueue::ToLen=%d AvailSize=%d => %d AddedLen=%d\n",
        len,nAvailBuffSize1,nAvailBuffSize2,addlen);
	return addlen;
}

//현재 받은 데이터 체크
int CKxSockSession::PeekDequeue()
{
    int rc = 0;
    IOBUFF_CONTEXT *iobuff = m_pSockCtx->iobuff;
    rc = CIOCQ::GetDataSize(iobuff->buff);
    return rc;
}

int CKxSockSession::InitContext(int buffsize)
{
	if(m_pSockCtx)
		return -1;
	m_pSockCtx = new SOCKET_CONTEXT;
	m_pSockCtx->iobuff = new IOBUFF_CONTEXT;//소켓용accept버퍼
	m_pSockCtx->iobuff->buff = CIOCQ::Init(buffsize);//MAX_IOBUFFSIZE);
	m_pSockCtx->iobuff->buflen = 0;
	return 0;
}

int CKxSockSession::FreeContext()
{
	if(m_pSockCtx){
		CIOCQ::Free(m_pSockCtx->iobuff->buff);
		delete m_pSockCtx->iobuff;
		delete m_pSockCtx;
		//kxTraceLog(0,"freecontext:Delete Recv Context\n");
	}
	m_pSockCtx = NULL;

    FreeSendContext();
	return 0;
}

int CKxSockSession::FreeSendContext()
{
    int no = 0;
    IOBUFF_CONTEXT *iobuff = NULL;
    while (!m_SendCtx.empty()){
        iobuff = m_SendCtx.front();
        m_SendCtx.pop();
        //송신일때는 버퍼해제
        CIOCQ::Free(iobuff->buff);
        delete iobuff;
        no++;
    }
    //kxTraceLog(0,"freecontext:Delete Send Context [%d]\n",no);
    return no;
}

int CKxSockSession::OnConnected()
{

    return 0;
}

int CKxSockSession::OnDisconnected()
{

    return 0;
}

//Edge Trigger를 사용할 경우에는 read 시에 모두 읽어와야 한다.
//왜냐하면, LT는 상태감지라 데이타를 읽어가지 않아도,네트워크버퍼에
//데이타가 존재하면 select나 poll 처럼 계속 이벤트가 발생하지만,
//ET는 상태변화가 있을때 단 한번만 이벤트가 발생하기 때문.
//krkim

/**
*
* @brief   데이터 로우레벨 수신처리
* @details Edge Trigger를 사용할 경우에는 read 시에 모두 읽어와야 한다.
* @param
* @date    2018-09-09
* @version 0.0.1
* @return  0 -> 성공, SOCK_TRY_CLOSE -> 실패(클라이언트 소켓 끊어버림)
* @todo
*/
int CKxSockSession::OnRecvPacket()
{
	int rc = 0,rc2 = 0,nc = 0;
	int readn,writen;
	int totalreaded = 0;
    int peeklen,availsize;
	char readbuf[PF_RECVSIZE+1] = { '\0', };
    int remainlen = 0;
    int sockerrno = 0;
	//Lock();
	for(;;){
		//edge trigger 모드에서는 이벤트가 한번만 발생하므로 모두 읽어줘야한다.
		//그렇지 않으면 LT 와는 다르게 다음번 epoll_wait 가 내려오지 않는다.
		//readn = read(m_sockfd, readbuf, PF_BUFSIZE);
#ifdef USE_CHECK_BUFF_AVAIL  //2018.10.02
        //여기서 데이터를 읽어버리면 nic 카드로 부터 읽고 처리는 하지않고,유실되므로
        //수신버퍼가 모자를경우는 아예 읽지를 않는다.(데이터 보존정책)
        //이렇게하면 epoll은 계속 즉시 읽어가라고 signal이 날라온다.(큐읽어가는 쓰레드를 따로 분리해야함)
        //또는 데이터를 버리거나.

        peeklen = PeekRecv();
        availsize = GetAvailIOBuffQueueSize();
        if (peeklen > availsize){
			kxTraceLog(0,"client(%d) warning,insufficent receive buffer,need to flush.\n",m_sockfd);
			rc = SOCK_BUFFER_RECV_NEEDTO_FLUSH;
			return rc;
        } 
#endif
        //TODO::송신도 서버에서 송신하게끔 중앙관리 필요,소켓 커넥션에서 이미 해제된 경우 대처.2018.10.17
        memset(readbuf, 0, sizeof(PF_RECVSIZE));
        //kxTraceLog(0,"recv client(%d) try\n",m_sockfd);
        //BUGFIX:: 수신버퍼에 연속된 뭉쳐진 패킷이 올경우,통신용 환형큐 버퍼를 초과할 경우 패킷의 처리오류.
        //따라서,읽기를 최소화 한다
        //통신수신데이터가 이전 수신데이터 처리속도보다 빨라서,수신이 밀릴경우 문제대응.
        //KRKIM 2018.10.17
        
        errno = 0;
        readn = recv(m_sockfd, readbuf, PF_RECVSIZE,0);
        sockerrno = errno;
        kxTraceLog(0,"Recv:sockfd(%d) [%d bytes] [%.*s] errno=%d(%s)\n",m_sockfd,readn,readn,readbuf,sockerrno,strerror(sockerrno));
		if(readn < 0){
			//epoll의 성능비교
			//http://www.joinc.co.kr/modules/moniwiki/wiki.php/Site/Network_Programing/AdvancedComm/epoll24
			//non-blocking 비동기식이므로 블럭여부를 처리한다.
			if(sockerrno == EAGAIN || sockerrno == 11) //=EWOULDBLOCK
			{
				kxTraceLog(0,"client(%d) EAGAIN = %d EWOULDBLOCK ip=[%s] wait..%d(%s)\n",
                    m_sockfd,EAGAIN,m_ip,sockerrno,strerror(sockerrno));
				break;
			}
			//if(errno == EAGAIN || errno == EPIPE || errno == ECONNREFUSED || errno == 111 || errno == EIO) //=EWOULDBLOCK
			//{
			//	kxTraceLog(0,"client(%d) readn = %d errno=%d(%s) but wait..\n",m_sockfd,readn,errno,strerror(errno));
			//	break;
			//}
            kxTraceLog(0,"client(%d) try closing errno=%d(%s) ip=[%s]\n",m_sockfd,sockerrno,strerror(sockerrno),m_ip);
			rc = SOCK_TRY_CLOSE;
			break;
		}
		else if(readn == 0){
			OnReceive();
			// 0 is end of file(client closed),
			kxTraceLog(0,"client(%d) closed - end of file detected total readed=%d,errno=%d(%s) ip=[%s]\n",
				m_sockfd,totalreaded,errno,strerror(errno),m_ip);
			if(sockerrno == EAGAIN || sockerrno == 11) //=EWOULDBLOCK
			{
				kxTraceLog(0,"client(%d) EAGAIN = %d EWOULDBLOCK ip=[%s] wait..%d(%s)\n",
                    m_sockfd,EAGAIN,m_ip,sockerrno,strerror(sockerrno));
				break;
			}
			rc = SOCK_TRY_CLOSE;
			break;
		}
		totalreaded += readn;
		//큐가 꽉찼으면,먼저 큐를 읽어 내보내는 처리를 한후,다시 큐에 삽입시도한다.
        int remainlen0;
        remainlen0 = remainlen = PeekDequeue();
        while (remainlen > 0){
            rc2 = OnReceive();
			if (rc2 == SOCK_DATA_INSUFFICENT || remainlen == remainlen0)
				break;
            remainlen = PeekDequeue();
        }
		writen = Enqueue(readbuf, readn);//버퍼오버런
		if(writen < readn){
			rc = OnReceive();
			if (rc == SOCK_DATA_INSUFFICENT)
				break;
			//이때,큐가 그래도 꽉찼으면 오류이다.
			if(Enqueue(readbuf+writen,readn-writen) < 0){//버퍼오버런
				kxTraceLog(0,"client(%d) error,receive buffer overflow - data lose.\n",m_sockfd);
				rc = SOCK_BUFFER_OVERRUN;
				break;
			}
			//break;
		}
        //통신수신데이터가 이전 수신데이터 처리속도보다 빨라서,수신이 밀릴경우 문제대응.
        else{
            remainlen = PeekDequeue();
            if (remainlen > 0){
                rc2 = OnReceive();
                if (rc2 == SOCK_DATA_INSUFFICENT || remainlen == remainlen0)
                    break;
            }
        }
		usleep(1);
		//char fmt[80];
		//sprintf(fmt,"client(%%d) Recv[%%.%ds] len = %%d\n",readn);
		//tracelog(0,fmt,m_sockfd,readbuf,readn);
	}
	//큐에 데이타가 있으면 수신데이타를 처리한다.
	nc = OnReceive();
	if (nc == SOCK_TRY_CLOSE)
		rc = SOCK_TRY_CLOSE;
	//UnLock();
	return rc;
}

/**
*
* @brief   데이터 수신처리
* @details
* @param
* @date    2018-09-09
* @version 0.0.1
* @return  0 -> 성공, -1 -> 실패(클라이언트 소켓 끊어버림)
* @todo
*/
int CKxSockSession::OnReceive()
{
	int rc = SOCK_DATA_OK;
	Tokenizer tkutil;	
	int pklen,cmdlen;
	IOBUFF_CONTEXT *iobuff = m_pSockCtx->iobuff;
	char read[1024+1];//패킷길이만큼 충분한 사이즈로 읽어온다.

	pklen = CIOCQ::GetDataSize(iobuff->buff);
	if(pklen <= 0){
		return SOCK_DATA_OK;
	}
	while(1){//완성된 패킷인 경우만 프로세싱 한다.붙어서 올수있으므로 loop
		int payloadlen = 0;
		int readlen = CIOCQ::Peek(iobuff->buff,read,1024);
		if(readlen == 0)
			break;
		int cmdlen = tkutil.FindCRLF(read,readlen);
		if(cmdlen > 0){
			if(cmdlen <= 2){
				CIOCQ::ShiftFront(iobuff->buff,cmdlen);//(\r\n)만 온경우 shift하여 스킵
				break;
			}
			//\r\n앞에서 끊는다.
			char *pcmd = tkutil.InitTokens(read,cmdlen -2);
			if(!strcmp(pcmd,"BYE")){
				rc = SOCK_TRY_CLOSE;
				break;//close socket
			}
			payloadlen = ProcessCmd(pcmd,iobuff);
			//if(payloadlen == 0){//데이타가 모자름
			if(payloadlen == SOCK_DATA_INSUFFICENT){//데이타가 모자름
				rc = SOCK_DATA_INSUFFICENT;
				break;
			}
			//if (payloadlen < 0){//소켓 끊기
			if(payloadlen == SOCK_TRY_CLOSE){//소켓 끊기
				rc = SOCK_TRY_CLOSE;
				break;
			}
			CIOCQ::ShiftFront(iobuff->buff,cmdlen + payloadlen);//처리한만큼 shift + payload가 있는경우
		}
		else
			break;
	}
	return rc;
}
//return 0: for next recv wait mode
//       1: for next write data still remains
//       -1: for close socket
int CKxSockSession::OnSendPacket()
{
	int rc;
	rc = SendPacket();
	return rc;
}

//바로직접 송신하기 위함
int CKxSockSession::SendPacket()
{
	int rc = SOCK_DATA_OK;
	int no = 0;
	bool retrynexttime;
	////Lock();
	retrynexttime = false;
	IOBUFF_CONTEXT *iobuff = NULL;
	while (!m_SendCtx.empty()){
		iobuff = m_SendCtx.front();
		//send buffer는 순차적으로 들어있다(front < rear)
		//char read[1024];//패킷길이만큼 충분한 사이즈로 읽어온다.
		char *pbuff = CIOCQ::GetFrontPtr(iobuff->buff);
		int ToSendLen = CIOCQ::GetDataSize(iobuff->buff);
		int sendflags = 0, SentLen = 0;
		int cnt = 0;
		char szbuff[PF_BUFSIZE];

		if (ToSendLen == 0){
			m_SendCtx.pop();
			kxTraceLog(TL_MID, "[SendPacket %d] flush(%d) ToSendLen=%d\n", no,cnt, ToSendLen);
			CIOCQ::Free(iobuff->buff);
			delete iobuff;
			continue;
		}
		for (; ToSendLen > 0; cnt++){
			//kxTraceLog(0, "SendPacket[%d] %d,[%s] ToSendLen=%d\n", no, cnt,pbuff, ToSendLen);
			strncpy(szbuff, pbuff, ToSendLen);
			*(szbuff + ToSendLen) = (char)NULL;
			SentLen = send(m_sockfd, pbuff, ToSendLen, sendflags);
			kxTraceLog(TL_MID, "[SendPacket %d] sockfd(%d) flush(%d) [%.*s] ToSendLen=%d SentLen=%d\n",
                no,m_sockfd, cnt, SentLen,pbuff, ToSendLen, SentLen);
			if (SentLen < 0){
				if (errno == EAGAIN){ //=EWOULDBLOCK
					retrynexttime = true;
					break;
				}
                kxTraceLog(TL_MID, "[SendPacket %d] sockfd(%d) errno(%d)(%s)\n",
                    no,m_sockfd,errno,strerror(errno));
                if (errno == EPIPE || errno == ECONNREFUSED || errno == 111){//EIDRM
					retrynexttime = true;
					break;
                }
                if (errno == EIO){ //Input/output error 2018.10.09
					retrynexttime = true;
					break;
                }
            }
			else if (ToSendLen == SentLen){
				m_SendCtx.pop();
				//kxTraceLog(TL_MID, "[SendPacket %d] flush(%d) SentLen=%d\n", no,cnt, SentLen);
				//송신일때는 버퍼해제
				CIOCQ::Free(iobuff->buff);
				delete iobuff;
				break;
			}
			else{//현재 패킷을 네트워크 송신버퍼에 공간이 없어 전부 보내지 아니한 경우
				CIOCQ::ShiftFront(iobuff->buff, SentLen);//처리한만큼 shift
				ToSendLen = CIOCQ::GetDataSize(iobuff->buff);//reload after ShiftFront
				retrynexttime = true;
				break;
			}
		}
		//현재 패킷을 네트워크 송신버퍼에 공간이 없어 전부 보내지 아니한 경우
		if (retrynexttime)
			break;
		usleep(1);
		no++;
	}
	//다음번 EPOLLOUT 이벤트가 필요함
	if (retrynexttime || !m_SendCtx.empty()){
		rc = SOCK_DATA_PENDING_SHIFT_SENDBUFFER;
	}
	////UnLock();
	return rc;
}

///////////////////////////////////////////////////////////////////////////////
// return value : > 0 : add to pending shift size
//                = 0 : ok
//                < 0 : insufficent data size need to more,so no shift now.
int CKxSockSession::ProcessCmd(char *pCommand,IOBUFF_CONTEXT *iobuff)
{
	int rc = SOCK_DATA_OK;
	Tokenizer tokenlist;
	tokenlist.InitTokens(pCommand);
	int nTokencnt = tokenlist.MakeTokens();
	//성공하면 >= 0 의 값을 리턴, > 0 인 경우 
	//추가처리할 payload 명령어가 있다는 의미.
	//-1인 경우,데이타가 모자르다는 의미.
	return rc;
}

int CKxSockSession::PostSend(char *pData, int nLen)
{
	IOBUFF_CONTEXT *iobuff = new IOBUFF_CONTEXT;
	//iobuff->operation = OP_WRITE;
	//tracelog(0,"PostSend: pData=[%.10s] nLen=%d\n",pData,nLen);
	iobuff->buff = CIOCQ::Init(nLen+1);//must bigger
	CIOCQ::Enqueue(iobuff->buff,pData,nLen,true);
	char *pbuff = CIOCQ::GetFrontPtr(iobuff->buff);

	int sendflags = 0,sent = nLen;
#if 0 //// 송신용 버퍼없이 보내는 경우
	sent = send(_cfd,pbuff,nLen,sendflags);
	tracelog(0,"PostSend:bc= %d,pbuff=[%s] nLen=%d sent=%d\r\n",bc,pbuff,nLen,sent);
	//송신일때는 버퍼해제
	CIOCQ::Free(iobuff->buff);
	delete iobuff;
	if(sent != nLen)
		return false;
#else //송신용 버퍼사용(서버소켓이 관리)
	m_SendCtx.push(iobuff);
#endif
	return 0;
}

int CKxSockSession::PeekRecv()
{
    int len;
    char readbuf[PF_BUFSIZE+1] = { '\0', };
    len = recv(GetSockFD(), readbuf, PF_BUFSIZE, MSG_PEEK);
    return len;
}

int CKxSockSession::PostSendNow(char *pData, int nLen)
{
    bool bc;
    bc =PostSend(pData, nLen);
    SendPacket();
    return 0;
}

bool CKxSockSessionList::Init()
{
	return true;
}

bool CKxSockSessionList::Destroy()
{
	Lock();
	m_clientlist.clear();
	UnLock();
	return true;
}

bool CKxSockSessionList::Add(CKxSockSession *pUser)
{
	Lock();
	m_clientlist.push_back(pUser);
	UnLock();
	return true;
}

int CKxSockSessionList::GetCount()
{
	Lock();
	int rc = m_clientlist.size();
	UnLock();
	return rc;
}

bool CKxSockSessionList::Remove(CKxSockSession *pUser)
{
	bool rc = false;
	list < CKxSockSession* >::iterator iter;
	Lock();
	for(iter = m_clientlist.begin(); iter != m_clientlist.end(); ++iter)
	{
		if(pUser == (*iter)){
			m_clientlist.erase(iter);
			rc = true;
			break;
		}
	}
	UnLock();
	return rc;
}

bool CKxSockSessionList::Remove(int sockfd)
{
	bool rc = false;
	list < CKxSockSession* >::iterator iter;
	Lock();
	for(iter = m_clientlist.begin(); iter != m_clientlist.end(); ++iter)
	{
		if(sockfd == (*iter)->GetSockFD()){
			delete(*iter);
			m_clientlist.erase(iter);
			rc = true;
			break;
		}
	}
	UnLock();
	return rc;
}

CKxSockSession *CKxSockSessionList::Find(int sockfd)
{
	bool rc = false;
	CKxSockSession *currentuser = NULL;

	list < CKxSockSession* >::iterator iter;
	Lock();
	for(iter = m_clientlist.begin(); iter != m_clientlist.end(); ++iter)
	{
		if(sockfd == (*iter)->GetSockFD()){
			currentuser = *iter;
			rc = true;
			break;
		}
	}
	UnLock();
	return currentuser;
}



