cmpp.c

基于linux下C语言实现的CMPP通信协议

/* 
	file:cmpp.c
	china mobile peer to peer (cmpp) interface by jiniang 2005.1.31 
	ver:2.0
*/

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <unistd.h>
#include <fcntl.h>
#include <sys/stat.h>
#include<unistd.h>
#include <pthread.h>
#include <semaphore.h>
#include <errno.h>
#include "global.h"
#include "md5.h"

#include "code.h"
#include "cmpp.h"
#include "fifo.h"

/* cmpp command define part */
#define CMPP_CONNECT 0x00000001
#define CMPP_CONNECT_RESP 0x80000001
#define CMPP_TERMINATE 0x00000002
#define CMPP_TERMINATE_RESP 0x80000002
#define CMPP_SUBMIT 0x00000004
#define CMPP_SUBMIT_RESP 0x80000004
#define CMPP_DELIVER 0x00000005
#define CMPP_DELIVER_RESP 0x80000005
#define CMPP_QUERY 0x00000006
#define CMPP_QUERY_RESP 0x80000006
#define CMPP_CANCEL 0x00000007
#define CMPP_CANCEL_RESP 0x80000007
#define CMPP_ACTIVE_TEST 0x00000008
#define CMPP_ACTIVE_TEST_RESP 0x80000008
#define CMPP_FWD 0x00000009
#define CMPP_FWD_RESP 0x80000009
#define CMPP_MT_ROUTE 0x00000010
#define CMPP_MT_ROUTE_RESP 0x80000010
#define CMPP_MO_ROUTE 0x00000011
#define CMPP_MO_ROUTE_RESP 0x80000011
#define CMPP_GET_ROUTE 0x00000012
#define CMPP_GET_ROUTE_RESP 0x80000012
#define CMPP_MT_ROUTE_UPDATE 0x00000013
#define CMPP_MT_ROUTE_UPDATE_RESP 0x80000013
#define CMPP_MO_ROUTE_UPDATE 0x00000014
#define CMPP_MO_ROUTE_UPDATE_RESP 0x80000014
#define CMPP_PUSH_MT_ROUTE_UPDATE 0x00000015
#define CMPP_PUSH_MT_ROUTE_UPDATE_RESP 0x80000015
#define CMPP_PUSH_MO_ROUTE_UPDATE 0x00000016
#define CMPP_PUSH_MO_ROUTE_UPDATE_RESP 0x80000016

/* define something for cmpp */
#define CMPP_C 180		//send a active_test every 180s
#define CMPP_T 60		//if not received send again after 60s
#define CMPP_N 3		//try 3 time 

#define CMPP_VERSION 0x20	//version:2.0

/* declare glabol private var */
char cmpp_source_addr[6];
char cmpp_sharesecret[256];

struct fifo cmpp_mo;
struct cmpp_head
{
	unsigned int total_length;
	unsigned int command_id;
	unsigned int sequence_id;
};

struct sockaddr_in cmpp_addr;
unsigned int cmpp_buf_size=0;		//buffer size for deliver
unsigned int cmpp_sequence_id=0;				
unsigned char cmpp_server_version=0;	//gateway version
int cmpp_sock;														//socket

int cmpp_recv_run;												//contral recv while
pthread_t cmpp_recv_pid;

pthread_mutex_t cmpp_recv_mutex=PTHREAD_MUTEX_INITIALIZER;	//recv lock
pthread_cond_t cmpp_recv_cond=PTHREAD_COND_INITIALIZER;			//recv wait

pthread_cond_t cmpp_send_cond=PTHREAD_COND_INITIALIZER;			//send wait
pthread_cond_t cmpp_conn_cond=PTHREAD_COND_INITIALIZER;			//conn wait

pthread_mutex_t cmpp_hasmo_mutex=PTHREAD_MUTEX_INITIALIZER;	
pthread_cond_t cmpp_hasmo_cond=PTHREAD_COND_INITIALIZER;

sem_t cmpp_recv_sem;
sem_t cmpp_send_sem;

/* declare glabol private func */
void cmpp_recv_thread();

