recvmessthread.cpp

基于Linux下Epoll技术的EchoSvr

#include "RecvMessThread.h"
#include "Log.h"




G_RecvMessThread::G_RecvMessThread(G_ThreadPool *pool,DMCThread* dmcThread,LogThread* logThread,int normal,int auth,int maxCount) 
{
	counter = 0;
	g_threadPool = pool;
	this->dmcThread = dmcThread;
	this->logThread = logThread;
	normalTimeout = normal;
	authTimeout = auth;
	
	epfd = epoll_create(maxCount*2);
}

G_RecvMessThread::~G_RecvMessThread()
{
	close(epfd);
}

unsigned int G_RecvMessThread::getCounter()
{
	return counter;
}

//set socket as non blocking
bool G_RecvMessThread::setNonBlock(int sockfd)
{
	int opts = fcntl(sockfd , F_GETFL);
	if(-1 == opts)
	{
		printf("%s\n" , "fcntl F_GETFL is faild");
		return false;
	}

	opts = opts | O_NONBLOCK;
	if(fcntl(sockfd , F_SETFL , opts) < 0)
	{
		printf("%s\n" , "fcntl F_SETFL is faild");
		return false;
	}
	return true;
}
void G_RecvMessThread::subCounter()
{
	counterLock.wLock();
	counter--;
	counterLock.unwLock();

}
bool G_RecvMessThread::addEvent(int nSocket,unsigned int event)
{
		struct epoll_event ev;
		bzero(&ev , sizeof(ev));
		ev.data.fd = nSocket;
		ev.events = event|EPOLLET;
		int err = epoll_ctl(epfd , EPOLL_CTL_ADD , nSocket , &ev);
		
		if(err<0)
		{
			debug_output("ERR:addEvent,socket:%d error:%d\n" , nSocket,err);	
		}
		else
		{
			debug_output("ERR:addEvent,socket:%d result:%d\n" , nSocket,err);
		}
		return err>=0;
}
bool G_RecvMessThread::modEvent(int nSocket,unsigned int event)
{
		struct epoll_event ev;
		bzero(&ev , sizeof(ev));
		ev.data.fd = nSocket;
		ev.events = event|EPOLLET;
		int err = epoll_ctl(epfd , EPOLL_CTL_MOD , nSocket , &ev);
		if(err<0)
		{
			debug_output("ERR:addEvent,socket:%d error:%d\n" , nSocket,err);	
		}
		return err>=0;
}
bool G_RecvMessThread::delEvent(int nSocket)
{
		struct epoll_event ev;
		bzero(&ev , sizeof(ev));
		ev.events = EPOLLIN | EPOLLET |  EPOLLERR | EPOLLHUP | EPOLLPRI;
		ev.data.fd = nSocket;
		epoll_ctl(epfd, EPOLL_CTL_DEL, nSocket, &ev)>=0; 
}
void G_RecvMessThread::addSocket(int nSocket,char* ip,unsigned short port)
{
	
	if(setNonBlock(nSocket))
	{
		dmcThread->initSession(nSocket,ip,port); 			
		if(!addEvent(nSocket,EPOLLIN))
		{
			close(nSocket);
		}
		else
		{
			//TODO:add to map
			counterLock.wLock();
			counter++;
			counterLock.unwLock();			
		}
	}
}
void G_RecvMessThread::closeSocket(int nSocket,struct epoll_event* ev)
{
	//delEvent(nSocket);
	ev->data.fd=-1;
	close(nSocket);
  debug_output("CLOSE:socket%d\n",nSocket);
}
/////////////////////////////////
//recv some bytes of nLen
////////////////////////////////
int G_RecvMessThread::recvn(int nSocket , char *str , unsigned int nLen)
{
	int n = nLen;

	while(n > 0)
	{
		int nRet = recv(nSocket , str , n , MSG_NOSIGNAL);
		if(nRet <= 0)
		{
			if(errno == EINTR)
			{
				continue;
			}
			break;
				
		}
		n -= nRet;
		str += nRet;
	}
	return (nLen - n);
}

void G_RecvMessThread::readBindReq(char* buffer,struct BINDREQ* bindReq)
{
	bindReq->command = *((unsigned short*)(buffer));
	bindReq->command = ntohs(bindReq->command);
	bindReq->len =  *((unsigned int*)(buffer+2));
	bindReq->len = ntohl(bindReq->len);;
	bindReq->clientid = *((unsigned short*)(buffer+7));
	bindReq->clientid = ntohs(bindReq->clientid);
}
void G_RecvMessThread::makeBindRep(char* buffer,struct BINDREP* bindRep)
{
		unsigned short command = htons(bindRep->command);
		memcpy(buffer,&command,2);
		unsigned int len = htonl(bindRep->len);
		memcpy(buffer+2,&len,4);
		unsigned short result = htons(bindRep->result);
		memcpy(buffer+6,&result,2);
		memcpy(buffer+8,bindRep->key,6);
}
void G_RecvMessThread::bind(char* buffer,struct SESSION* session,int fd)
{
	struct BINDREQ bindReq;
	struct BINDREP bindRep;
	readBindReq(buffer,&bindReq);
	
	bindRep.command = COMMAND_BIND_REP;
	bindRep.len = 14;
	bindRep.result = SUCCESS;
	if(bindReq.command != COMMAND_BIND_REQ || bindReq.len != 9)
	{
		bindRep.result = ERR_COMMAND;
	}
	if(bindRep.result == 0)
	{
		struct CLIENT* client = dmcThread->getClientByID(bindReq.clientid);
		if(client == NULL || client->flag == -1)
			bindRep.result = ERR_USERID;
		else
		{	
			//makekey
			srand( (unsigned)time( NULL ) );
			int r = rand();
			memcpy(session->key,&r,4);
			r = rand();
			memcpy(session->key,&r,2);
			
			memcpy(bindRep.key,session->key,6);
			session->flag = BOUND;
			session->clientid = bindReq.clientid;
		}
	}
	if(bindRep.result != 0)
	{
		session->flag = SENDERR;
	}
	char temp[14];
	makeBindRep(temp,&bindRep);
	
	dmcThread->saveBuffer(fd,temp,14);
	modEvent(fd,EPOLLOUT);
}

