server.cpp

一个用vc开发的Sgip通讯程序

	strcpy(szMobile,szUserNumber);
	strcpy(szFeeNum,szChargeNumber);


		pstServerNodeBuf =(SServerNodeBuffer *)m_stServerNodeBuf.GetAt(index);
		if ( iMsgType == 3) //免提短信
		    m_stPdu.SubmitSingle(szMobile,szMessageContent,szFeeNum,szSender,szServiceType,szLinkID,
			cFeeType,szFeeValue,szGivenValue,szScheduleTime,cMessageCoding,cMOrelatetoMTFlag,iStatusRpt,iPriority,1,0,m_nExpireSpan);
		else
			m_stPdu.SubmitSingle(szMobile,szMessageContent,szFeeNum,szSender,szServiceType,szLinkID,
			cFeeType,szFeeValue,szGivenValue,szScheduleTime,cMessageCoding,cMOrelatetoMTFlag,iStatusRpt,iPriority,1,0,m_nExpireSpan);
//---------
/*
void CSGipPdu::SubmitSingle(const char* szUserNumber,const char* szMessageContent,const char* szChargeNumber,
							const char* szSender,const char* szServiceType,char cFeeType,
							const char *szFeeValue,const char* szGivenValue,
							const char* szScheduleTime,char cMessageCoding,char cMOrelatetoMTFlag,
							int iReportFlag,int iPriority,int iTP_pid,int iTP_udhi,int iExpireTimeSpan)
							*/
//----------	
		sprintf(szMsg,"[短信]接收人:%s发送人:%s付费人:%s信息:%s",szMobile,szSender,szFeeNum,szMessageContent);
		LogMsg(szMsg);
		if(SendPacket(&pstServerNodeBuf->m_stSocket,m_stPdu.GetBuffer(),m_stPdu.GetLen()) == m_stPdu.GetLen())
		{
			AddToTaskList(pstServerNodeBuf,m_stPdu.GetBuffer(),m_stPdu.GetLen(),m_stPdu.m_iSequenceNumber1,m_stPdu.m_iSequenceNumber2,m_stPdu.m_iSequenceNumber3);
			return m_stPdu.GetLen();
		}
	}

	return -1;
}

int CServerThread::GetMinUnHandleServerIndex()
{
	SServerNodeBuffer *pstServerNodeBuf;
	int nMinUnHandleNumber;
	int nIndex = -1;

	for(int i=0;i<m_stServerNodeBuf.GetSize();i++)
	{
		pstServerNodeBuf =(SServerNodeBuffer *)m_stServerNodeBuf.GetAt(i);
		if(nIndex == -1)
		{
			if(pstServerNodeBuf->m_bConnected)
			{
				nMinUnHandleNumber = pstServerNodeBuf->m_nUnHandleNumber;
				nIndex = i;
			}
		}
		else
		{
			if(pstServerNodeBuf->m_bConnected && pstServerNodeBuf->m_nUnHandleNumber < nMinUnHandleNumber)
			{
				nMinUnHandleNumber = pstServerNodeBuf->m_nUnHandleNumber;
				nIndex = i;
			}
		}
	}

	return nIndex;
}

BOOL CServerThread::IsConnected()
{
	SServerNodeBuffer *pstServerNodeBuf;

	for(int i=0;i<m_stServerNodeBuf.GetSize();i++)
	{
		pstServerNodeBuf =(SServerNodeBuffer *)m_stServerNodeBuf.GetAt(i);
		if(pstServerNodeBuf->m_bConnected)
			return true;
	}

	return false;
}

BOOL CServerThread::IsAllConnected()
{
	SServerNodeBuffer *pstServerNodeBuf;

	for(int i=0;i<m_stServerNodeBuf.GetSize();i++)
	{
		pstServerNodeBuf =(SServerNodeBuffer *)m_stServerNodeBuf.GetAt(i);
		if(!pstServerNodeBuf->m_bConnected)
			return false;
	}

	return true;
}

void CServerThread::AddToMoniterList(CTcpAsyncSocket* pstClientSocket)
{
	m_stMoniterNodeBuf.Add(pstClientSocket);
}

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::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_bConnected)
		{ //检查是否IDLE已经超过规定时间
			lDiffTime = abs(difftime(tCurTime,pstServerNodeBuf->m_tServerActiveTime)) - m_nMaxIdleTimes;
			if(lDiffTime >0)
			{
				LogMsg("超过最大空闲时间，重连");
				g_bConnect=false;
				SendSGipUnBind(pstServerNodeBuf);
				if(ConnectServer(pstServerNodeBuf))
					SendSGipBind(pstServerNodeBuf);
				time(&pstServerNodeBuf->m_tServerActiveTime);
				pstServerNodeBuf->m_bConnected	= false;
			}
		}
		else
		{
			if(!g_bConnect)
			{
				lDiffTime = abs(difftime(tCurTime,pstServerNodeBuf->m_tConnectServerTime)) - 10;
				if(lDiffTime >0)
				{
					if(ConnectServer(pstServerNodeBuf))
						SendSGipBind(pstServerNodeBuf);
				}
			}
		}
	}
}