/*	define func	*/
int cmpp_init(const char *ip,unsigned int port,unsigned int buf_size)
{
	fifo_init(&cmpp_mo,buf_size);
	cmpp_buf_size=buf_size;
	bzero(&cmpp_addr,sizeof(cmpp_addr));
	cmpp_addr.sin_family=AF_INET;
	cmpp_addr.sin_port=htons(port);
	cmpp_addr.sin_addr.s_addr=inet_addr(ip);
	
	sem_init(&cmpp_recv_sem,0,0);
	sem_init(&cmpp_send_sem,0,0);
	
	cmpp_recv_run=1;
	int ret=pthread_create(&cmpp_recv_pid,NULL,(void*)cmpp_recv_thread,NULL);
		
	if(ret)
	{
			perror("create cmpp_recv_thread error");
			return -2;
	}
	return 0;
}

int cmpp_exit()
{
	cmpp_recv_run=0;
	sem_post(&cmpp_recv_sem);
	pthread_mutex_unlock(&cmpp_recv_mutex);
	pthread_mutex_destroy(&cmpp_recv_mutex);
	pthread_join(cmpp_recv_pid,NULL);
	return 0;
}

int cmpp_connect(const char *source_addr,const char *sharesecret)
{
	struct cmpp_head head;
	bzero(&head,12);
	char cmpp_connect[39]="";
	
	/* init socket */
	cmpp_sock=socket(AF_INET,SOCK_STREAM,0);
	if(cmpp_sock<0)
	{
		perror("cmpp socket error");
		return -1;	
	}
	/* head */
	head.total_length=htonl(39);
	head.command_id=htonl(CMPP_CONNECT);
	head.sequence_id=htonl(++cmpp_sequence_id);
	
	/* body */ 
	memcpy(cmpp_connect,&head,12);
	memcpy(cmpp_connect+12,source_addr,6);

	int n=strlen(sharesecret);
	char *md5string=(char*)malloc(35+n);
	memcpy(md5string,source_addr,6);
	bzero(md5string+6,9);
	memcpy(md5string+15,sharesecret,n);
	time_t timep;
	time(&timep);
	struct tm *now=localtime(&timep);
	char timestamp[11];
	sprintf(timestamp,
		"%2.2d%2.2d%2.2d%2.2d%2.2d",
		now->tm_year-100,
		now->tm_mon+1,
		now->tm_mday,
		now->tm_hour,
		now->tm_min);
	//strftime(timestamp,sizeof(timestamp),"%y%m%d%H%M",now);
	memcpy(md5string+15+n,timestamp,10);
	MD5_CTX context;
	MD5Init(&context);
	MD5Update(&context,md5string,25+n);
	MD5Final(cmpp_connect+18,&context);
	free(md5string);
	cmpp_connect[34]=CMPP_VERSION;
	unsigned int time=htonl(atoi(timestamp));
	memcpy(cmpp_connect+35,&time,4);
	
	/* connect */
	int rsflags=fcntl(cmpp_sock,F_GETFL);
	if(rsflags==-1)
	{
		perror("get rsflags error")	;
	}
	rsflags|=O_NONBLOCK;
	fcntl(cmpp_sock,F_SETFL,O_NDELAY);	//set nonblock
	n=connect(cmpp_sock,(struct sockaddr *)&cmpp_addr,sizeof(cmpp_addr));
	int ret=1;//timeout or error happen,when ret equ 1
	if(n==-1)
	{
		if(errno==EINPROGRESS)
		{
			fd_set writefds;
			FD_ZERO(&writefds);
			FD_SET(cmpp_sock,&writefds);

			struct timeval tv;
			tv.tv_sec=5;			//5s timeout
			tv.tv_usec=0;
			if(select(cmpp_sock+1,NULL,&writefds,NULL,&tv)>0)
			{
				int error=0,errlen=sizeof(int);
				getsockopt(cmpp_sock,SOL_SOCKET,SO_ERROR,&error,&errlen);
				if(!error) ret=0;
			}
		}
	}
	else
	{
		ret=0;
	}
	if(ret) 
	{	
		perror("cmpp connect error or timeoutssss");
		return -1;	
	}
	if(send(cmpp_sock,cmpp_connect,39,0)<=0)
	{
		perror("cmpp connect send error");
		return -2;
	}
	
	/* cmpp_connect_resp */
	char connect_resp[30]="";
	
	struct timeval tv;
	tv.tv_sec=30;			//30s timeout
	tv.tv_usec=0;
	fd_set readfds;
	FD_ZERO(&readfds);
	FD_SET(cmpp_sock,&readfds);
	n=select(cmpp_sock+1,&readfds,NULL,NULL,&tv);
	if(n<=0)
	{
		perror("connect resp timeout or error");
		return -3;	
	}
	if((n=recv(cmpp_sock,connect_resp,30,0))<30)
	{
		perror("cmpp connect recv error");
		return -4;
	}
	memcpy(&head,connect_resp,12);
	if(head.command_id!=htonl(CMPP_CONNECT_RESP))
	{
			return -5;
	}
	if(connect_resp[12])	//login 
	{
		return connect_resp[12];
	}
	cmpp_server_version=connect_resp[29];
	strncpy(cmpp_source_addr,source_addr,6);
	strcpy(cmpp_sharesecret,sharesecret);
	sem_post(&cmpp_recv_sem);		//active recv threaed;
	return 0;
}

