serverthread.cpp

vc开发的一个cmpp(中国移动点对点通讯)通讯程序

	int  iStrLength = strMsg.GetLength();
	char *m_newBuffer = new char [iStrLength];
	LPSTR newTempPtr = m_newBuffer;
	unsigned short halfChinese=0;
	int  iNumOfWords = 0;
	int  iPos=0;
	while( iPos<iStrLength ){
		*newTempPtr = strMsg.GetAt(iPos);
		//大于0xa1的字节为汉字字节
		if((unsigned char)*newTempPtr >= 0xa1)
			halfChinese++;
		if( (halfChinese%2)==0 )//对汉字每两个字节算一个字
		    iNumOfWords++;
		iPos++;
	}
	delete m_newBuffer;

	return iNumOfWords;

}

CString CServerThread::GetLeftTChar(CString strSrc, int iWordsNum, int iStartPos, int &iDestStrLength)
{
//把字母数字空格等单字节的字符当作双字节，一个汉字为双字节，因此这里定义两个字节为一个汉字
	long  lMsgLength = strSrc.GetLength();
	CString NullStr("") ;
	if (!lMsgLength || iStartPos>lMsgLength || iStartPos<0) return NullStr;//当源字串为空或起始位置非法时,返回空串
	char *m_newBuffer = new char [lMsgLength];////////
    LPSTR newTempPtr = m_newBuffer;
	int iCounter;
	char TCharbuf[255];
    //以防止将一半汉字分为两行
	unsigned short halfChinese=0;
	int i=iStartPos;
    iCounter = 0 ;
	while(i<lMsgLength && iCounter<iWordsNum){
		*newTempPtr = strSrc.GetAt(i);
		//大于0xa1的字节为汉字字节
		if((unsigned char)*newTempPtr >= 0xa1)
			halfChinese++;
		TCharbuf[i-iStartPos]=*newTempPtr++;
		if( (halfChinese%2)==0 )//对汉字每两个字节算一个字
            iCounter++;
		i++;
	}
	if (halfChinese%2==1)
		TCharbuf[i-iStartPos-1]='\0';
    else
		TCharbuf[i-iStartPos]='\0';

	CString strResult;
	strResult.Format("%s",TCharbuf);

	delete m_newBuffer;
	iDestStrLength = strResult.GetLength();
	return strResult;

}

