deliverthread.cpp

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

// DeliverThread.cpp : implementation file
//

#include "stdafx.h"
#include "cmpp.h"
#include <process.h>
#include "DeliverThread.h"
#include ".\\odbcapi\\sqlExec.h"
#include "mailcoder.h"
#ifdef _DEBUG
#define new DEBUG_NEW
#undef THIS_FILE
static char THIS_FILE[] = __FILE__;
#endif

/////////////////////////////////////////////////////////////////////////////
// CDeliverThread
#define LINKID_LENGTH 8
extern char CMPP_INI_FILE[100];           //配置信息
extern CArray <SServerNodeBuffer*,SServerNodeBuffer*>  m_stServerNodeBuf;
extern SESSION_ITEM	m_stTask[MAX_TASK_SIZE];
extern int				m_nTaskHead;
extern int				m_nTaskTail;
extern char WEIMA[100];
extern char WEIMANEW[100];

IMPLEMENT_DYNCREATE(CDeliverThread, CWinThread)

CDeliverThread::CDeliverThread()
{
	m_bAutoDelete = false;
	m_astCmdTab = NULL;
	m_pSection=NULL;
	srand(time(NULL));

    m_bConnectDeliverDB = FALSE;

}

CDeliverThread::~CDeliverThread()
{
	if( m_astCmdTab)
	{
		delete [] m_astCmdTab ;
		m_astCmdTab = NULL;
	}
	
	if(m_pSection != NULL)	
	{
		delete[] m_pSection;
		m_pSection = NULL;
	}
}

void AcceptMoniterDeliverProc(void* pParam)
{
	CDeliverThread* pstThread=(CDeliverThread*)pParam;
	SOCKADDR saClient;
	CTcpAsyncSocket* pstAcceptSocket;

	try 
	{
		pstAcceptSocket = new CTcpAsyncSocket();
		if(!pstThread->m_stMoniterSocket.Accept(*pstAcceptSocket,&saClient))
		{
			delete pstAcceptSocket;
			_endthread();
		}
		pstThread->AddToMoniterList(pstAcceptSocket);
		_beginthread(AcceptMoniterDeliverProc,0,pParam);		//get ready for another connection
	}
	catch(const char* e)
	{
		printf("%s\n",e);
		delete pstAcceptSocket;
	}
	_endthread();
}

void RunDeliverProc(void* pParam)
{
	CDeliverThread* pstThread=(CDeliverThread*)pParam;

	pstThread->Run();
	_endthread();
}

BOOL CDeliverThread::InitInstance()
{
	// TODO:  perform and per-thread initialization here
	BOOL	bTrue	= true;
	LoadIni();
	while(bTrue && !g_bDeliverThreadExit)
	{
		if(ConnectDB())
		{
			bTrue	= false;
		}
		else
		{
			Sleep(500);
		}
	}

	LoadCmdTab();
	_beginthread(RunDeliverProc,0,this);

	return TRUE;
}

int CDeliverThread::ExitInstance()
{
	// TODO:  perform any per-thread cleanup here
	return CWinThread::ExitInstance();
}

BEGIN_MESSAGE_MAP(CDeliverThread, CWinThread)
	//{{AFX_MSG_MAP(CDeliverThread)
		// NOTE - the ClassWizard will add and remove mapping macros here.
	//}}AFX_MSG_MAP
END_MESSAGE_MAP()

/////////////////////////////////////////////////////////////////////////////
// CDeliverThread message handlers

