managersocket.cpp

CMPP3.0模拟网关

// ManagerSocket.cpp: implementation of the CManagerSocket class.
//
//////////////////////////////////////////////////////////////////////

#include "stdafx.h"
#include "CmppGateway.h"
#include "ManagerSocket.h"
#include <AFXMT.H>

#ifdef _DEBUG
#undef THIS_FILE
static char THIS_FILE[]=__FILE__;
#define new DEBUG_NEW
#endif

#define MAX_BUFFER_SIZE 4096
#define MAX_SHOW_SIZE 20480

//////////////////////////////////////////////////////////////////////
// Construction/Destruction
//////////////////////////////////////////////////////////////////////

CManagerSocket::CManagerSocket()
{
	m_nSequnceID = 0;
	m_nMsgSeqID = 0;
	m_IsExit = FALSE;
	m_sShowMsg = _T("");
	m_ServerSock = NULL;
	m_ClientSock = NULL;
	m_SendThread = NULL;
}

CManagerSocket::~CManagerSocket()
{

}

int CManagerSocket::ReceiveMsg(CClientSocket *sock)
{
	char buffer[MAX_BUFFER_SIZE];
	//接受客户端发送过来的消息，这种处理不够好？？？？？？？？
	int len = sock->Receive(buffer, sizeof(buffer));

	if(len < 1)
		return -1;

	buffer[len] = 0;
	//检查消息的合法性及类型
	return CheckReceiveMsg(sock, buffer, len);

}

int CManagerSocket::CheckReceiveMsg(CClientSocket *sock,const char* buffer, int len)
{
	CCmppMessage msg, get_msg;
	int retcode = 0;
	UINT64 nMsgID = 0;
	CString sPrompt("");

	if(get_msg.DecodeMessage(buffer)<0)
		return -3;

	switch(get_msg.m_nCommandID)
	{
	case CMPP_Connect:
		if (!get_msg.m_ValidMessage) retcode = 1;
		else retcode = 0;
		sPrompt = _T("===收到SP发送的Connect包\r\n");
		msg.Set_Connect_REP(retcode, "ok");
		msg.m_nSequenceID = get_msg.m_nSequenceID;
		SendCmppReply(sock, &msg);
		break;
	case CMPP_Submit:
		if (!get_msg.m_ValidMessage) retcode = 1;
		else retcode = 0;
		sPrompt = _T("===收到SP发送的Sumbit包\r\n");
		if(get_msg.m_nMsgID)
			nMsgID = get_msg.m_nMsgID;
		else
			nMsgID = GetMessageID(m_nMsgSeqID);
		m_nMsgSeqID++;
		msg.m_nSequenceID = get_msg.m_nSequenceID;
		msg.Set_Submit_REP(nMsgID, 0);//retcode);
		SendCmppReply(sock, &msg);
		if(get_msg.m_nRegisteredDelivery == 1)
		{
			CCmppMessage *msg_status = new CCmppMessage();
//			m_nMsgSeqID++;
			CString s;
			s.Format("%s", get_msg.m_sDestTerminalIDs[0]);

			msg_status->m_nSequenceID = m_nSequnceID++;
//			msg_status.Set_Deliver(GetMessageID(m_nMsgSeqID), get_msg.m_sSrcTerminalID, "", 0,
//				0, 15, get_msg.m_sDestTerminalIDs[0], 2, "bz");

			msg_status->Set_DeliverStatus(GetMessageID(m_nMsgSeqID),get_msg.m_sSrcTerminalID,NULL,
				0,0,4,get_msg.m_sDestTerminalIDs[0],60,nMsgID,"DELIVRD",CTime::GetCurrentTime().Format("%y%m%d%H%M"),
				CTime::GetCurrentTime().Format("%y%m%d%H%M"),get_msg.m_sDestTerminalIDs[0],atoi(s.Right(10)));
			AddRxMessageQueue(msg_status);
//			SendCmppCommand(sock, &msg_status);
		}
		break;
	case CMPP_Active_Test:
		if (!get_msg.m_ValidMessage) retcode = 1;
		else retcode = 0;
		sPrompt = _T("===收到SP发送的Active_Test包\r\n");
		msg.Set_Active_Test_REP(retcode);
		msg.m_nSequenceID = get_msg.m_nSequenceID;
		SendCmppReply(sock, &msg);
		break;
	case CMPP_Deliver_REP:
		sPrompt = _T("===收到SP返回的Deliver_Response包\r\n");;
		break;
	default:
		break;
	}
	ShowMessage(sPrompt, buffer, len);
	return retcode;
}

