upnodecom.cpp

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

/**********************************************************************
  FileName            : UpNodeCom.cpp
  Description         : 上节点通信主模块
  Version             : 1.0
  Date                : 2003年6月16日
  Author              : 刘荣辉
  Other               : 连接/登陆/接收,产生发送线程
***********************************************************************/

#include "UpNode.h"

void * CUpNode::NodeCom(void *Param)
{
  int RetCode;
  CUpNode *UpNode;
  int imode;
  pthread_attr_t ThreadAttr;
  char sSysEvent[SYS_EVENT_LEN];	//记录系统事件的字符串
  
  UpNode = (CUpNode*)( ((ThreadParam*)Param)->MyPointer);
  imode =  ((ThreadParam*)Param )->MyInt;
  
  float Connect_delay = (float)CONNECT_WAIT;
  ReConnect:
  //建立TCP连接
  sleep((int)Connect_delay);
  Connect_delay *= 1.5;
  UpNode->TcpSock.sock=0;
  RetCode = UpNode->TcpSock.Make_Connect_Sock(UpNode->NodeIp, UpNode->Port);
  if(RetCode <= 0)
  {
	sprintf(sSysEvent,"Error[%d]: Failed to Connect to UpNode[%s]!Reconnecting.",RetCode,UpNode->NodeCode);
	UpNode->WrSystemLog->WriteLog(sSysEvent,SYSTEMLOG);
	goto ReConnect;
  }
  Connect_delay = (float)CONNECT_WAIT;
  sprintf(sSysEvent,"LOG: TCP connection[%d] to UpNode[%s] established!",UpNode->TcpSock.sock,UpNode->NodeCode);
  UpNode->WrSystemLog->WriteLog(sSysEvent,SYSTEMLOG);
  
  //登陆到上节点,建立应用层连接
  RetCode = UpNode->NodeLogin(imode);
  if(RetCode)
  {
  	if(RetCode==1)
  	   sprintf(sSysEvent,"Error[%d]: Connection with UpNode[%s] lost while logining.",RetCode,UpNode->NodeCode);
       	else if(RetCode==2)
       	   sprintf(sSysEvent,"Error[%d]: Login_rsp from UpNode[%s] is abnormal!",RetCode,UpNode->NodeCode);
       	else if(RetCode==3)
       	   sprintf(sSysEvent,"Error[%d]: The Protocal Version UpNode[%s] supported is too low!",RetCode,UpNode->NodeCode);
      	else
      	{
      	   sprintf(sSysEvent,"Error[%d]: Failed to Login to UpNode[%s]!",RetCode,UpNode->NodeCode);
      	}
      	UpNode->WrSystemLog->WriteLog(sSysEvent,SYSTEMLOG);
	goto ReConnect;
  }
  sprintf(sSysEvent,"LOG: Login to UpNode[%s] successfully!",UpNode->NodeCode);
  UpNode->WrSystemLog->WriteLog(sSysEvent,SYSTEMLOG);
  
  int GotNum=0;
  char FileName[100];
  //---------从备份文件中导入接收队列数据---------
  FileOpr *RecvQFile = new FileOpr();	//用于从备份文件中获取队列数据的操作对象
  memset(FileName,0,sizeof(FileName));
  sprintf(FileName,"%s%s%s",UpNode->UpQueueBak,UpNode->SP_Id,UpNode->UpRecvQ_File);
  RecvQUnit * RUnit = RecvQFile->RecvQ_Restore(FileName,&GotNum);	//在函数中打开关闭文件，malloc空间
  //printf("\n UpNodeCom: RecvQFile GotNum=%d \n",GotNum);
  if (RUnit!=NULL)	//成功从备份文件中获取到队列数据
  {
	  RecvQUnit * pRUnit;
	  for(int i=0;i<GotNum;i++)
	  {
		pRUnit = (RecvQUnit *) ((char *)RUnit + i*sizeof(RecvQUnit));
		//printf("\n UpNodeCom: Loaded a CMPP pack[%x] for RecvQ, Sequence=[%x]. Packlen=%d.\n",ntohl(*(int *)(pRUnit->Pack+4)),ntohl(*(int *)(pRUnit->Pack+8)),ntohl(*(int *)pRUnit->Pack));
		
			/*/---------debug--------
			CMPP_Deliver *mydeliver = (CMPP_Deliver *)pRUnit->Pack;
			int len=ntohl(*(int *)mydeliver);
			printf("\n NodeCom: Deliver= ");
			for(int k=0;k<len;k++)
				printf("%02x ",*((unsigned char *)mydeliver+k));
			printf("\n");
			//---------debug end--------*/

		UpNode->RecvQ->Wait_Put(*pRUnit);
	  }
	  free(RUnit);
  }
  //else 没有从文件中读到数据,可能文件不存在或为空或者文件打开失败
  delete RecvQFile;
  sprintf(sSysEvent,"LOG: Got [%d] RecvQ Units from File[%s].",GotNum, FileName);
  UpNode->WrSystemLog->WriteLog(sSysEvent,SYSTEMLOG);

  //---------从备份文件中导入发送队列数据---------
  FileOpr *SendQFile = new FileOpr();	//用于从备份文件中获取队列数据的操作对象
  memset(FileName,0,sizeof(FileName));
  sprintf(FileName,"%s%s%s",UpNode->UpQueueBak,UpNode->SP_Id,UpNode->UpSendQ_File);
  SendQUnit * SUnit = SendQFile->SendQ_Restore(FileName,&GotNum);	//在函数中打开关闭文件，malloc空间
  //printf("\n UpNodeCom: SendQFile GotNum=%d \n",GotNum);
  if (SUnit!=NULL)	//成功从备份文件中获取到队列数据
  {
	  SendQUnit * pSUnit;
	  for(int i=0;i<GotNum;i++)
	  {
		pSUnit = (SendQUnit *) ((char *)SUnit + i*sizeof(SendQUnit));
		*(int *)(pSUnit->Pack+8) = htonl(UpNode->GetSequence());	//修改包流水号
		//printf("\n UpNodeCom: SendQUnit: packlen=%d, packtype=%x, SrcNode=%d. \n", ntohl(*(unsigned int *)pSUnit->Pack), ntohl(*(unsigned int *)(pSUnit->Pack+4)), pSUnit->SrcNode);
		UpNode->SendQ->Wait_Put(*pSUnit);
	  }
	  free(SUnit);
  }
  //else 没有从文件中读到数据,可能文件不存在或为空或者文件打开失败
  delete SendQFile;
  sprintf(sSysEvent,"LOG: Got [%d] SendQ Units from File[%s].",GotNum, FileName);
  UpNode->WrSystemLog->WriteLog(sSysEvent,SYSTEMLOG);

  //---------重置已发送队列的等待时间标识---------
  RetCode = UpNode->SentQReset();
  UpNode->State = 0;
  sprintf(sSysEvent,"LOG: [%d] Units in SentQ of UpNode[%s] are reseted!",RetCode, UpNode->NodeCode);
  UpNode->WrSystemLog->WriteLog(sSysEvent,SYSTEMLOG);
  
  pthread_attr_init(&ThreadAttr);
  pthread_attr_setscope(&ThreadAttr, PTHREAD_SCOPE_SYSTEM); /*设定线程属性为绑定*/
  /*
  //产生上节点接收线程
  RetCode = pthread_create(UpNode->ReadThr,&ThreadAttr,UpReceiver,(void *)UpNode);
  if(RetCode > 0)
  {
	sprintf(sSysEvent,"Error[%d]: Cann't create thread for UpNode[%s].UpReceiver()!",RetCode,UpNode->NodeCode);
	UpNode->WrSystemLog->WriteLog(sSysEvent,SYSTEMLOG);
  }
  */
  
  //产生上节点发送线程
  if(UpNode->WriteThr==0)
  {
	  RetCode = pthread_create(&UpNode->WriteThr,&ThreadAttr,UpNode->UpSender,UpNode);
	  if(RetCode)
	  {
		sprintf(sSysEvent,"Error[%d]: Cann't create thread for UpNode[%s].UpSender()!",RetCode,UpNode->NodeCode);
		UpNode->WrSystemLog->WriteLog(sSysEvent,SYSTEMLOG);
	  }
  }
  else
  {
	  #ifdef DEBUG
	  printf("\n UpNodeCom: UpSender's ThreadNum=[%d], UpSender won't be recreated!\n",(int)UpNode->WriteThr);
	  #endif
  }
  
  UpNode->UpReceiver((void *)UpNode);	//接收模块
  if(UpNode->ToExit)
  {
	  sprintf(sSysEvent,"LOG: UpNodeCom of UpNode[%s] is ended!",UpNode->NodeCode);
	  UpNode->WrSystemLog->WriteLog(sSysEvent,SYSTEMLOG);
	  return (void *)NULL;
  }
  if(UpNode->State==1) goto ReConnect;	//TCP连接已断,需重新连接/登陆
  else if(UpNode->State==2)
  {
  }
  else if(0)	//其他情况?应用层连接被关闭(收到Logout/LogoutRsp)
  {}
  
  sprintf(sSysEvent,"Warning: NodeCom of UpNode[%s] QUIT! ",UpNode->NodeCode);
  UpNode->WrSystemLog->WriteLog(sSysEvent,SYSTEMLOG);
  return (void *)NULL;
}