upreceiver.cpp

SMS gateway. SMS protocol for CHINA mobile, unicom, lingtong. Using mysql to exchange message.

/**********************************************************************
  FileName            : UpReceiver.cpp
  Description         : 上节点接收(线程)模块
  Version             : 1.0
  Create Date         : 2003年6月17日
  Update Date		  : 2003年11月17日
  Author              : 刘荣辉
  Other               : 
***********************************************************************/

#include "UpNode.h"

void * CUpNode::UpReceiver(void *node)
{
  int threadid;
  int RetCode;
  char sSysEvent[SYS_EVENT_LEN];	//记录系统事件的字符串
  int PackTag;
  //PackTag为-1表示应答包,为0表示正确的请求包
  //	   >0表示错误的请求包(不放入接收队列,但返回应答包,仍维持连接)
  //       为999表示需要重新连接
  void **pPack;
  char *Pack,*RspPack;
  void **pRspPack;
  unsigned int PackType;
  RecvQUnit  RUnit;
  SendQUnit  SUnit;
  time_t nowtime;
  
  threadid = pthread_self();
  pthread_detach(threadid);  
  sprintf(sSysEvent,"LOG: UpReceiver[%d] for UpNode[%s] is started.",threadid,NodeCode);
  WrSystemLog->WriteLog(sSysEvent,SYSTEMLOG);    //写系统日志
  
  pPack = (void **)malloc(sizeof(char **));
  pRspPack = (void **)malloc(sizeof(char **));

  while(1)	//不断从上节点接收数据包放入接收队列
  {
  		RetCode = TcpSock.RecvPack(pPack,MAX_PACK_SIZE);	//接收一个完整包
  		Pack = (char *)*pPack;
    	if(RetCode)
    	{
			TcpSock.Close_sock();
    		State = 1;	//节点状态为正在连接
    		if(RetCode==-1)
			{
    		   if(ToExit)
			   {
				   State=2;
				   break;
			   }
			   else
				   sprintf(sSysEvent,"Error[%d]: Connection with UpNode[%s] lost!Reconnecting.",errno,NodeCode);
			}
            else if(RetCode==-2)
               sprintf(sSysEvent,"Error: Packet length from UpNode[%s] is abnormal! Reconnecting.",NodeCode);
            else
			   sprintf(sSysEvent,"Error[%d]: when RecvPack return.Reconnecting UpNode[%s].",RetCode,NodeCode);
			if(!ToExit)
				WrSystemLog->WriteLog(sSysEvent,SYSTEMLOG);
            break;  //return (void *)NULL;	//退出接收模块
    	}
	  
	  //-----------------debug----------------------
	  unsigned char *ptmp;
	  ptmp=(unsigned char *)Pack;
	  int len;
	  len = ntohl(*(int *)Pack); 
	  if(len > 13)
	  {
		  if(Protocol==0)
				printf("\n=====UpReceiver: checking a CMPP_Pack Received==== \n");
		  if(Protocol==1)
				printf("\n=====UpReceiver: checking a SMGP_Pack Received==== \n");
		  printf("PackLen=[%d]\n",len);
		  for(int i=0;i<len;i++)
		  {
			 printf("%02x ",*(ptmp+i));
			 if(((i+1)%4)==0) printf("  ");
			 if((i+1)==12) printf("\n");
		  }
		  printf("\n=========================================== \n");
	  }
	  //--------------------debug end-------------------

		//--------------------------应答处理-----------------------------
    	PackType = ntohl(*(unsigned int *)(Pack+4));
    	PackTag = -1;	//先假设收到的包为应答包
    	if(PackType<0x80000000)	//若为请求包,则形成应答包,返回值为状态码
    	{
			if(ToExit)	//系统即将退出，对请求包一律抛弃，只对应答包进行处理
			{
				free(Pack);	//释放请求包所占内存
				continue;
			}
    		PackTag = MakeRspPack(Pack, pRspPack);	//返回值为请求包的格式校验结果
    		RspPack = (char *)*pRspPack;
			  /*/-----------------debug----------------------
			  if(RspPack)
			  {
				  printf("\n=========checking the RspPack made========= \n");
				  unsigned char *ptmp;
				  ptmp=(unsigned char *)RspPack;
				  int len;
				  len = ntohl(*(int *)RspPack);
				  for(int i=0;i<len;i++)
				  {
					 printf("%02x ",*(ptmp+i));
					 if(((i+1)%4)==0) printf("  ");
					 if((i+1)==12) printf("\n");
				  }
				  printf("\n=========================================== \n");
			  }//if(RspPack)
			  //--------------------debug end-------------------  */
    	}//if,若为请求包
		
		if(Protocol==0 && PackType == CMPP_TERMINATE_RESP)
		{
			sprintf(sSysEvent,"LOG: Got a CMPP_TERMINATE_RESP from UpNode[%s]!\n",NodeCode);
			WrSystemLog->WriteLog(sSysEvent,SYSTEMLOG);
			free(Pack);	//释放请求包所占内存
			TcpSock.Close_sock();
			State = 2;	//节点状态为暂时不再连接
			break;	//退出接收模块
		}
		#ifdef DEBUG
		if(Protocol==0 && PackType == CMPP_ACTIVE_TEST)
		{
			printf("\n UpReceiver: Got a CMPP_ActiveTest!\n");
		}
		else if(Protocol==0 && PackType == CMPP_ACTIVE_TEST_RESP)
		{
			//printf("\n UpReceiver: Got a CMPP_ActiveTest_Rsp!\n");
		}
		#endif

		if(PackTag>=999)	//收到的包严重错误,需要重新建立连接
		{
			free(Pack);	//释放请求包所占内存
			TcpSock.Close_sock();
			State = 1;	//节点状态为正在连接
			if(PackTag==1000)
				sprintf(sSysEvent,"Error[%d]: Got Terminate/Exit Pack from UpNode[%s]! Reconnecting.",PackTag,NodeCode);
			else
				sprintf(sSysEvent,"Error[%d]: Got error PDU from UpNode[%s]! Reconnecting.",PackTag,NodeCode);
			WrSystemLog->WriteLog(sSysEvent,SYSTEMLOG);
			break;  //return (void *)NULL;	//退出接收模块
		}

    	//将收到的正确请求包（Deliver）写入日志文件、放入接收队列
		if(PackTag==0)
		{
			if(Protocol==0 && PackType == CMPP_DELIVER)	//CMPP_Deliver
			{
				CMPP_Deliver *pDeliver = (CMPP_Deliver *)Pack;
				Cmpp_Deliver_Log DeliverUpLog;
				WrSystemLog->getTime(DeliverUpLog.RecvTime);
				DeliverUpLog.SrcNode = NodeID;
				DeliverUpLog.Deliver = pDeliver;
				if(!pDeliver->Registered_Delivery)	//写入Deliver日志
				{
					//编码格式转换Unicode->GBK,并修改消息长度和包总长度
					if(pDeliver->Msg_Fmt==8)
					{
						int OldMsgLen;
						OldMsgLen = pDeliver->Msg_Length;
						char tmpReserve[8];
						memcpy(tmpReserve,&pDeliver->Deliver_Msg.Msg_Content[pDeliver->Msg_Length],8);
						/*printf("\n UpReceiver: Content=[%s],Unicode= [",pDeliver->Deliver_Msg.Msg_Content);
						for(int a=0; a<pDeliver->Msg_Length; a++)
							printf("%02x ",(unsigned char)pDeliver->Deliver_Msg.Msg_Content[a]);
						printf("]\n");*/
						pDeliver->Msg_Length = CFunc::CodeConvert("UTF-16BE","GB2312",pDeliver->Deliver_Msg.Msg_Content,pDeliver->Deliver_Msg.Msg_Content,pDeliver->Msg_Length);
						//pDeliver->Msg_Length = CFunc::CodeConvert("Unicode","GB2312",pDeliver->Deliver_Msg.Msg_Content,pDeliver->Deliver_Msg.Msg_Content,pDeliver->Msg_Length);
						memcpy(&pDeliver->Deliver_Msg.Msg_Content[pDeliver->Msg_Length],tmpReserve,8);
						pDeliver->Head.Total_Length = htonl(ntohl(pDeliver->Head.Total_Length) + pDeliver->Msg_Length - OldMsgLen);
						pDeliver->Msg_Fmt = 15;
						/*printf("\n UpReceiver: GB2312= [");
						for(int a=0; a<pDeliver->Msg_Length; a++)
							printf("%02x ",(unsigned char)pDeliver->Deliver_Msg.Msg_Content[a]);
						printf("]\n");*/
					}
					/*printf("\n==============UpReceiver: After Convert=================\n");
					ptmp=(unsigned char *)Pack;
					len = ntohl(*(int *)Pack);
					for(int i=0;i<len;i++)
					  {
						 printf("%02x ",*(ptmp+i));
						 if(((i+1)%4)==0) printf("  ");
						 if((i+1)==12) printf("\n");
					  }
					printf("\n=========================================== \n");*/

					if((RetCode = WrDeliverLog->WriteLog(&DeliverUpLog,CMPP_DELIVER_LOG))>0)
					  {
							sprintf(sSysEvent,"Error[%d]: UpReceiver Writing Cmpp_Deliver_Log File!",RetCode); 
							WrSystemLog->WriteLog(sSysEvent,SYSTEMLOG);    /*写系统日志*/
							break;
					  }
					#ifdef DEBUG
					//printf("\n UpReceiver: Cmpp_Deliver_UpLog is Writen sucessfully!\n");
					#endif
				}
				else	//写入Report日志
				{
					if((RetCode = WrReportLog->WriteLog(&DeliverUpLog,CMPP_REPORT_LOG))>0)
					  {
							sprintf(sSysEvent,"Error[%d]: UpReceiver Writing Cmpp_Report_Log File!",RetCode); 
							WrSystemLog->WriteLog(sSysEvent,SYSTEMLOG);    /*写系统日志*/
							break;
					  } 
					#ifdef DEBUG
					printf("\n UpReceiver: Cmpp_Report Log file is Writen sucessfully!\n");
					#endif
				}
			}//CMPP_Deliver

			//-----------------------------------------------------------------
			else if(Protocol==1 && PackType == REQUEST_ID_DELIVER)	//SMGP
			{
				SMGP_Deliver *pDeliver = (SMGP_Deliver *)Pack;
				Smgp_Deliver_Log DeliverUpLog;
				WrSystemLog->getTime(DeliverUpLog.RecvTime);
				DeliverUpLog.SrcNode = NodeID;
				DeliverUpLog.Deliver = pDeliver;
				if(!pDeliver->nIsReport)	//写入Deliver日志
				{
					if((RetCode = WrDeliverLog->WriteLog(&DeliverUpLog,SMGP_DELIVER_LOG))>0)
					  {
							sprintf(sSysEvent,"Error[%d]: UpReceiver Writing Smgp_Deliver_UpLog File!",RetCode); 
							WrSystemLog->WriteLog(sSysEvent,SYSTEMLOG);    /*写系统日志*/
							break;
					  }
					#ifdef DEBUG
					printf("\n UpReceiver: Smgp_Deliver_UpLog is Writen sucessfully!\n");
					#endif
				}
				else	//写入Report日志
				{
					if((RetCode = WrReportLog->WriteLog(&DeliverUpLog,SMGP_REPORT_LOG))>0)
					  {
							sprintf(sSysEvent,"Error[%d]: UpReceiver Writing Smgp_Report Log File!",RetCode); 
							WrSystemLog->WriteLog(sSysEvent,SYSTEMLOG);    /*写系统日志*/
							break;
					  }
					#ifdef DEBUG
					printf("\n UpReceiver: Smgp_Report Log file is Writen sucessfully!\n");
					#endif
				}
			}//SMGP
		}//正确请求包（Deliver）写入日志文件、放入接收队列
		
		//收到的正确请求包（除ActiveTest外）或应答包,形成接收队列单元,放入接收队列
    	if(PackTag<=0)
    	{
    		RUnit.Pack = Pack;
    		time(&nowtime);
    		RUnit.RecvTime = nowtime;
			if(State && ToExit)
			{
			   State = 2;	//节点状态为暂时不再连接
			   break;
			}
			if(PackType<0x80000000)	//放进请求包接收缓冲队列
			{
				while(RecvQ->Put(RUnit)==false)
				{
					if(State && ToExit)
					{
					   State = 2;	//节点状态为暂时不再连接
					   goto UpRecvEnd;
					}
					sprintf(sSysEvent,"Warning: RecvQ of UpNode[%s] is full!Recv_Q_Full_Wait=[%d]!",NodeCode,Recv_Q_Full_Wait);
					WrSystemLog->WriteLog(sSysEvent,SYSTEMLOG);
					usleep(Recv_Q_Full_Wait);
				}
			}
			else	//放进应答包接收缓冲队列
			{
				while(RspQ->Put(RUnit)==false)
				{
					if(State && ToExit)
					{
					   State = 2;	//节点状态为暂时不再连接
					   goto UpRecvEnd;
					}
					sprintf(sSysEvent,"Warning: RspQ of UpNode[%s] is full!Recv_Q_Full_Wait=[%d]!",NodeCode,Recv_Q_Full_Wait);
					WrSystemLog->WriteLog(sSysEvent,SYSTEMLOG);
					usleep(Recv_Q_Full_Wait);
				}
			}
			#ifdef DEBUG
			//printf("\n UpReceiver: A pack[PackTag=%d] is put into RecvQ.\n",PackTag);
			#endif
		}

    	//待发应答包,形成发送队列单元,放入发送队列
    	if(PackTag>=0 && RspPack!=NULL)
    	{
    		SUnit.Pack = RspPack;
			//SUnit.SrcRecvTime
			SUnit.SrcNode = NodeID;
			//SUnit.ID_Relate
    		time(&nowtime);
    		SUnit.SendTime = nowtime;
	    	SUnit.iResent = -1;		//-1为应答包,发送成功就删除
    		//SUnit.SrcSequence = 0;
    		while(SendQ->Put(SUnit)==false)
			{
				if(State && ToExit)
				{
				   State = 2;	//节点状态为暂时不再连接
				   goto UpRecvEnd;
				}
				sprintf(sSysEvent,"Warning: SendQ of UpNode[%s] is full!Send_Q_Full_Wait=[%d]!",NodeCode,Send_Q_Full_Wait);
				WrSystemLog->WriteLog(sSysEvent,SYSTEMLOG); 
				usleep(Send_Q_Full_Wait);
			}
			#ifdef DEBUG
			//printf("\n UpReceiver: A RspPack is put into SendQ.\n");
			#endif
		}

		if(State)	//用于当UpSender先发现连接断开的情况
			break;

		/*/--------- 测试： 模拟连接断开 -------------
		if(SeqId%10 == 0)
		{
			printf("\n ================UpReceiver: connection is cut for testing!========\n");
			TcpSock.Close_sock();
			State = 1;	//节点状态为正在连接
			break;
		}*/
		//--------- 测试： 模拟连接断开 -------------
  
  }//while(1)

UpRecvEnd:

  free(pPack);		//释放请求包指针的指针本身所占内存
  free(pRspPack);	//释放应答包指针的指针本身所占内存
  ReadThr = 0;
  if(ToExit)
  {
	State = 2;	//节点状态为暂时不再连接 
	sprintf(sSysEvent,"LOG: UpReceiver of UpNode[%s] is ended!",NodeCode);
  }
  else
	sprintf(sSysEvent,"Warning: UpReceiver of UpNode[%s] QUIT! ",NodeCode);
  WrSystemLog->WriteLog(sSysEvent,SYSTEMLOG);

  return (void *)NULL;
}