void CServerThread::AddToMoniterList(CTcpAsyncSocket *pstClientSocket)
{
	m_stMoniterNodeBuf.Add(pstClientSocket);
}
int CServerThread::SendShortMessage(const char* szUserNumber,const char* szMessageContent,const char* szChargeNumber,
		 const char* szSender,const char* szServiceType,const char *szLinkID,char cFeeType,const char *szFeeValue,const char* szGivenValue,
							const char* szScheduleTime,char cMessageCoding,char cMOrelatetoMTFlag,int iStatusRpt,int iPriority,int iMsgType,int iExpireTimeSpan,int mobiletype,int feenumtype)
{
	SServerNodeBuffer *pstServerNodeBuf;
	char szMobile[200],szFeeNum[200],szMsg[2048];
	CString	strMsgID,strSql;

	if(strlen(szUserNumber)<11)  
		return -1;

	int index = GetMinUnHandleServerIndex();
	
	if(index > -1)
	{
/*		if(szUserNumber[0] =='1')
		{
			strcpy(szMobile,"86");
			strcat(szMobile,szUserNumber);
		}
		else  */
		{
			strcpy(szMobile,szUserNumber);
		}
		
/*		if(strlen(szChargeNumber) !=0)
		{
			if(szChargeNumber[0] =='1')
			{
				strcpy(szFeeNum,"86");
				strcat(szFeeNum,szChargeNumber);
			}  
			else  */
			{
				strcpy(szFeeNum,szChargeNumber);
			}
//		}
//		else
//			strcpy(szFeeNum,"");  

		pstServerNodeBuf =(SServerNodeBuffer *)m_stServerNodeBuf.GetAt(index);
	//	m_stPdu.SubmitSingle(szMobile,szMessageContent,szFeeNum,szSender,szServiceType,szLinkID,
	//	cFeeType,szFeeValue,szGivenValue,szScheduleTime,cMessageCoding,cMOrelatetoMTFlag,iStatusRpt,iPriority,1,0,m_nExpireSpan,mobiletype,feenumtype);
		
		//Modify by lzq ,2005-7-18,主要是修改tp_pid的值
		m_stPdu.SubmitSingle(szMobile,szMessageContent,szFeeNum,szSender,szServiceType,szLinkID,
		cFeeType,szFeeValue,szGivenValue,szScheduleTime,cMessageCoding,cMOrelatetoMTFlag,iStatusRpt,iPriority,0,0,m_nExpireSpan,mobiletype,feenumtype);

		
		sprintf(szMsg,"[短信]接收人:%s发送人:%s付费人:%s信息:%s",szMobile,szSender,szFeeNum,szMessageContent);
		LogMsg(szMsg);
		AddToTaskList(pstServerNodeBuf,m_stPdu.GetBuffer(),m_stPdu.GetLen(),m_stPdu.m_nSequenceNumber);
		if(SendPacket(&pstServerNodeBuf->m_stSocket,m_stPdu.GetBuffer(),m_stPdu.GetLen()) == m_stPdu.GetLen())
		{
			/*
			if(m_nRptMsgFlag)
			{
			
				CString iFeeType = cFeeType;	
				CString strFeeValue = szFeeValue;
				if(iFeeType == 3 && strFeeValue == "300")//说明是包月短信
				{
					LogByFeeMsg(m_stPdu.m_nSequenceNumber,szMobile);
					
				}
			}*/
			return m_stPdu.GetLen();
		}
		else
		{
			Sleep(50);
//			pstServerNodeBuf->m_iConnected = CMPP_STATUS_NOT_CONNECT;
		}
	}

	return -1;
}
void CServerThread::SubmitToSendMsgLog(const char *szMsg,const char *szMobile,const char *szSender,const char *szFeeNum, const char * szServiceType,const char * szLinkID,int cFeeType,const char * szFeeValue,const char * szGivenValue,int nMotoMtFlag)
{
	CSqlExec sql_sendmsg(&m_database);
	CString strSql,strTabName,strMsg;
	CTime ctTime = CTime::GetCurrentTime();
    
	strMsg.Format("%s",szMsg);
	strMsg.Replace("'","''");

    if(!m_bSubmitLogToEveryDayTable)
	{
		strSql.Format("insert into %s(message,mobile,sender,feenum,servicetype,linkid,feetype,feevalue,givenvalue,motomtflag) \
		           values('%s','%s','%s','%s','%s','%s','%d','%s','%s','%d')",
				   m_szSubmitLogTableName,
				   strMsg,
				   szMobile,
				   szSender,
				   szFeeNum,
				   szServiceType,
				   szLinkID,
				   cFeeType,
				   szFeeValue,
				   szGivenValue,
				   nMotoMtFlag);
	}
	else
	{
		strSql.Format("insert into %s_%02d(message,mobile,sender,feenum,servicetype,linkid,feetype,feevalue,givenvalue,motomtflag) \
		           values('%s','%s','%s','%s','%s','%s','%d','%s','%s','%d')",
				   m_szSubmitLogTableName,
				   ctTime.GetDay(),
				   strMsg,szMobile,
				   szSender,
				   szFeeNum,
				   szServiceType,
				   szLinkID,
				   cFeeType,
				   szFeeValue,
				   szGivenValue,
				   nMotoMtFlag);
	}
	sql_sendmsg.SetSqlStatement(strSql);

	if(!sql_sendmsg.Open())
	{
		CString	strMsg;
		strMsg.Format("执行SQL出错：%s",strSql);
		LogMsg(strMsg);
		return;
	}
}