int cmpp_terminate()
{
	struct cmpp_head head;
	bzero(&head,12);
	
	/* head */ // no body
	head.total_length=htonl(12);
	head.command_id=htonl(CMPP_SUBMIT);
	head.sequence_id=htonl(++cmpp_sequence_id);
	
	/* send */
	if(send(cmpp_sock,&head,12,0)<=0)
	{
		perror("cmpp terminate send error");
		return -1;
	}
	return 0;
}

int cmpp_active_test()
{
	struct cmpp_head head;
	bzero(&head,12);
	
	// head //	no body
	head.total_length=htonl(12);
	head.command_id=htonl(CMPP_ACTIVE_TEST);
	head.sequence_id=htonl(++cmpp_sequence_id);
	
	// send //
	fd_set writefds;
	FD_ZERO(&writefds);
	FD_SET(cmpp_sock,&writefds);
	if(select(cmpp_sock+1,NULL,&writefds,NULL,NULL)<0)
	{
		return -1;
	}
	if(send(cmpp_sock,&head,12,0)<=0)
	{
			perror("cmpp terminate send error");
			return -2;
	}
	return 0;
}

int cmpp_submit(char *const msg_id,
								char pk_total,
								char pk_number,
								char recmpp_delivery,
								char msg_level,
								const char *service_id,
								char fee_user_type,
								const char *fee_terminal_id,
								char tp_pid,
								char tp_udhi,
								char msg_fmt,
								const char *msg_src,
								const char *fee_type,
								const char *fee_code,
								const char *valid_time,
								const char *at_time,
								const char *src_id,
								char dest_usr_tl,
								const char *dest_terminal_id,
								char msg_len,
								const char *msg_content,
								const char *reserve)
{
	struct cmpp_head head;
	bzero(&head,12);
	
	// code //
	char text[160];
	memset(text,0,160);
	switch(msg_fmt)
	{
		case 8:	//unicode
			msg_len=gb2unicode((unsigned char *)text,(unsigned char *)msg_content,(unsigned int)msg_len);
			break;
		case 0:	//ascii
		case 3:	//write card
		case 4:	//binary
		case 15://gb
		default:
			memcpy(text,msg_content,strlen(msg_content));
			break;
	}
	
	// head //
	head.total_length=htonl(159+msg_len);
	head.command_id=htonl(CMPP_SUBMIT);
	head.sequence_id=htonl(++cmpp_sequence_id);
	
	// body //
	char *cmpp_submit=(char*)malloc(159+msg_len);
	memset(cmpp_submit,0,159+msg_len);
	memcpy(cmpp_submit,&head,12);	//head
	//memset(cmpp_submit+12,0,8);		//msg_id
	*(cmpp_submit+20)=pk_total;
	*(cmpp_submit+21)=pk_number;
	*(cmpp_submit+22)=recmpp_delivery,
	*(cmpp_submit+23)=msg_level,
	strncpy(cmpp_submit+24,service_id,strlen(service_id));
	*(cmpp_submit+34),fee_user_type;
	strncpy(cmpp_submit+35,fee_terminal_id,strlen(fee_terminal_id));
	*(cmpp_submit+56)=tp_pid,
	*(cmpp_submit+57)=tp_udhi,
	*(cmpp_submit+58)=msg_fmt,
	strncpy(cmpp_submit+59,msg_src,strlen(msg_src));
	strncpy(cmpp_submit+65,fee_type,strlen(fee_type));
	strncpy(cmpp_submit+67,fee_code,strlen(fee_code));
	strncpy(cmpp_submit+73,valid_time,strlen(valid_time));
	strncpy(cmpp_submit+90,at_time,strlen(at_time));
	strncpy(cmpp_submit+107,src_id,strlen(src_id));
	*(cmpp_submit+128)=dest_usr_tl;
	strncpy(cmpp_submit+129,dest_terminal_id,strlen(dest_terminal_id));
	*(cmpp_submit+150)=msg_len;
	memcpy(cmpp_submit+151,text,msg_len);
	memcpy(cmpp_submit+151+msg_len,reserve,8);
	
	/* send */
	fd_set writefds;
	FD_ZERO(&writefds);
	FD_SET(cmpp_sock,&writefds);
	if(select(cmpp_sock+1,NULL,&writefds,NULL,NULL)<0)
	{
		free(cmpp_submit);
		perror("cmpp submit select error");
		return -1;
	}
	pthread_mutex_lock(&cmpp_recv_mutex);
	if(send(cmpp_sock,cmpp_submit,159+msg_len,0)<=0)
	{
		free(cmpp_submit);
		perror("cmpp submit send error");
		return -2;
	}
	free(cmpp_submit);
	
	/* cmpp_submit_resp */
	char connect_resp_body[9];
	memset(connect_resp_body,0,9);
	
	struct timespec abstime;			//15s timeout
	clock_gettime(CLOCK_REALTIME,&abstime); //get current time
	abstime.tv_sec+=15;		//set current time + 15sec
	int timeout=0;
	int n=0;					//recv size
	if(ETIMEDOUT==pthread_cond_timedwait(&cmpp_send_cond,&cmpp_recv_mutex,&abstime))//timeout after 15s
	{
		timeout=1;
		perror("submit resp timeout");
	}
	else
	{
		n=recv(cmpp_sock,connect_resp_body,9,0);
	}
	pthread_cond_signal(&cmpp_recv_cond);
	pthread_mutex_unlock(&cmpp_recv_mutex);
	if(timeout)
	{
		return -3;		//recv submit resp timeout
	}
	if(n<9)
	{
		return -4;		//recv error
	}
	memcpy(msg_id,connect_resp_body,8);
	return *(connect_resp_body+8);
}