int CManagerSocket::SendCmppReply(CClientSocket *sock,CCmppMessage *lpMessage)
{
	char buffer[MAX_BUFFER_SIZE];
	int retcode = 0;

	memset(buffer, 0, MAX_BUFFER_SIZE);

	int totallen = lpMessage->EncodeMessage(buffer);
	if(totallen < 4) return -3;
	if(sock->Send(buffer, totallen) < 0) return -2;
	ShowMessage(_T("response to sp!\r\n"), buffer, totallen);
	return retcode;
}

UINT64 CManagerSocket::GetMessageID(WORD nSeqID)
{
	UINT64 nMsgId = 0;
	CString sTime;

	CTime time = CTime::GetCurrentTime();
	sTime = time.Format("%m");
	nMsgId += atoi(sTime);
	nMsgId = nMsgId << 5;
	sTime = time.Format("%d");
	nMsgId += atoi(sTime);
	nMsgId = nMsgId << 5;
	sTime = time.Format("%H");
	nMsgId += atoi(sTime);
	nMsgId = nMsgId << 6;
	sTime = time.Format("%M");
	nMsgId += atoi(sTime);
	nMsgId = nMsgId << 6;
	sTime = time.Format("%S");
	nMsgId += atoi(sTime);
	nMsgId = nMsgId << 22;
	nMsgId += 1010;
	nMsgId = nMsgId << 16;
/*
	sTime = time.Format("%m%d%H%M%s");
	nMsgId =atoi(sTime);//*100000;//+1010;
	nMsgId =(nMsgId*100000+1010);
	nMsgId = nMsgId*100000;
*/	return nMsgId + nSeqID;
}

void CManagerSocket::DeliverMsg(const char *Destnationid, const char *service_id, 
							   int TP_pid, int TP_udhi, int msg_fmt, const char *src_term_id, int msg_len, const char *msg)
{
	__int64 nMsgId;
	CCmppMessage cmpp_msg;

	cmpp_msg.m_nSequenceID = m_nSequnceID++;
	nMsgId = GetMessageID(m_nMsgSeqID);
	m_nMsgSeqID++;

//	CString s;
//	s.Format("%s", src_term_id);
//	cmpp_msg.Set_DeliverStatus(nMsgId,Destnationid,"",
//				0,0,4,src_term_id,60,nMsgId + 10,"DELIVRD",CTime::GetCurrentTime().Format("%y%m%d%H%M"),
//				CTime::GetCurrentTime().Format("%y%m%d%H%M"),src_term_id,atoi(s.Right(10)));
	cmpp_msg.Set_Deliver(nMsgId, Destnationid, service_id, TP_pid,
				TP_udhi, msg_fmt, src_term_id, msg_len, msg);
	POSITION n = m_sockQueue.Find(m_ClientSock);
	int sum= m_sockQueue.GetCount();
	POSITION nPos = m_sockQueue.GetHeadPosition();
	int i = 0;
	while(nPos != NULL)
	{
		i++;
		if(n ==nPos)
			printf("I get");
		m_sockQueue.GetNext(nPos);
	}
	SendCmppCommand(m_ClientSock, &cmpp_msg);

}

int CManagerSocket::SendCmppCommand(CClientSocket *sock,CCmppMessage *msg)
{
	char buffer[MAX_BUFFER_SIZE];
	CString sPrompt;
	int retcode = 0;

	switch(msg->m_nCommandID)
	{
	case CMPP_Deliver:
		if(msg->m_nRegisteredDelivery == 0)
			sPrompt = _T("===发送Deliver包给SP\r\n");
		else
			sPrompt = _T("===返回Report包给SP\r\n");
		break;
	case CMPP_Terminate:
		sPrompt = _T("===发送Terminate包给SP\r\n");
		break;
	default:
		sPrompt = _T("===有包发送给SP\r\n");
	}

	memset(buffer, 0 , MAX_BUFFER_SIZE);
	int totallen = msg->EncodeMessage(buffer);
	if(totallen <4)
		return -3;

	ShowMessage(sPrompt, buffer, totallen);
	if(sock->Send(buffer, totallen)<0) retcode = -1;

	return retcode;
}

void CManagerSocket::ShowMessage(const char *buffer1, const char *buffer2, int buffer2len)
{
	CCriticalSection lockMsg;
	CString sOut;

	lockMsg.Lock();
	if(NULL == buffer2)
		m_sShowMsg = buffer1 + m_sShowMsg;
	else
	{
		m_sShowMsg = Convert2AsII(buffer2, buffer2len) + m_sShowMsg;
		m_sShowMsg = buffer1 + m_sShowMsg;
	}

	if(m_sShowMsg.GetLength() >	MAX_SHOW_SIZE)
		m_sShowMsg.Empty();
	::AfxGetMainWnd()->SetDlgItemText(IDC_OUTPUT,m_sShowMsg);
	lockMsg.Unlock();
}

