comm.cpp

通讯中间件

		OnReceive(&packet);

		return COMM_ERR_NONE;
	}

	memset(buf, 0, sizeof(buf));
	*(unsigned long *)&buf[0] = htonl(packet.m_BodyLen + 16);
	*(unsigned long *)&buf[4] = htonl(COMM_DATA);
	*(unsigned long *)&buf[8] = htonl(m_ModuleID);
	*(unsigned long *)&buf[12] = htonl(packet.m_ModuleID);
	memcpy(&buf[16], packet.m_BodyBuf, packet.m_BodyLen);

	nFoundIndex = -1;
	nSendSuccess = -1;

	m_Lock.Lock();
	for (i = 0; i < COMM_LINK_MAX; i++)
	{
		if (m_Link[i].m_SocketID != INVALID_SOCKET)
		{
			if (m_Link[i].m_ModuleID == packet.m_ModuleID)
			{
				nFoundIndex = i;

				ret = Send(m_Link[i].m_SocketID, buf, (int)(packet.m_BodyLen + 16), 0);
				if (ret == (int)(packet.m_BodyLen + 16))
				{
					nSendSuccess = 0;
				}
				break;
			}
		}
	}
	m_Lock.UnLock();

	if (nFoundIndex < 0 || nFoundIndex >= COMM_LINK_MAX)
	{
		g_CommLog.Print("[应用错误] 发送消息包异常，没找到模块：%u", packet.m_ModuleID);
	}
	else
	{
		if (nSendSuccess < 0)
		{
			g_CommLog.Print("[应用错误] 发送消息包失败，模块：%u", packet.m_ModuleID);
		}
	}

	if (nSendSuccess == 0)
	{
		return COMM_ERR_NONE;
	}
	else
	{
		ret = PutPacket(&packet);
		if (ret == COMM_ERR_NONE)
		{
			return COMM_ERR_WAITSEND;
		}
		else
		{
			g_CommLog.Print("[提示信息] 重发缓冲区满");

			return COMM_ERR_WAITFULL;
		}
	}

	return COMM_ERR_NONE;
}

int TComm::AddServer(unsigned short uListenPort)
{
	if (uListenPort == 0)
	{
		g_CommLog.Print("[系统错误] 增加服务端，监听端口为零");

		return COMM_ERR_PARAM;
	}

	int i, found;

	// 首先查找是否已经存在该监听端口
	found = -1;
	m_Lock.Lock();
	for (i = 0; i < COMM_LISTEN_MAX; i++)
	{
		if (m_Listen[i].m_ListenPort == uListenPort)
		{
			found = i;
			break;
		}
	}
	m_Lock.UnLock();
	if (found >= 0 && found < COMM_LISTEN_MAX)
	{
		return COMM_ERR_NONE;
	}

	// 如果是新监听端口则放入监听端口数组中
	m_Lock.Lock();
	for (i = 0; i < COMM_LISTEN_MAX; i++)
	{
		if (m_Listen[i].m_ListenPort == 0)
		{
			if (m_Listen[i].m_ListenSocket != INVALID_SOCKET)
			{
				closesocket(m_Listen[i].m_ListenSocket);
				m_Listen[i].m_ListenSocket = INVALID_SOCKET;
			}
			m_Listen[i].m_ListenPort = uListenPort;
			m_Listen[i].m_ListenTime = 0;

			break;
		}
	}
	m_Lock.UnLock();

	if (i < 0 || i >= COMM_LISTEN_MAX)
		return COMM_ERR_TOOMANYLISTEN;

	return COMM_ERR_NONE;
}