int cmpp_deliver_resp(const char *msg_id,char result)
{
	struct cmpp_head head;
	bzero(&head,12);
	
	/* head */
	head.total_length=htonl(21);
	head.command_id=htonl(CMPP_DELIVER_RESP);
	head.sequence_id=htonl(++cmpp_sequence_id);
	
	/* body */
	char deliver_resp[21];
	memcpy(deliver_resp,&head,12);
	memcpy(deliver_resp+12,msg_id,8);
	*(deliver_resp+20)=result;
	
	/* send */
	if(send(cmpp_sock,&head,21,0)<=0)
	{
		perror("cmpp terminate send error");
		return -1;
	}
	return 0;
}

void cmpp_recv_thread()
{
	struct cmpp_head head;
	memset(&head,0,12);
	sem_wait(&cmpp_recv_sem);	//start run
	time_t active_time=time(NULL);//set active time at now

	fd_set readfds;
	
	while(cmpp_recv_run)
	{
		/* head */
		FD_ZERO(&readfds);
		FD_SET(cmpp_sock,&readfds);	
		//fd_set writefds
		//FD_ZERO(&writefds);
		//FD_SET(cmpp_sock,&writefds);
		struct timeval tv;
		tv.tv_sec=0;			//0.010 s timeout
		tv.tv_usec=100;
		if(active_time+CMPP_C <= time(NULL))
		{
			if(!cmpp_active_test())
				active_time=time(NULL);//update acive time
		}
		int ret=select(cmpp_sock+1,&readfds,NULL,NULL,&tv);
		if(ret==0) continue;	//if timeout ,while
		if(ret<0)			//socket error
		{
			perror("\nsocket error ret");
			goto close;	//reconnect if error
		}
		
		int n=recv(cmpp_sock,(char*)&head,12,0);
		
		if(n<0) goto close;	//reconnect

		unsigned int total_length=ntohl(head.total_length);
		switch(ntohl(head.command_id))
		{
			case CMPP_TERMINATE:
			{
				cmpp_terminate();
				break;	
			}
			case CMPP_TERMINATE_RESP:
			{
				close(cmpp_sock);
				break;
			}
			case CMPP_ACTIVE_TEST_RESP:
			{
				break;
			}
			case CMPP_SUBMIT_RESP:
			{
				pthread_mutex_lock(&cmpp_recv_mutex);
				pthread_cond_signal(&cmpp_send_cond);
				struct timespec abstime;
				clock_gettime(CLOCK_REALTIME,&abstime); //get current time
				abstime.tv_sec+=15;//15s timeout
				pthread_cond_timedwait(&cmpp_recv_cond,&cmpp_recv_mutex,&abstime);//lock me
				pthread_mutex_unlock(&cmpp_recv_mutex);
				break;
			}
			case CMPP_DELIVER:
			{
				char *buf=(char*)malloc(total_length-12);
				n=recv(cmpp_sock,(char*)buf,total_length-12,0);
				if(n<=0)
				{
					free(buf);
					continue;
				}
				if(n!=total_length-12)
				{
						cmpp_deliver_resp(buf,4);	//err:message size error
				}
				if(!fifo_put(&cmpp_mo,buf,n))
				{
					pthread_mutex_lock(&cmpp_hasmo_mutex);
					pthread_cond_signal(&cmpp_hasmo_cond);	//message cmpp_has_mo
					pthread_mutex_unlock(&cmpp_hasmo_mutex);
					cmpp_deliver_resp(buf,0);		//ok
				}
				else
				{
					cmpp_deliver_resp(buf,9);		//err:fifo buffer no enough(9=other error)
				}
				free(buf);
				break;
			}
			default:
			{
				break;
			}
		}
		continue;
close:		//reconnect
		if(!cmpp_recv_run) return;	//exit this thread
		close(cmpp_sock);
		if(cmpp_connect(cmpp_source_addr,cmpp_sharesecret))
			goto close;
	}
	return;
}

