cmppsocket.cpp

CMPP——Gate

/******************************************************************************
  FileName                : CcmppSocket.cpp
  Description             : 短消息发送程序
  Version                 : 1.0
  Date                    : 2003年3月10日
  Author                  : 罗天煦
  Other                   : 目前只支持CMPP
******************************************************************************/

#include <stdlib.h>
#include <time.h>
#include "md5.h"
#include "cmppSocket.h"

CcmppSocket::CcmppSocket()
{
	//	初始化私有变量
	_binitialized = false;
	_bexitting = false;
	_seqid	= 0;
	memset( (void *)_window, 0, sizeof( _window));
	InitializeCriticalSection( &_csec_wnd);
	InitializeCriticalSection( &_csec_snd);
	InitializeCriticalSection( &_csec_recv);
	InitializeCriticalSection( &_csec_seq);
	_hsema_wnd = CreateSemaphore(			//	创建计数信号量
		NULL,
		nCMPP_WINDOW_SIZE,
		nCMPP_WINDOW_SIZE,
		NULL);
	_hevnt_data = CreateEvent(				//	创建提示发送的事件
		NULL,
		false,
		false,
		NULL);
	//	初始化网络
	WSADATA wsaData;
	WSAStartup( MAKEWORD( 2, 2 ), &wsaData );
}

CcmppSocket::~CcmppSocket()
{
	//	发送CMPP_TERMINATE，从服务器注销
	if( _binitialized)
		_exit();
	WSACleanup();
	//	清除临界区等
	DeleteCriticalSection( &_csec_wnd);
	DeleteCriticalSection( &_csec_snd);
	DeleteCriticalSection( &_csec_recv);
	DeleteCriticalSection( &_csec_seq);
	CloseHandle( _hsema_wnd);
	CloseHandle( _hevnt_data);
}

/******************************************************************************
  spid		企业代码		例如：901001	
  passwd	登陆口令		例如：test
  ismg		短信网关的地址，例如：127.0.0.1
  port		短信网关的端口，例如：7890
******************************************************************************/
int CcmppSocket::init( char *spid, char *passwd, char *ismg, unsigned short port)
{
	//	如果接口已经初始化，则先删除接口
	if( _binitialized)
		_exit();
	int err;
	err = _connect( ismg, port);
	if( err != 0)
		return eCMPP_INIT_CONNECT;
	
	err = _login( spid, passwd);
	if( err != 0)
	{
		closesocket( _soc);
		return err;
	}
	//	保存配置，以备后用
	strcpy( _spid,	spid);
	strcpy( _passwd,passwd);
	strcpy( _ismg,	ismg);
	_port = port;

	//	启动发送、接收数据的线程
	_hsend = CreateThread(	//	短信发送线程
		NULL,
		NULL,
		thread_send,
		(LPVOID)this,
		0,
		NULL);
	_hrecv = CreateThread(	//	短信接收线程
		NULL,
		NULL,
		thread_recv,
		(LPVOID)this,
		0,
		NULL);
	_hactv = CreateThread(	//	链路检查
		NULL,
		NULL,
		thread_actv,
		(LPVOID)this,
		0,
		NULL);
	//	初始化成功，设置成功标志
	_binitialized = true;
	//	放弃当前时间片
	Sleep( 0);

	return eCMPP_INIT_OK;
}