int CServerThread::SendPacket(CTcpAsyncSocket *pstSocket,unsigned char* szSendBuf,int nSendLen)
{

	int nBytes,nBytesLeft=nSendLen;
	int nBytesSent=0;
	while (nBytesLeft>0)
	{
		nBytes=pstSocket->Send(szSendBuf+nBytesSent,nBytesLeft);
		if(nBytes >nBytesLeft)
			return 0;
		if(nBytes <=0)
		{
			int nError=GetLastError( );
			CleanUnConnectSocket(pstSocket);
			return -1;
		}
		nBytesSent+=nBytes;
		nBytesLeft-=nBytes;
	}

	return nBytesSent;
}

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

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

	if(m_stTask[nTaskID].m_nResendTimes < m_nMaxRetryTimes)
	{
		if(!bConnectServer)
		{
			bConnectServer = true;
			time(&tOldTime);
			sprintf(szMsg,"任务%d第%d次超时,进行重连.",nTaskID,m_stTask[nTaskID].m_nResendTimes);
			LogMsg(szMsg);
			LogFile(szMsg);
			SendSGipUnBind(m_stTask[nTaskID].m_pstServerNode);
			if(ConnectServer(m_stTask[nTaskID].m_pstServerNode))
			{
				SendSGipBind(m_stTask[nTaskID].m_pstServerNode);
			}
		}
		else
		{
			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);
		LogFile(szMsg);
		
		m_stTask[nTaskID].m_pstServerNode->m_nUnHandleNumber --;
		m_stTask[nTaskID].m_bState = S_IDLE;
		ProcessIdle();
	}
	return;
}

void CServerThread::WriteSubmitLog(int nTaskID,char cResult)
{
	SSGipSubmitSingle stSubmit; 
	CString strSql;
	CSqlExec sql(&m_database);
	CSGipPdu stPdu;
	char szTableName[200],szMsg[200];
	int nCommandID,nSequence1,nSequence2,nSequence3;

	stPdu.GetHeaderInfo((const char*)m_stTask[nTaskID].m_baPduBuf.GetData(),nCommandID,nSequence1,nSequence2,nSequence3);
	memset(&stSubmit,'\0',sizeof(SSGipSubmitSingle));
	memcpy(&stSubmit,m_stTask[nTaskID].m_baPduBuf.GetData() + PDU_HEADER_SIZE,m_stTask[nTaskID].m_baPduBuf.GetSize() -PDU_HEADER_SIZE);

	strcpy(szTableName,m_szSubmitLogTableName);
	if(m_bSubmitLogToEveryDayTable)
	{
		CTime stCurTime;
		CString		strMDay;

		stCurTime = CTime::GetCurrentTime();
		strcat(szTableName,"_");
		strMDay.Format("%02d", stCurTime.GetDay());
		strcat(szTableName,strMDay.Right(2));
	}
	strSql.Format("insert into %s(mobile,message,sender) \
			values('%s','%s','%s')",szTableName,stSubmit.szUserNumber,stSubmit.szMessageContent,stSubmit.szChargeNumber);
	sql.SetSqlStatement(strSql);
	if(!sql.Open())
	{
		sprintf(szMsg,"写%s表错误.",szTableName);
		LogMsg(szMsg);
		LogFile(strSql);
	}
}

BOOL CServerThread::CleanUnConnectSocket(CTcpAsyncSocket *pstSocket)
{
	SServerNodeBuffer *pstServerNodeBuf;
	int i;

	for(i=0;i<m_stServerNodeBuf.GetSize();i++)
	{
		pstServerNodeBuf =(SServerNodeBuffer *)m_stServerNodeBuf.GetAt(i);
		if(&pstServerNodeBuf->m_stSocket == pstSocket)
		{
			if(ConnectServer(pstServerNodeBuf))
				SendSGipBind(pstServerNodeBuf);
			return true;
		}
	}
	
	return 0;
}