int TComm::AddClient(const char *szHost, unsigned short uPort)
{
	char *param;
	DWORD dwThreadID;
	HANDLE hRet;

	if (szHost == NULL)
	{
		g_CommLog.Print("[系统错误] 增加客户端，目的主机为空");

		return COMM_ERR_PARAM;
	}

	if (uPort == 0)
	{
		g_CommLog.Print("[系统错误] 增加客户端，目的端口为零");

		return COMM_ERR_PARAM;
	}

	param = NULL;
	param = (char *)new char[80];
	if (param == NULL)
	{
		g_CommLog.Print("[系统错误] 增加客户端，内存分配失败");

		return COMM_ERR_MEMORY;
	}
	memset(param, 0, 80);
	*(unsigned long *)&param[0] = (unsigned long)this;
	*(unsigned long *)&param[4] = (unsigned long)uPort;
	strncpy(&param[8], szHost, 64 - 1);

	hRet = (HANDLE)NULL;

#ifdef _WIN32
	hRet = CreateThread((LPSECURITY_ATTRIBUTES )NULL,
					0,
					(LPTHREAD_START_ROUTINE)ClientConnectThread,
					(LPVOID)param,
					0,
					&dwThreadID);

	if (hRet == (HANDLE)NULL)
	{
		delete param;
		param = NULL;

		g_CommLog.Print("[系统错误] 增加客户端，创建线程失败");

		return COMM_ERR_CREATETHREAD;
	}
#else
	if (pthread_create(&hRet, NULL, ClientConnectThread, param) != 0)
	{
		delete param;
		param = NULL;

		g_CommLog.Print("[系统错误] 增加客户端，创建线程失败");

		return COMM_ERR_CREATETHREAD;
	}
#endif

	return COMM_ERR_NONE;
}

SOCKET TComm::Connect(const char *host, unsigned short port)
{
	SOCKET sockfd = INVALID_SOCKET;
	struct sockaddr_in serv_addr;
	struct in_addr  addr;
	struct hostent  *hp = NULL;
	struct protoent *protos = NULL;
	int i;
/*#ifdef _WIN32
	int bNodelay = 1;
	int err;
#endif*/
	
	protos = getprotobyname("tcp");

    if ((hp = gethostbyname(host)) == NULL) 
	{
        return INVALID_SOCKET;
    }
    
	if ((sockfd = socket(PF_INET, SOCK_STREAM, 
		(protos != NULL) ? protos->p_proto : 0)) == SOCKET_ERROR) 
	{
    	return INVALID_SOCKET;
  	}

/*#ifdef _WIN32
	err = setsockopt(sockfd, 
					IPPROTO_TCP, 
					TCP_NODELAY, 
					(char *)&bNodelay, 
					sizeof(bNodelay));//不采用延时算法
	if (err != 0)
	{
		closesocket(sockfd);
    	return INVALID_SOCKET;
	}
#endif*/

	for (i=0; hp->h_addr_list[i] != NULL;i++)
	{
		memcpy(&addr, hp->h_addr_list[i] , hp->h_length);
		memset(&serv_addr,0, sizeof(serv_addr));
  		serv_addr.sin_family      = AF_INET;
  		serv_addr.sin_addr		  = addr;
  		serv_addr.sin_port        = htons(port);
		if (connect(sockfd, 
			(struct sockaddr *)&serv_addr, 
			sizeof(serv_addr)) != SOCKET_ERROR) 
		{
			i = -1;
			break;
		} 
	}
	if (i != -1)
	{
		closesocket(sockfd);
    	return INVALID_SOCKET;
	}

	return sockfd;
}

SOCKET TComm::Listen(unsigned short port)
{
	SOCKET sockfd;
	struct sockaddr_in serv_addr;
	struct protoent *protos = NULL;
	int bReuseFlag;
	int err;
    
	protos = getprotobyname("tcp");

	if ((sockfd = socket(PF_INET, SOCK_STREAM, 
		(protos != NULL) ? protos->p_proto : 0)) == INVALID_SOCKET) 
	{
    	return INVALID_SOCKET;
  	}
	
	bReuseFlag = 1;
	err = setsockopt(sockfd, 
					SOL_SOCKET, 
					SO_REUSEADDR, 
					(char *)&bReuseFlag, 
					sizeof(bReuseFlag));//重新使用旧的地址
	if (err != 0)
	{
		closesocket(sockfd);
    	return INVALID_SOCKET;
	}

	memset(&serv_addr, 0, sizeof(serv_addr));
  	serv_addr.sin_family      = AF_INET;
  	serv_addr.sin_addr.s_addr = htonl(INADDR_ANY);  
	serv_addr.sin_port        = htons(port);
  	
  	if (bind(sockfd, 
		(struct sockaddr *)&serv_addr, 
		sizeof(serv_addr)) == SOCKET_ERROR) 
	{
		closesocket(sockfd);
    	return INVALID_SOCKET;
  	}
	
    if (listen(sockfd, SOMAXCONN) == SOCKET_ERROR) 
	{
		closesocket(sockfd);
        return INVALID_SOCKET;
    }

    return sockfd;
}