/******************************************************************************
  msg				向服务器提交的数据
  dwMilliseconds	在成功的将数据插入数据窗口前等待的时间

  返回值			描述
  ===============	==============================		
  0					成功
  eCMPP_NEED_INIT	接口未初始化
  WAIT_TIMEOUT		操作超时
  WAIT_ABANDONED	工作线程异常退出，可能是网络故障
******************************************************************************/
int	CcmppSocket::Submit( CMPP_SUBMIT &msg, DWORD dwMilliseconds)
{
	CMPP_PACKAGE	pkg;
	CMPP_HEAD		&head = (CMPP_HEAD		&)pkg.head;
	int	err, nsize;

	if(!_binitialized)
		return eCMPP_NEED_INIT;

	pkg.n = 3;							//	发送失败，则重发两次
	pkg.t = time( NULL);				//	立即发送

	nsize = sizeof( CMPP_HEAD) + sizeof( CMPP_SUBMIT);
	//	因为CMPP协议中包的长度是可变的，而我定义的数据结构中
	//	包的长度采用的是最大长度，所以这里需要修正
	nsize = nsize + msg.msglen - sizeof( msg.msgcontent);
	head.size = htonl( nsize);
	head.cmdid= htonl( nCMPP_SUBMIT);
	head.seqid= htonl( _getseqid());

	memcpy( (void *)pkg.data, (void *)&msg, sizeof( msg));
	//	将最后8个字节的保留数据拷贝到适当的位置
	memcpy(
		(void *)(pkg.data + nsize - sizeof( msg.reserve) - sizeof( CMPP_HEAD)),
		(void *)msg.reserve,
		sizeof( msg.reserve));
	//	等候数据发送窗口的空位
	err = WaitForSingleObject( _hsema_wnd, dwMilliseconds);
	//	等待超时或程序异常
	if( err != WAIT_OBJECT_0)
		return err;
	//	申请数据发送窗口的使用权
	EnterCriticalSection( &_csec_wnd);
	for( int i=0; i<nCMPP_WINDOW_SIZE; i++)
	{
		//	找到一个空位
		if( _window[i].head.cmdid == 0)
			break;
	}
	memcpy( (void *)&_window[i], (void *)&pkg, sizeof( pkg));
	LeaveCriticalSection( &_csec_wnd);
	//	唤醒数据发送线程
	PulseEvent( _hevnt_data);

	return 0;
}

int CcmppSocket::_connect( char *ismg, unsigned short port)
{
	int err;
	struct sockaddr_in addr;

	_soc = socket( AF_INET, SOCK_STREAM, 0);

	addr.sin_family = AF_INET;
	addr.sin_port   = htons( port);
	addr.sin_addr.s_addr   = inet_addr( ismg);

	err = connect( _soc, (struct sockaddr *)&addr, sizeof( addr));
	return err;
}

int CcmppSocket::_login( char *spid, char *passwd)
{
	CMPP_PACKAGE	pkg;
	CMPP_CONNECT	&msg = (CMPP_CONNECT	&)pkg.data;
	int err, nsize;

	MD5 ctx;
	char authsrc[50], *pos, timestr[20];

	memset( (void *)&msg, 0, sizeof( msg));

	nsize = sizeof( CMPP_HEAD) + sizeof( CMPP_CONNECT);
	pkg.head.size  = htonl( nsize);
	pkg.head.cmdid = htonl( nCMPP_CONNECT);
	pkg.head.seqid = htonl( _getseqid());

	strcpy( (char *)msg.spid, spid);
	//	计算单向HASH函数的值
	memset( authsrc, 0, sizeof( authsrc));
	strcpy( authsrc, spid);
	pos = authsrc + strlen( spid) + 9;
	strcpy( (char *)pos, passwd);
	pos += strlen( passwd);
	strcpy( pos, _timestamp( timestr ));
	pos += strlen( timestr);

	ctx.update( (unsigned char *)authsrc, (int)(pos - authsrc) );
	ctx.finalize();
	ctx.raw_digest( msg.digest);

	msg.ver = nCMPP_VERSION;
	msg.timestamp = htonl( atol( timestr));
	//	发送身份验证数据
	err = _send( (char *)&pkg, nsize);
	if( err != nsize)
		return eCMPP_INIT_CONNECT;

	//	接收返回的数据包
	CMPP_CONNECT_RESP &msgresp = (CMPP_CONNECT_RESP &)pkg.data;
	nsize = sizeof( CMPP_HEAD) + sizeof( CMPP_CONNECT_RESP);
	
	err = _recv( (char *)&pkg, nsize);
	if( err != nsize )
		return eCMPP_INIT_CONNECT;

	return ntohl( msgresp.status);
}