void CServerThread::AddToTaskList(SServerNodeBuffer *pstServerNodeBuf,unsigned char* szBuf,int nLen,int nSequence1,int nSequence2,int nSequence3,int nMsgType)
{
	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;
	m_stTask[m_nTaskTail].m_iSequenceNumber1 = nSequence1;
	m_stTask[m_nTaskTail].m_iSequenceNumber2 = nSequence2;
	m_stTask[m_nTaskTail].m_iSequenceNumber3 = nSequence3;
	m_stTask[m_nTaskTail].m_bMsgType=nMsgType;


	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::HandleServerMsg(SServerNodeBuffer *pstNode,char *szRecvBuf,int nLen)
{
	CSGipPdu stPdu;
	int nCommandID,nSequence1,nSequence2,nSequence3;
	char szErrorMsg[200];
	char cResult;
	int  nTaskID;

	time(&pstNode->m_tServerActiveTime);
	stPdu.GetHeaderInfo(szRecvBuf,nCommandID,nSequence1,nSequence2,nSequence3);
	switch(nCommandID)
	{
	case SGIP_BIND_RESP: //bind rep
		cResult = stPdu.GetBindRespResult(szRecvBuf);
		GetStatusMessage(cResult,szErrorMsg);
		if(cResult != SGIP_OK)
		{
			LogMsg(szErrorMsg);
			LogFile(szErrorMsg);
		}
		if (cResult == SGIP_OK)
		{
			pstNode->m_bConnected = true;
			g_bConnect=true;
			LogMsg("Bind成功!");
		}
		else
		{
			LogMsg("Bind失败!");
			pstNode->m_bConnected = false;
		}
		break;

	case SGIP_SUBMIT_RESP:
		nTaskID=SeekTask(nSequence1,nSequence2,nSequence3);
		cResult = stPdu.GetBindRespResult(szRecvBuf);
		GetStatusMessage(cResult,szErrorMsg);
		if(cResult != SGIP_OK)
		{
			LogMsg(szErrorMsg);
			LogFile(szErrorMsg);
		}
		if(nTaskID == -1)
		{
			LogMsg("收到的包无效或已经过时!");
			LogFile("收到的包无效或已经过时!");
			break;
		}
		if (cResult==SGIP_OK)
			LogMsg("发送短信成功!");
		else
			LogMsg("发送短信失败!");
		m_stTask[nTaskID].m_pstServerNode->m_nUnHandleNumber --;
		m_stTask[nTaskID].m_bState = S_IDLE;
		ProcessIdle();
		break;
	case SGIP_UNBIND:
		SendSGipUnBindResp(pstNode,szRecvBuf);
		LogMsg("服务器UNBIND包.");
		//进行重连
		if(ConnectServer(pstNode))
			SendSGipBind(pstNode);
		break;
	case SGIP_UNBIND_RESP:
		LogMsg("服务器UNBIND_RESP包.");
		break;
	default:
		LogMsg("未知包。");
		break;
	}
}

void CServerThread::QueryRecord()
{
	CSqlExec		sql_sm(&m_database),sql_delete(&m_database),sql_Log(&m_database);
	DWORD			tCurTime;
	static DWORD	tLastTime=GetTickCount();
	double			lDiffTime;
	CString			strID,strMessage,strSender,strMobile,strShortMsg,strSql,strFlash;
	CString			strServiceType,strFeeValue,strGivenValue,strMSQL,strMsg,strMsgType,strLinkID;
	char			cFeeType;
	int				nRet,nMOtoMTFlag,iCount,iFeeValue;
	CString			strChargeNumber,strReBack;
	int 			iStatusRpt=0,iDataEnCode=15;
	BOOL            bFeedBack=false;
	
	    //没有可用的连接
	if(GetMinUnHandleServerIndex()<0)	return;

	tCurTime=GetTickCount();
	lDiffTime = tCurTime-tLastTime-m_nQuerRecordTimes;
	
	if (lDiffTime < 0)
		return;
	tLastTime = tCurTime;
//	strSql.Format("select top %d * from %s order by id desc ",m_nContinueSendCount,m_szTableName); //modify by ycf 20020825
	strSql.Format("select top %d * from %s order by priority asc,id ",m_nContinueSendCount,m_szTableName); //modify by ycf 20020825
	sql_sm.SetSqlStatement(strSql);
	if(!sql_sm.Open())
	{
		strMessage.Format("执行SQL语句发生错误:%s",strSql);
		LogMsg(strMessage);
		LogFile(strMessage);
		m_bConnectToDB=FALSE;
		ACTIVE_FLAG = false;
		return;
	}
    
	//写程序正常标志
	if (ACTIVE_FLAG == true)
	    WriteActiveFlag();

	sql_sm.FetchAllData();
	int nRecordCount;
	nRecordCount = sql_sm.RowCount();

    //modify by ycf20020825
	for(int row=0;row <= nRecordCount-1;row++)
	{
		strID       = TrimStr(sql_sm.GetFieldValue(row,"id"));
		strMessage  = TrimStr(sql_sm.GetFieldValue(row,"message"));
		strSender   = TrimStr(sql_sm.GetFieldValue(row,"sender"));
		strMobile   = TrimStr(sql_sm.GetFieldValue(row,"mobile"));
		strServiceType  = TrimStr(sql_sm.GetFieldValue(row,"servicetype"));
		strLinkID  = TrimStr(sql_sm.GetFieldValue(row,"linkid"));
		cFeeType = atoi(sql_sm.GetFieldValue(row,"feetype"));
		strFeeValue = TrimStr(sql_sm.GetFieldValue(row,"feevalue"));
		strGivenValue = TrimStr(sql_sm.GetFieldValue(row,"givenvalue"));
		nMOtoMTFlag = atoi(sql_sm.GetFieldValue(row,"motomtflag"));
		strMsgType= TrimStr(sql_sm.GetFieldValue(row,"MsgType"));
		strChargeNumber= TrimStr(sql_sm.GetFieldValue(row,"FeeNum"));
		iStatusRpt  =atoi(sql_sm.GetFieldValue(row,"statusrpt"));
		iDataEnCode =atoi(sql_sm.GetFieldValue(row,"DataEnCode")); 


		if(m_bCustomSpNumber)
		{
			if(strSender.Find(m_szSPNumber))//自定义的发送人没有以特服号开头
			{