void CDeliverThread::LoadIni()
{
	CString  strField;
	char szServiceType[20];

	m_nMoniterPort = GetPrivateProfileInt("系统","MO监控端口",6998,CMPP_INI_FILE);
	GetPrivateProfileString("系统","数据源","DSN=sm;UID=sa;PWD=",m_szDSN,100,CMPP_INI_FILE);
	m_nMaxIdleTimes = GetPrivateProfileInt("系统","MO最大空闲时间",360,CMPP_INI_FILE);
	GetPrivateProfileString("系统","MO指令错误日志表","gr_deliver_error",m_szErrorTable,50,CMPP_INI_FILE);
	GetPrivateProfileString("系统","指令错误帮助信息","",m_szHelp,140,CMPP_INI_FILE);

	GetPrivateProfileString("系统","MO接收日志表名","gr_deliver_log",m_szDeliverLogTableName,100,CMPP_INI_FILE);
	GetPrivateProfileString("系统","MT发送表名","gr_sendmessage",m_szTableName,50,CMPP_INI_FILE);
	GetPrivateProfileString("系统","企业代码","67188",m_szCorpId,21,CMPP_INI_FILE);

	m_bDeliverToEveryDayTable=GetPrivateProfileInt("系统","MO是否每天一个日志表记录",0,CMPP_INI_FILE);

	m_deffee.iFeeType=GetPrivateProfileInt("系统","错误指令计费类别",2,CMPP_INI_FILE);
	m_deffee.iMotomtFlag=GetPrivateProfileInt("系统","错误指令MOTOMTFLAG",0,CMPP_INI_FILE);
	m_deffee.iFeeValue=GetPrivateProfileInt("系统","错误指令信息费",10,CMPP_INI_FILE);
	m_deffee.iGivenValue=GetPrivateProfileInt("系统","错误指令赠话费",0,CMPP_INI_FILE);
	GetPrivateProfileString("系统","错误指令业务代码","SHZX",szServiceType,20,CMPP_INI_FILE);
	m_deffee.strServiceType.Format("%s",szServiceType);

 	GetPrivateProfileString("系统","SP接入号码","8788",m_szSPNumber,21,CMPP_INI_FILE);
	m_bReport     = GetPrivateProfileInt("系统","开启错误状态报告",0,CMPP_INI_FILE);

	m_bMonth = GetPrivateProfileInt("系统","开启包月功能",0,CMPP_INI_FILE);
    GetPrivateProfileString("系统","包月接收表","sm_Game_Deliver",m_szMonthTable,50,CMPP_INI_FILE);

	m_bChat     = GetPrivateProfileInt("系统","开启聊天功能",0,CMPP_INI_FILE);
    m_nChatLen  = GetPrivateProfileInt("系统","会员号长度",4,CMPP_INI_FILE);
    m_iRptMsgID = GetPrivateProfileInt("系统","RptMsgID",0,CMPP_INI_FILE);
	
	GetPrivateProfileString("系统","聊天接收表","gr_deliver_matrix",m_szChatTable,50,CMPP_INI_FILE);
	g_bDeliverThreadExit=false;
	m_stPdu.InitData(m_szSPNumber,m_szCorpId);
	iFirst = 0;
}

BOOL CDeliverThread::ConnectDB()
{
	try
	{
		m_database.OpenEx(m_szDSN,CDatabase::noOdbcDialog);
	}
    catch( CDBException* e )
	{
		LogMsg(e->m_strError);
		LogMsg("连接MO数据库失败!");
		m_bConnectDeliverDB = FALSE;
		return FALSE;
	}

    m_bConnectDeliverDB = TRUE;
	LogMsg("连接MO数据库成功1!");
	
	return TRUE;

}

void CDeliverThread::LoadCmdTab()
{
    CSqlExec sql_get(&m_database) ;
    CString  strSql,strMsg;

	strSql.Format("SELECT * FROM gr_Cmd_MoTab ORDER BY OrdIndex, SenderType DESC, MatchType DESC ");
	sql_get.SetSqlStatement( strSql );
	if( !sql_get.Open())
	{
		strMsg.Format("执行SQL语句出错：%s",strSql);
		LogMsg(strMsg);
		return;
	}
	sql_get.FetchAllData();

	m_nNumOfGameCmd = sql_get.RowCount();
	if(m_astCmdTab != NULL)
	{
		delete [] m_astCmdTab;
		m_astCmdTab = NULL;
	}

	m_astCmdTab = new SCmdTable[m_nNumOfGameCmd];

	for(int row=0; row<m_nNumOfGameCmd;row++) 
	{
		m_astCmdTab[row].strCmd=TrimStr(sql_get.GetFieldValue(row,"Cmd"));
		m_astCmdTab[row].iMatchType=atoi(sql_get.GetFieldValue(row,"MatchType"));
		m_astCmdTab[row].iMoTabNum=atoi(sql_get.GetFieldValue(row,"MoTabNum"));
		m_astCmdTab[row].strCmd.MakeUpper();
		m_astCmdTab[row].strMoTab=TrimStr(sql_get.GetFieldValue(row,"MoTabName"));
		m_astCmdTab[row].strSender=TrimStr(sql_get.GetFieldValue(row,"Sender"));
		m_astCmdTab[row].iSenderType=atoi(sql_get.GetFieldValue(row,"SenderType"));
	}

}