void CcmppSocket::_exit()
{
	HANDLE	threads[] = { _hsend, _hrecv, _hactv};
	int		i,
			nthreads;
	//	请求工作线程退出
	_bexitting = true;

	Sleep( 1000);
	//	关闭请求，强制结束所有尚未退出的线程
	_bexitting = false;

	nthreads = sizeof( threads) / sizeof( HANDLE);
	for( i=0; i<nthreads; i++)
	{
		TerminateThread( threads[i], 0);
		CloseHandle( threads[i]);
	}
	//	注销
	_logout();
	closesocket( _soc);
}

void CcmppSocket::_logout()
{
	CMPP_HEAD	msg;
	int			err, nsize;

	nsize = sizeof( msg);
	msg.size  = htonl( nsize);
	msg.cmdid = htonl( nCMPP_TERMINATE);
	msg.seqid = htonl( _getseqid());

	err = _send( (char *)&msg, sizeof( msg));
	if( err != nsize)
		return;

	err = _recv( (char *)&msg, sizeof( msg));

	return;
}

/******************************************************************************
  数据发送线程

  对于新提交的数据报，立即发送
  超过60秒未收到回应，则重发
******************************************************************************/
DWORD	WINAPI	CcmppSocket::thread_send( LPVOID pdata)
{
	CcmppSocket		&cmpp = *( CcmppSocket *)pdata;
	CMPP_PACKAGE	window[nCMPP_WINDOW_SIZE];
	int	i;
	int	err;
	int	nsize;
	int	dwMilliseconds = 1000;		//	轮询间隔为1000毫秒
	for( ;;)
	{
		//	轮询间隔1秒
		err = WaitForSingleObject(
			cmpp._hevnt_data,
			dwMilliseconds);
		//	出错了，结束线程
		if( err == WAIT_FAILED)
			break;
		//	申请数据发送窗口的使用权
		EnterCriticalSection( &cmpp._csec_wnd);
		//	拷贝数据到自己的内存区域，避免在临界区中发送数据
		memcpy( (void *)window, (void *)cmpp._window, sizeof( window));
		//	搜寻并修改到期时间
		for( i=0; i<nCMPP_WINDOW_SIZE; i++)
		{
			//	空位，跳过
			if( cmpp._window[i].head.cmdid == 0)
				continue;
			//	未到发送时间，跳过
			if( cmpp._window[i].t > time( NULL))
				continue;
			//	设置下一次发送的时间，当前时间＋60秒
			cmpp._window[i].t += 60;
		}
		LeaveCriticalSection( &cmpp._csec_wnd);
		//	搜寻并发送到期的数据
		for( i=0; i<nCMPP_WINDOW_SIZE; i++)
		{
			//	空位，跳过
			if( window[i].head.cmdid == 0)
				continue;
			//	未到发送时间，跳过
			if( window[i].t > time( NULL))
				continue;
			//	发送
			nsize = ntohl( window[i].head.size);
			err = cmpp._send( (char *)&window[i], nsize);
			//	发送出错，结束线程
			if( err != nsize)
				return 0;
		}
		//	主线程请求退出
		if( cmpp._bexitting)
			break;
	}
	return 0;
}