int TComm::Send(SOCKET s, void *buf, int len, int flags)
{
	SOCKET sock;
	int left_bytes; 
	int send_bytes; 
	char *ptr;

	/*fd_set wset, eset;
	struct timeval timeout;
	int nready;*/
	
	if (s == INVALID_SOCKET)
	{
		g_CommLog.Print("[系统错误] 底层发送函数接收参数中描述字无效");
		return -1;
	}

	if (buf == NULL)
	{
		g_CommLog.Print("[系统错误] 底层发送函数接收参数中缓冲区为空");
		return -1;
	}

	if (len <= 0)
	{
		g_CommLog.Print("[系统错误] 底层发送函数接收参数中长度小于等于零");
		return -1;
	}

	sock = s;

	ptr = (char *)buf; 
	left_bytes = len; 
	while (left_bytes > 0) 
	{ 
		/*memset(&timeout, 0, sizeof(timeout));
		timeout.tv_sec = 5;
		timeout.tv_usec = 0;
		
		FD_ZERO(&wset);
		FD_ZERO(&eset);
		FD_SET(sock, &wset);
		FD_SET(sock, &eset);

		nready = select(sock + 1,  NULL, &wset, &eset, &timeout);
		if (nready < 0)
		{
			closesocket(sock);

			g_CommLog.Print("[系统错误] 底层发送函数等待出错");
			return -1;
		}
		else if (nready == 0)
		{
			continue;
		}
		else
		{
			if (FD_ISSET(sock, &eset))
			{
				closesocket(sock);

				g_CommLog.Print("[系统错误] 底层发送函数等待异常");
				return -1;
			}
		}*/

		send_bytes = send(sock, ptr, left_bytes, flags);
		if (send_bytes <= 0) 
		{ 
#ifdef _WIN32
			if (errno == WSAEINTR) 
				send_bytes = 0; 
			else
			{
				closesocket(sock);
				return -1; 
			}
#else
			if (errno == EINTR) 
				send_bytes = 0; 
			else
			{
				closesocket(sock);
				return -1; 
			}
#endif
		} 
		
		left_bytes -= send_bytes; 
		ptr += send_bytes; 
	}
	 
	return (len - left_bytes);
}

TLink * TComm::AssignLink() 
{
	int i;
	
	m_Lock.Lock();
	for (i = 0; i < COMM_LINK_MAX; i++)
	{
		if (m_Link[i].Status() == 0)
		{
			m_Link[i].Assign();
			break;
		}
	}
	m_Lock.UnLock();

	if (i >= 0 && i < COMM_LINK_MAX)
		return &m_Link[i];

	return NULL;
}

int TComm::GetPacket(TPacket *lpPacket)
{
	if (lpPacket == NULL)
	{
		return COMM_ERR_PARAM;
	}

	int ret;

	ret = COMM_ERR_WAITEMPTY;

	m_Lock.Lock();
	if (m_WaitSendGetPos != m_WaitSendPutPos)
	{
		if (m_WaitSendGetPos >= 0 && m_WaitSendGetPos < COMM_WAITSEND_MAX)
		{
			*lpPacket = m_WaitSendPacket[m_WaitSendGetPos];
			m_WaitSendGetPos++;
			if (m_WaitSendGetPos >= COMM_WAITSEND_MAX)
			{
				m_WaitSendGetPos = 0;
			}

			ret = COMM_ERR_NONE;
		}
	}
	m_Lock.UnLock();

	return ret;
}