void CDeliverThread::LogMsg(const char *szMsg)
{
	CTcpAsyncSocket *pstMoniterSocket;
	CString strMsg;
	CTime stTime = CTime::GetCurrentTime();

	printf("[%d-%d %02d:%02d:%02d]%s\r\n",stTime.GetMonth(),stTime.GetDay(),stTime.GetHour(),stTime.GetMinute(),stTime.GetSecond(),szMsg);
	strMsg.Format("[%d-%d %02d:%02d:%02d]%s\r\n",stTime.GetMonth(),stTime.GetDay(),stTime.GetHour(),stTime.GetMinute(),stTime.GetSecond(),szMsg);
	for(int i=0;i<m_stMoniterNodeBuf.GetSize();i++)
	{
		pstMoniterSocket =(CTcpAsyncSocket *)m_stMoniterNodeBuf.GetAt(i);
		if(SendPacket(pstMoniterSocket,(unsigned char*)strMsg.LockBuffer(),strMsg.GetLength()) == -1)
		{	
			pstMoniterSocket->Close();
			delete m_stMoniterNodeBuf.GetAt(i);
			m_stMoniterNodeBuf.RemoveAt(i);
		}
		strMsg.UnlockBuffer();
	}

}

int CDeliverThread::SendPacket(CTcpAsyncSocket *pstSocket, unsigned char *szSendBuf, int nSendLen)
{
	int nBytes,nBytesLeft=nSendLen;
	int nBytesSent=0;
	if(pstSocket->m_hSocket)	
	{
		while (nBytesLeft>0)
		{
			nBytes=pstSocket->Send(szSendBuf+nBytesSent,nBytesLeft);
			if(nBytes >nBytesLeft)
				return 0;
			if(nBytes <=0)
			{
				int nError=GetLastError( );
				pstSocket->m_hSocket=INVALID_SOCKET;
				return -1;
			}
			nBytesSent+=nBytes;
			nBytesLeft-=nBytes;
		}
	}
	return nBytesSent;

}

int CDeliverThread::Run() 
{
	SServerNodeBuffer *pstDeliverNodeBuf;
	char szRecvBuf[MAX_PACKET_SIZE +1];
	int  nRecvLen,k;

	InitSocket();
    while(!g_bDeliverThreadExit)
	{
		
		if (m_bConnectDeliverDB)
		{
			for(k=0;k<m_stServerNodeBuf.GetSize();k++)
			{
					pstDeliverNodeBuf =(SServerNodeBuffer *)m_stServerNodeBuf.GetAt(k);
					memset(szRecvBuf,'\0',MAX_PACKET_SIZE +1);
					nRecvLen = ReceivePdu(&pstDeliverNodeBuf->m_stSocket,szRecvBuf,MAX_PACKET_SIZE);
					if(nRecvLen >0)
					{
						szRecvBuf[nRecvLen] ='\0';
						HandleDeliverMsg(pstDeliverNodeBuf,szRecvBuf,nRecvLen);
					}
					else
					{
						if(nRecvLen < 0)
							pstDeliverNodeBuf->m_iConnected = CMPP_STATUS_NOT_CONNECT;
					}

			}
			Sleep(5);
		}
		else
		{
			ConnectDB();
			Sleep(1000);
		}
	}

    m_database.Close();
	return 0;
}

BOOL CDeliverThread::InitSocket()
{
	SServerNodeBuffer *pstNode;

	m_stMoniterSocket.Close();
	m_stMoniterSocket.Create(m_nMoniterPort);
	m_stMoniterSocket.SetNonBlockingMode(false);
	m_stMoniterSocket.Listen();
	_beginthread(AcceptMoniterDeliverProc,0,this);

	//start connect to SMG
	for(int i=0;i<m_stServerNodeBuf.GetSize();i++)
	{
		pstNode =(SServerNodeBuffer *)m_stServerNodeBuf.GetAt(i);
		if(ConnectServer(pstNode))
			SendCmppConnect(pstNode);
	}

	return TRUE;

}

BOOL CDeliverThread::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;

}

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

}

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

}