DWORD	WINAPI	CcmppSocket::thread_recv( LPVOID pdata)
{
	int		err, i;
	long	prevcount;
	FD_SET	fdset;
	TIMEVAL	timeout;
	CcmppSocket		&cmpp = *( CcmppSocket *)pdata;
	CMPP_PACKAGE	pkg;

	FD_ZERO( &fdset);
	FD_SET( cmpp._soc, &fdset);
	//	轮询间隔1秒
	timeout.tv_sec = 1;
	timeout.tv_usec= 0;

	for( ;;)
	{
		err = select( 
			0,
			&fdset,
			NULL,
			NULL,
			&timeout);
		//	出错
		if( err == SOCKET_ERROR)
			break;
		//	超时
		if( err == 0)
			continue;
		//	先接收包头部分，以确定包的大小、类型
		err = cmpp._recv( (char *)&pkg.head, sizeof( pkg.head));
		if( err != sizeof( pkg.head))
			break;
		//	接收包体
		err = cmpp._recv( pkg.data, ntohl( pkg.head.size));
		if( err == SOCKET_ERROR)
			break;
		//	申请数据发送窗口的使用权
		EnterCriticalSection( &cmpp._csec_wnd);
		//	删除对应流水号的包
		for( i=0; i<nCMPP_WINDOW_SIZE; i++)
		{
			if( cmpp._window[i].head.cmdid == 0)
				continue;
			if( cmpp._window[i].head.seqid == pkg.head.seqid)
			{
				cmpp._window[i].head.cmdid = 0;
				break;
			}
		}
		LeaveCriticalSection( &cmpp._csec_wnd);
		//	释放一个窗口格子
		ReleaseSemaphore(
			cmpp._hsema_wnd,
			1,
			&prevcount);
		//	主线程请求退出
		if( cmpp._bexitting)
			break;
	}
	return 0;
}

/******************************************************************************
*	为了保证线程及时正常退出，设定轮询间隔为1秒
*	CMPP2.0建议的链路检测间隔为180秒
*	为了兼容各个厂家的实现，设为30秒
******************************************************************************/
DWORD	WINAPI	CcmppSocket::thread_actv( LPVOID pdata)
{
	CcmppSocket &cmpp = *( CcmppSocket *)pdata;
	CMPP_HEAD	msg;
	int			c = 0,		//	计数
				n = 30;		//	等待的秒数
	for( ;;)
	{
		Sleep( 1000);
		//	主线程请求退出
		if( cmpp._bexitting)
			break;
		if( ++ c < n)
			continue;
		c = 0;
		//	获得数据发送权利，发送链路测试包
		msg.size = htonl( sizeof( msg));
		msg.cmdid= htonl( nCMPP_ACTIVE_TEST);
		msg.seqid= htonl( cmpp._getseqid());

		cmpp._send( (char *)&msg, sizeof( msg));
	}
	return 0;
}

char * CcmppSocket::_timestamp( char *buf)
{
	time_t tt;
	struct tm *now;

	time( &tt);
	now = localtime( &tt);
	sprintf( buf, "%02d%02d%02d%02d%02d",
		now->tm_mon + 1,
		now->tm_mday,
		now->tm_hour,
		now->tm_min,
		now->tm_sec);
	return buf;
}

__int64	CcmppSocket::_ntoh64( __int64 inval)
{
	__int64 outval = 0;
	for( int i=0; i<8; i++)
		outval = ( outval << 8) + ( ( inval >> ( i * 8)) & 255);

	return outval;
}

__int64	CcmppSocket::_hton64( __int64 inval)
{
	return _ntoh64( inval);
}

int	CcmppSocket::_getseqid()
{
	int id;
	EnterCriticalSection( &_csec_seq);
	id = ++_seqid;
	LeaveCriticalSection( &_csec_seq);
	return id;
}

int	CcmppSocket::_send( char *buf, int len)
{
	int err;

	EnterCriticalSection( &_csec_snd);
	err = send( _soc, buf, len, 0);
	LeaveCriticalSection( &_csec_snd);

	return err;
}

int	CcmppSocket::_recv( char *buf, int len)
{
	int err;

	EnterCriticalSection( &_csec_recv);
	err = recv( _soc, buf, len, 0);
	LeaveCriticalSection( &_csec_recv);

	return err;
}

int main()
{
	CcmppSocket cmpp;

	cmpp.init( "09995", "1234",  "192.168.0.181");

	CMPP_SUBMIT msg;
	memset( (void *)&msg, 0, sizeof( msg));
	msg.desttotal = 1;
	strcpy( (char *)msg.destnumbers, "13808425257");
	msg.msglen = 10;
	strcpy( (char *)msg.msgcontent, "0123456789");
	strcpy( (char *)msg.reserve, "abcdefgh");

	for( int i=0; i<1000; i++)
		cmpp.Submit( msg);

	Sleep( 1000);
	return 0;
}