int CServerThread::SendCmppTerminate(SServerNodeBuffer *pstNode)
{
	LogMsg("[MT]发送Terminate包给SMG.");
	m_stPdu.Terminate();
	if(SendPacket(&pstNode->m_stSocket,m_stPdu.GetBuffer(),m_stPdu.GetLen()) >0)
	{   
		return 1;
	}
	return 0;
}

void CServerThread::AddToTaskList(SServerNodeBuffer *pstServerNodeBuf, unsigned char *szBuf, int nLen, int nSequence)
{
	BYTE* pBuf; 

	m_stTask[m_nTaskTail].m_baPduBuf.SetSize(nLen);
	m_stTask[m_nTaskTail].m_bState = S_WAITANS;
	m_stTask[m_nTaskTail].m_pstServerNode = pstServerNodeBuf;
	if(nSequence > 0)
		nSequence--;
	else
		nSequence = 0xffffffff;
	m_stTask[m_nTaskTail].m_iSequenceNumber = nSequence;

	pstServerNodeBuf->m_nUnHandleNumber ++;
	pBuf=m_stTask[m_nTaskTail].m_baPduBuf.GetData(); 
	memcpy(pBuf,szBuf,nLen);

	m_stTask[m_nTaskTail].m_nResendTimes =0;
	time(&m_stTask[m_nTaskTail].m_cTimer);

	m_nTaskTail = (m_nTaskTail +1) % MAX_TASK_SIZE;
}

void CServerThread::HandleTask()
{
	int head = m_nTaskHead;
	time_t			tCurTime;
	double			lDiffTime;

	time(&tCurTime);
	while(head != m_nTaskTail)
	{
		if(m_stTask[head].m_bState != S_IDLE)
		{
			lDiffTime = abs(difftime(tCurTime , m_stTask[head].m_cTimer)) - m_nWaitAnsTimes;
			if (lDiffTime > 0) //超时
				ResendMessage(head);
		}
		head = (head +1) % MAX_TASK_SIZE;
	}

}

void CServerThread::ResendMessage(int nTaskID)
{
	static time_t tOldTime=time(NULL),tCurTime;
	long lDiffTime;
	char szMsg[200];

	time(&tCurTime);
	lDiffTime = abs(difftime(tCurTime,tOldTime));
	if(lDiffTime > m_nMaxIdleTimes)
	{
		time(&tOldTime);
	}

	if(m_stTask[nTaskID].m_nResendTimes < m_nMaxRetryTimes)
	{
		sprintf(szMsg,"任务%d第%d次重做.",nTaskID,m_stTask[nTaskID].m_nResendTimes);
		LogMsg(szMsg);
//		LogFile(szMsg);
		time(&m_stTask[nTaskID].m_cTimer);
		m_stTask[nTaskID].m_nResendTimes ++;
		SendPacket(&m_stTask[nTaskID].m_pstServerNode->m_stSocket,m_stTask[nTaskID].m_baPduBuf.GetData(),m_stTask[nTaskID].m_baPduBuf.GetSize());

	}
	else
	{					
		sprintf(szMsg,"任务%d重做%d次失败.",nTaskID,m_stTask[nTaskID].m_nResendTimes);
		LogMsg(szMsg);
	
		m_stTask[nTaskID].m_pstServerNode->m_nUnHandleNumber --;
		m_stTask[nTaskID].m_bState = S_IDLE;
		ProcessIdle();
		LogFile(szMsg);
	}
}

void CServerThread::ProcessIdle()
{
	while(m_nTaskHead != m_nTaskTail)
	{
		if(m_stTask[m_nTaskHead].m_bState == S_IDLE)
			m_nTaskHead = (m_nTaskHead +1) % MAX_TASK_SIZE;
		else
			break;
	}
}