void G_RecvMessThread::readValidateReq(char* buffer,struct VALIDATEREQ* validateReq)
{
	validateReq->command = *((unsigned short*)(buffer));
	validateReq->command = ntohs(validateReq->command);
	validateReq->len =  *((unsigned int*)(buffer+2));
	validateReq->len = ntohl(validateReq->len);;
	memcpy(validateReq->password,buffer+6,6);
}
void G_RecvMessThread::makeValidateRep(char* buffer,struct VALIDATEREP* validateRep)
{
		unsigned short command = htons(validateRep->command);
		memcpy(buffer,&command,2);
		unsigned int len = htonl(validateRep->len);
		memcpy(buffer+2,&len,4);
		unsigned short result = htons(validateRep->result);
		memcpy(buffer+6,&result,2);
}
/*
public void reportOnline(int client_id,String remote_ip,int remote_port){
		String date = getCurrentTime();
		String sql = "call REPORT_STATUS("
				+client_id+","+date+","+date+",'"
				+remote_ip+"',"+remote_port+",'"+serverName+"',0)";
		execute(sql);
	}
	public void reportOffline(int client_id){
		String date = getCurrentTime();
		String sql = "call REPORT_STATUS("
			+client_id+","+date+","+date+",' ',0,'"+serverName+"',1)";
		execute(sql);
	}
	public void reportMO(int client_id,int size,int protocal){
		String sql = "call REPORT_TRANSFER("
							+client_id+",0,"+size+","+getCurrentTime()+","+protocal+",'"+serverName+"')";
		execute(sql);
		Client c = getClient(new Integer(client_id));
		if(c!=null)
			c.mo += size;
	}
	public void reportMT(int client_id,int size,int protocal){
		String sql = "call REPORT_TRANSFER("+client_id+",1,"+size+","+getCurrentTime()+","+protocal+",'"+serverName+"')";
		execute(sql);
		Client c = getClient(new Integer(client_id));
		if(c!=null)
			c.mt += size;
	}
	public void reportAuthResult(int client_id,int result){
		String sql = "call REPORT_AUTH_RESULT("+client_id+","+result+","+getCurrentTime()
					+",'"+serverName+"')";
		execute(sql);
	}
*/

void G_RecvMessThread::validate(char* buffer,struct SESSION* session,int fd)
{
	struct VALIDATEREQ validateReq;
	struct VALIDATEREP validateRep;
	readValidateReq(buffer,&validateReq);
	validateRep.command = COMMAND_VALIDATE_REP;
	validateRep.len = 8;
	validateRep.result = SUCCESS;
	struct CLIENT* client = NULL;
	if(validateReq.command != COMMAND_VALIDATE_REQ || validateReq.len != 12)
	{
		validateRep.result = ERR_COMMAND;
	}
	if(validateRep.result == 0)
	{
		client = dmcThread->getClientByID(session->clientid);
		if(client == NULL || client->flag == -1)
			validateRep.result = ERR_USERID;
		else
		{	
			bool isValidated = true;
			
			for(int i = 0;isValidated && i<6;i++)
			{
				isValidated = (session->key[i] ^ client->password[i]) % (i+23) == validateReq.password[i];
			}
			if(!isValidated)
			{
				validateRep.result = ERR_PASSWORD;
				//TODO:::::::::::err pass
				logThread->ReportAuthResult(session->clientid,5);
			}
		}
	}
	if(validateRep.result == SUCCESS)
	{
			int other = connectTarget(&client->servaddr);
			if(other>=0)
			{
				modEvent(fd,EPOLLET);
				session->other = other;
				session->flag = CONNECT;
				struct SESSION* otherSession = dmcThread->initSession(other,session->ip,session->port);		
				otherSession->flag = CONNECTING;
				otherSession->other = fd;
				otherSession->clientid = session->clientid;
				addEvent(other,EPOLLIN|EPOLLOUT);
				debug_output("Ready to connect:%d\n",other);
			}
			else
			{
				validateRep.result = ERR_TARGETSVR;
				struct CLIENT* c = dmcThread->getClientByID(session->clientid);
				debug_output("CONN ERR: id:%d,ip:%s,port:%d------\n",session->clientid,c->ip,c->port);
			
			}
	}
	if(validateRep.result != SUCCESS)
	{
		session->flag = SENDERR;