void CDeliverThread::HandleDeliverMsg(SServerNodeBuffer *pstDeliverNode, char *szRecvBuf, int nLen)
{
	CCmppPdu stPdu;
	int nCommandID,nSequence;
	int iResult;
	char szErrorMsg[200];
	int  nTaskID;
			CString strTemp;
	
	CString strMsgID = "";
	time(&pstDeliverNode->m_tServerActiveTime);
	stPdu.GetHeaderInfo(szRecvBuf,nCommandID,nSequence);
	switch(nCommandID)
	{
	case CMPP_CONNECT_RESP:
		iResult = stPdu.GetConnectResult(szRecvBuf);
		GetStatusMessage(nCommandID,iResult,szErrorMsg);
		if (iResult == 0)
		{
			pstDeliverNode->m_iConnected = CMPP_STATUS_CONNECTED;
			LogMsg(szErrorMsg);
		}
		else
		{
			LogMsg(szErrorMsg);
			pstDeliverNode->m_iConnected = CMPP_STATUS_NOT_CONNECT;
			Sleep(2000);
		}
		break;
	case CMPP_DELIVER:
		HandleDeliverPdu(pstDeliverNode,szRecvBuf);
		break;
	case CMPP_SUBMIT_RESP:
//			strTemp.Format("应答%d",nSequence);
//			LogMsg(strTemp);
		
		nTaskID=SeekTask(nSequence);

		iResult = stPdu.GetSubmitRespResult(szRecvBuf,strMsgID);
		strTemp.Format("SUBMIT应答:%d",iResult);
		LogMsg(strTemp);

		
		GetStatusMessage(nCommandID,iResult,szErrorMsg);
		if(m_iRptMsgID)
			UpdateMsgID(nTaskID,strMsgID);
		if(iResult != 0)
		{
			LogMsg(szErrorMsg);
			LogFile(szErrorMsg);
		}
		if(nTaskID == -1)
		{
			LogMsg("收到的包无效或已经过时!");
			LogFile("收到的包无效或已经过时!");
			break;
		}
		if (iResult==0)
			LogMsg("发送短信成功!");
		else
		{
			LogMsg("发送短信失败!");
			SCmppSubmit	*pstSubmit;
			pstSubmit =(SCmppSubmit*)(m_stTask[nTaskID].m_baPduBuf.GetData() + PDU_HEADER_SIZE);
			CString strMobile = pstSubmit->szDest_terminal_Id;
			LogMsg(pstSubmit->szDest_terminal_Id );
			WriteFaildLog(strMobile,iResult);
			//HandleSubmitRespPdu(pstDeliverNode,szRecvBuf);
		}
		m_stTask[nTaskID].m_pstServerNode->m_nUnHandleNumber --;
		m_stTask[nTaskID].m_bState = S_IDLE;

		ProcessIdle();
		break;
	case CMPP_TERMINATE:
		SendCmppTerminateResp(pstDeliverNode,szRecvBuf);
		LogMsg("服务器TERMINATE包.");
		//进行重连
		if(ConnectServer(pstDeliverNode))
			SendCmppConnect(pstDeliverNode);
		break;
	case CMPP_TERMINATE_RESP:
		LogMsg("收到终止连接应答!");
		pstDeliverNode->m_stSocket.Close();
		break;
	case CMPP_ACTIVE_TEST_RESP:
		pstDeliverNode->m_iActiveTimes = 0;
		LogMsg("收到激活测试应答!");
		break;
	default:
		LogMsg("未知包。");
		break;
	}

}


int CDeliverThread::SendCmppDeliverResp(SServerNodeBuffer *pstNodeBuf, const char *szRecvBuf, const char * szMsgId,int result)
{
	CCmppPdu stPdu;

	stPdu.DeliverResp(szRecvBuf,szMsgId,result);
	return SendPacket(&pstNodeBuf->m_stSocket,stPdu.GetBuffer(),stPdu.GetLen());

}

BOOL CDeliverThread::HandleDeliverPdu(SServerNodeBuffer *pstNodeBuf, const char *szRecvBuf)
{
	char			szTrace[300];
	char			szShortMsg[200];
	char			szLinkID[LINKID_LENGTH+1];
	CString			strSql,strMoTable,strTmp,strTempMsg,strLinkID;  
	CSqlExec		sql_db(&m_database);
	CString			strSender,strShortMsg;
	CString			strDest,strResult,strTmp1;
	int				nMsgLen;
	SCmppDeliver	*pstDeliver;

	pstDeliver =(SCmppDeliver*)(szRecvBuf + PDU_HEADER_SIZE);
	
	SendCmppDeliverResp(pstNodeBuf,szRecvBuf,(char *)pstDeliver->szMsg_Id,0);

/*	char buffer[100];
	memset(buffer,'\0',100);
	char temp[2];
	memset(temp,'\0',2);

	for(int i=0;i<8;i++)
	{
		wsprintf(temp,"%02x",pstDeliver->szMsg_Id[i]);
		lstrcat(buffer,temp);
	}
	printf(buffer);