CString CManagerSocket::Convert2AsII(const char *buffer, int len)
{
	CString sRet(""), sTmp;

	if(len<1)
		return "";
	for(int i=1; i<=len; i++)
	{
		sTmp.Format("%02x ", (BYTE)buffer[i-1]);
		sRet += sTmp;
		if(!(i%16))
			sRet += "\r\n";
	}
	sRet += "\r\n";
	return sRet;
}

void CManagerSocket::AcceptClient()
{
	m_ClientSock = new CClientSocket(this);
	//接受客户端的 
	if(!m_ServerSock->Accept(*m_ClientSock))
	{
		delete m_ClientSock;
		m_ClientSock = NULL;
	}
	//加入socket队列中
	m_sockLock.Lock();
	m_sockQueue.AddTail(m_ClientSock);
	m_sockLock.Unlock();
}

void CManagerSocket::StartSocketSrv() 
{
	// TODO: Add your control notification handler code here
	int nPort = ::AfxGetMainWnd()->GetDlgItemInt(IDC_PORT);
	//关闭先前存在的socket
	if(m_ServerSock != NULL)
	{
		m_ServerSock->Close();
		delete m_ServerSock;
		m_ServerSock = NULL;
	}

	//删除列表中所有的客户端socket
	POSITION nPos = m_sockQueue.GetHeadPosition();
	while(nPos != NULL)
	{
		m_ClientSock = (CClientSocket *)m_sockQueue.GetNext(nPos);
		m_ClientSock->Close();
		delete m_ClientSock;
		m_ClientSock = NULL;
	}
	m_sockQueue.RemoveAll();

	//创建新的侦听socket
	m_ServerSock = new CServerSocket(this);
	if(!m_ServerSock->Create(nPort, SOCK_STREAM))
	{
		AfxMessageBox("创建sokcet失败!");
		delete m_ServerSock;
		m_ServerSock = NULL;
		return;
	}

	if(!m_ServerSock->Listen())
	{
		AfxMessageBox("启动Listen失效!");
		delete m_ServerSock;
		m_ServerSock = NULL;
		return;
	}

//	m_SendThread = CreateThread(NULL, 0, (LPTHREAD_START_ROUTINE)MessageSendThread, this, 0, NULL);

}

void CManagerSocket::StopSocketSrv()
{
	CCmppMessage cmpp_msg;

	QueueRemoveAll();
	m_IsExit = TRUE;
	if(m_SendThread) WaitForSingleObject(m_SendThread, INFINITE);
	CloseHandle(m_SendThread);
	m_SendThread = NULL;
	POSITION nPos = m_sockQueue.GetHeadPosition();
	while(nPos != NULL)
	{
		m_ClientSock = (CClientSocket *)m_sockQueue.GetNext(nPos);
//		cmpp_msg.m_nSequenceID = m_nSequnceID++;
//		cmpp_msg.Set_Terminate();
//		SendCmppCommand(m_ClientSock, &cmpp_msg);
//		m_ClientSock->Close();
		delete m_ClientSock;
		m_ClientSock = NULL;
	}
	m_sockQueue.RemoveAll();

	if(m_ServerSock != NULL)
	{
		m_ServerSock->Close();
		delete m_ServerSock;
		m_ServerSock = NULL;
	}
}

void CManagerSocket::AddRxMessageQueue(const CCmppMessage *lpMessage)
{
	m_SendLock.Lock();
	m_SendQueue.AddTail(*lpMessage);
	m_SendLock.Unlock();
	return;
}

bool CManagerSocket::GetRxMessageQueue(CCmppMessage *lpMessage)
{
	bool find = false;
	m_SendLock.Lock();
	if (m_SendQueue.GetCount())
	{
		*lpMessage = m_SendQueue.RemoveHead();
		find = true;
	}
	m_SendLock.Unlock();
	return find;
}

DWORD WINAPI CManagerSocket::MessageSendThread(CManagerSocket *lpSession)
{
	while(!lpSession->m_IsExit)
	{
		CCmppMessage *lpMessage = new CCmppMessage();
		if(lpSession->GetRxMessageQueue(lpMessage))
		{
			lpSession->m_sockLock.Lock();
			CClientSocket *lpClient =(CClientSocket *)lpSession->m_sockQueue.GetTail();
			lpSession->SendCmppCommand(lpSession->m_ClientSock, lpMessage);
			lpSession->m_sockLock.Unlock();

			delete lpMessage;
		}
		else
		{
			delete lpMessage;
			continue;
		}
	}
	return 0;
}

void CManagerSocket::QueueRemoveAll()
{
	m_SendLock.Lock();
	POSITION nPos = m_SendQueue.GetHeadPosition();
	while(nPos != NULL)
	{
		CCmppMessage *lpMessage;
		*lpMessage = m_SendQueue.GetNext(nPos);
		delete lpMessage;
		lpMessage = NULL;
	}
	m_SendQueue.RemoveAll();
	m_SendLock.Unlock();
}