int cmpp_has_mo(unsigned int timeout)
{
	int ret=0;				//return value
	struct timespec abstime;		//set timeout
	abstime.tv_sec=timeout/1000;
	abstime.tv_nsec=timeout%1000;
	
	pthread_mutex_lock(&cmpp_hasmo_mutex);
	if(fifo_isempty(&cmpp_mo))
		ret=(pthread_cond_timedwait(&cmpp_hasmo_cond,&cmpp_hasmo_mutex,&abstime)!=ETIMEDOUT); //timeout return 1 else return 0
	else
		ret=1;	//no empty return 0
	pthread_mutex_unlock(&cmpp_hasmo_mutex);
	
	return ret;
}

int cmpp_get_mo(void *const mo)
{
	unsigned int size;
	char buf[233];		//max mo size
	memset(buf,0,233);
	int ret=fifo_get(&cmpp_mo,buf,&size);
	if(ret)
	{
		return CMPP_MO_NULL;	
	}
	else
	{
		if(*(buf+63))	//report
		{
			memcpy(mo,buf+65,60);
			return CMPP_MO_REPORT;
		}
		else			//deliver
		{
			memcpy(mo,buf,size-8);
			memcpy(((struct cmpp_deliver*)mo)->reserved,buf+size-8,8);
			return CMPP_MO_DELIVER;
		}
		return 0;
	}
}