int TComm::PutPacket(const TPacket *lpPacket)
{
	if (lpPacket == NULL)
	{
		return COMM_ERR_PARAM;
	}

	int ret;

	ret = COMM_ERR_WAITFULL;

	m_Lock.Lock();
	if (((m_WaitSendPutPos + 1)%COMM_WAITSEND_MAX) != m_WaitSendGetPos)
	{
		m_WaitSendPacket[m_WaitSendPutPos] = *lpPacket;
		m_WaitSendPutPos++;
		if (m_WaitSendPutPos >= COMM_WAITSEND_MAX)
		{
			m_WaitSendPutPos = 0;
		}

		ret = COMM_ERR_NONE;	
	}
	m_Lock.UnLock();

	return ret;
}

void TComm::AddTrustIP(const char *ip)
{
	if (ip == NULL)
		return;

	unsigned long uIP;
	int i;

	uIP = (unsigned long)inet_addr(ip);
	if (uIP == 0)
		return;

	m_Lock.Lock();
	for (i = 0; i < COMM_TRUSTIP_MAX; i++)
	{
		if (m_TrustIP[i] == uIP)
			break;
	}
	if (i >= COMM_TRUSTIP_MAX)
	{
		for (i = 0; i < COMM_TRUSTIP_MAX; i++)
		{
			if (m_TrustIP[i] == 0)
			{
				m_TrustIP[i] = uIP;
				break;
			}
		}
	}
	m_Lock.UnLock();
}

void TComm::DelTrustIP(const char *ip)
{
	if (ip == NULL)
		return;

	unsigned long uIP;

	uIP = (unsigned long)inet_addr(ip);
	if (uIP == 0)
		return;

	m_Lock.Lock();
	for (int i = 0; i < COMM_TRUSTIP_MAX; i++)
	{
		if (m_TrustIP[i] == uIP)
		{
			m_TrustIP[i] = 0;
			break;
		}
	}
	m_Lock.UnLock();
}

int TComm::AllowIP(const char *ip)
{
	if (ip == NULL)
		return COMM_ERR_NOTRUSTIP;

	unsigned long uIP;
	int i;

	uIP = (unsigned long)inet_addr(ip);
	if (uIP == 0)
		return COMM_ERR_NOTRUSTIP;

	m_Lock.Lock();
	for (i = 0; i < COMM_TRUSTIP_MAX; i++)
	{
		if (m_TrustIP[i] == uIP)
		{
			break;
		}
	}
	m_Lock.UnLock();

	if (i < 0 || i >= COMM_TRUSTIP_MAX)
		return COMM_ERR_NOTRUSTIP;
	
	return COMM_ERR_NONE;
}

int TComm::ExistID(unsigned long uID) 
{
	if (uID == 0)
		return COMM_ERR_NOMODULE;

	if (m_ModuleID == uID)
		return COMM_ERR_NONE;

	int i;

	m_Lock.Lock();
	for (i = 0; i < COMM_LINK_MAX; i++)
	{
		if (m_Link[i].m_SocketID != INVALID_SOCKET)
		{
			if (m_Link[i].m_ModuleID == uID)
			{
				break;
			}
		}
	}
	m_Lock.UnLock();

	if (i < 0 || i >= COMM_LINK_MAX)
		return COMM_ERR_NOMODULE;

	return COMM_ERR_NONE;
}

void TComm::OnReceive(const TPacket *lpPacket)
{
	TPacket packet;

	if (lpPacket == NULL)
	{
		return;
	}

	packet = *lpPacket;

	return;
}

void TComm::OnConnect(const TPacket *lpPacket)
{
	TPacket packet;

	if (lpPacket == NULL)
		return;

	packet = *lpPacket;

	return;
}

void TComm::OnClose(const TPacket *lpPacket)
{
	TPacket packet;

	if (lpPacket == NULL)
		return;

	packet = *lpPacket;

	return;
}