void CServerThread::HandleActiveTest()
{
	time_t			tCurTime;
	double			lDiffTime;
	SServerNodeBuffer *pstServerNodeBuf;

	time(&tCurTime);
	for(int i=0;i<m_stServerNodeBuf.GetSize();i++)
	{
		pstServerNodeBuf =(SServerNodeBuffer *)m_stServerNodeBuf.GetAt(i);
		if(pstServerNodeBuf->m_iConnected == CMPP_STATUS_CONNECTED)
		{
			if(pstServerNodeBuf->m_iActiveTimes > 2)
			{
				SendCmppTerminate(pstServerNodeBuf);
				pstServerNodeBuf->m_iConnected = CMPP_STATUS_NOT_CONNECT;
				pstServerNodeBuf->m_iActiveTimes = 0;
				return;
			}
			//检查是否IDLE已经超过规定时间
			lDiffTime = abs(difftime(tCurTime,pstServerNodeBuf->m_tServerActiveTime)) - m_nMaxIdleTimes;
			if(lDiffTime >0)
			{
				pstServerNodeBuf->m_iActiveTimes++;
				LogMsg("超过最大空闲时间,激活测试");
				if(!SendCmppActiveTest(pstServerNodeBuf))
				{
					printf("发送激活测试失败\n");
					pstServerNodeBuf->m_iConnected = CMPP_STATUS_NOT_CONNECT;
				}
				else
				{
					printf("发送激活测试成功\n");
				}
				time(&pstServerNodeBuf->m_tServerActiveTime);
			}
		}
		else
		{
			if(pstServerNodeBuf->m_iConnected == CMPP_STATUS_NOT_CONNECT)
			{
				pstServerNodeBuf->m_stSocket.Close();
				Sleep(3000);
				if(ConnectServer(pstServerNodeBuf))
				{
					SendCmppConnect(pstServerNodeBuf);
					Sleep(1000);
				}
			}
			else
			{
				lDiffTime = abs(difftime(tCurTime,pstServerNodeBuf->m_tConnectServerTime)) - 5;
				if(lDiffTime >0)
				{
					pstServerNodeBuf->m_iConnected = CMPP_STATUS_NOT_CONNECT;
				}
			}
		}
		
	}

}

BOOL CServerThread::ConnectServer(SServerNodeBuffer *pstNode)
{
	CString strMsg;

	pstNode->m_stSocket.Close();
	CString str ;
	str=pstNode->m_szLocalIP ;
	if(str.GetLength()==0)
	{
		pstNode->m_stSocket.Create(0,SOCK_STREAM);
	}
	else
	{
		pstNode->m_stSocket.Create(0,SOCK_STREAM,pstNode->m_szLocalIP);
	}

	pstNode->m_iConnected = CMPP_STATUS_NOT_CONNECT;
	pstNode->m_stSocket.SetNonBlockingMode(false);

	if(pstNode->m_stSocket.Connect(pstNode->m_szServerIP,pstNode->m_nServerPort))
	{
		pstNode->m_iConnected = CMPP_STATUS_PENDING;
		strMsg.Format("连接到服务器%s,端口%d成功。",pstNode->m_szServerIP,pstNode->m_nServerPort);
		LogMsg(strMsg);
	}
	else
	{
		int nError = GetLastError();
		strMsg.Format("连接到服务器%s,端口%d失败。",pstNode->m_szServerIP,pstNode->m_nServerPort);
		LogMsg(strMsg);
		time(&pstNode->m_tConnectServerTime);
		time(&pstNode->m_tServerActiveTime);
		return false;
	}

	pstNode->m_stSocket.SetNonBlockingMode(true);
	time(&pstNode->m_tConnectServerTime);
	time(&pstNode->m_tServerActiveTime);

	return true;
}

BOOL CServerThread::SendCmppActiveTest(SServerNodeBuffer *pstNode)
{
	m_stPdu.ActiveTest();
	return SendPacket(&pstNode->m_stSocket,m_stPdu.GetBuffer(),m_stPdu.GetLen());
}

int CServerThread::SendCmppConnect(SServerNodeBuffer *pstNode)
{
	m_stPdu.ConnectSvr(pstNode->m_szUser,pstNode->m_szPassword);
	return SendPacket(&pstNode->m_stSocket,m_stPdu.GetBuffer(),m_stPdu.GetLen());
}