upsender.cpp

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

/**********************************************************************
  FileName            : UpSender.cpp
  Description         : 上节点发送线程模块
  Version             : 1.0
  Date                : 2003年6月17日
  Author              : 刘荣辉
  Other               : 
***********************************************************************/

#include "UpNode.h"

void * CUpNode::UpSender(void *node)
{
  int threadid;
  char sSysEvent[SYS_EVENT_LEN];	//记录系统事件的字符串
  SendQUnit  SUnit;
  int PackLen;
  int iSent;
  time_t nowtime;
  char timebuf[16];
  //int SendFail;
  CUpNode *UpNode; 
  unsigned int PackSeq;
  
  UpNode = (CUpNode *)node;
  threadid = pthread_self();
  pthread_detach(threadid);
  sprintf(sSysEvent,"LOG: UpSender[%d] for UpNode[%s] is started.",threadid,UpNode->NodeCode);
  UpNode->WrSystemLog->WriteLog(sSysEvent,SYSTEMLOG);    //写系统日志
  
//sleep(10000);goto UpSenderEnd;

  while(1)	//不断从发送队列中取出数据包进行发送
  { 
		//UpNode->SendQ->Wait_Get(SUnit);	//阻塞等待获取发送队列单元,不能及时响应退出命令
		if(UpNode->SendQ->Get(SUnit)==false)	//发现队列为空
		{
			if(UpNode->ToExit && UpNode->State)	//系统收到了退出命令，且连接已断
			{
				UpNode->State=2;
				break;	//连接已断，退出发送线程
			}
			usleep(UpNode->Send_Q_Empty_Wait);
			#ifdef DEBUG
			//printf("\nUpSender:::SendQ is empty,Send_Q_Empty_Wait=[%d]......\n",UpNode->Send_Q_Empty_Wait);
			#endif
			continue;
		}
		//-----------------------------------------
		if((UpNode->ShowLog==3) || (UpNode->ShowLog==2))	//显示正在发送的Submit包
		{			
			  //-----------------debug----------------------
			  unsigned char *ptmp;
			  ptmp=(unsigned char *)SUnit.Pack;
			  int len;
			  len = ntohl(*(int *)SUnit.Pack);
			  if(len>12)
			  {
				  /*printf("\n=======UpSender: checking a Pack to be sent======= \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");
				  for(int i=0;i<len;i++)
				  {   
					 printf("%d ",*(ptmp+i));
					 if(((i+1)%4)==0) printf("  ");
					 if((i+1)==12) printf("\n");
				  }
				  printf("\n=========================================== \n");*/
			  }
			  //--------------------debug end-------------------/			
			if(UpNode->Protocol==0)
			{
				CMPP_Submit * tReqPack = (CMPP_Submit *)SUnit.Pack;
				#ifdef DEBUG
				UpNode->WrSystemLog->getTime(timebuf);
				if(ntohl(tReqPack->Head.Command_Id) == CMPP_SUBMIT)
				{
					printf("\n================正向网关发CMPP_Submit包==================\n");
					printf("发送时间:%s\n用户手机号：%s\n源号码：%s\n短信内容：%s\n",timebuf,tReqPack->Dest_terminal_Id,tReqPack->Src_Id,tReqPack->Msg_Content);
				}
				else if(ntohl(tReqPack->Head.Command_Id) == CMPP_ACTIVE_TEST)
				{
					//printf("\n UpSender[%s]: Sending an Active_Test pack. \n",timebuf);
				}
				else if(ntohl(tReqPack->Head.Command_Id) == CMPP_ACTIVE_TEST_RESP)
				{
					printf("\n UpSender[%s]: Sending an Active_Test_Rsp pack. \n",timebuf);
				}
				else if(ntohl(tReqPack->Head.Command_Id) == CMPP_DELIVER_RESP)
				{
					printf("\n UpSender[%s]: Sending a CMPP_Deliver_Rsp pack. \n",timebuf);
				}
				#endif
			}
			else if(UpNode->Protocol==1)
			{
				SMGP_Submit * tReqPack = (SMGP_Submit *)SUnit.Pack;
				#ifdef DEBUG
				UpNode->WrSystemLog->getTime(timebuf);
				if(ntohl(tReqPack->msgHead.nRequestId) == REQUEST_ID_SUBMIT)
				{
					printf("\n================正向网关发SMGP_Submit包==================\n");
					printf("发送时间:%s\n用户手机号：%s\n源号码：%s\n短信内容：%s",timebuf,tReqPack->sDestTermId,tReqPack->sReplyPath,tReqPack->sMsgContent);
				}
				else if(ntohl(tReqPack->msgHead.nRequestId) == REQUEST_ID_ACTIVE_TEST)
				{
					//printf("\n UpSender[%s]: Sending a SMGP_Active_Test pack. \n",timebuf);
				}
				else if(ntohl(tReqPack->msgHead.nRequestId) == REQUEST_ID_DELIVER_RESP)
				{
					printf("\n-------- UpSender[%s]: Sending a SMGP_DELIVER_RESP pack. \n",timebuf);
				}
				#endif
			}
		}
		//--------------------发送数据包---------------------
		//SendFail = 0;
		PackLen = htonl(*(int *)SUnit.Pack);
		if(UpNode->State)
		{
			if(UpNode->ToExit)
				UpNode->State=2;
			UpNode->SendQ->Put(SUnit);	//放回发送队列以便备份数据
			break;
		}
		iSent = UpNode->TcpSock.Writen(SUnit.Pack, PackLen);	//发送数据包
		if(iSent!=PackLen)	//发送失败，连接异常
    	{
			#ifdef DEBUG
			printf("\n UpSender: Send failed! PackLen=[%d],iSent=[%d] \n",PackLen,iSent);
			#endif
           	//SendFail = 1;	//连接异常,发送线程准备退出
			if(UpNode->State==0)
			{
				if(!UpNode->ToExit)
				{
					UpNode->State = 1;	//节点状态为正在连接
					UpNode->TcpSock.Close_sock();
				}
				else	//收到系统退出命令
					UpNode->State = 2;
				UpNode->SendQ->Put(SUnit);	//放回发送队列以便重发或者备份数据
			}

			if(!UpNode->ToExit)
			{
				sprintf(sSysEvent,"Error[%d]: Connection with UpNode[%s] lost! UpSender is preparing to exit!",iSent,UpNode->NodeCode);
				UpNode->WrSystemLog->WriteLog(sSysEvent,SYSTEMLOG);
			}
			break;	//退出发送线程
    	}
    	//else printf("\n=====================发送成功=======================\n");
    	
    	//-------------------放进已发送队列/删包----------------------		
		PackSeq=ntohl(*(unsigned int *)(SUnit.Pack + 8));
		#ifdef DEBUG
		//printf("\n UpSender: Sequence=[%d],iResent=[%d],SendFail=[%d] \n",PackSeq,SUnit.iResent,SendFail);
    	#endif
		//if( (SUnit.iResent>0) || SendFail)	//需要等应答或者发送失败，则放进已发送队列
		if(SUnit.iResent>0)	//已发送成功的请求包（需要等应答），则放进已发送队列
    	{
    		/*if(UpNode->ToExit)
			{
				UpNode->SendQ->Put(SUnit);	//放回发送队列以便备份数据
				break;
			}
			else
			{*/
				//修改队列单元标识
				time(&nowtime);
				if(UpNode->Protocol==0)
					SUnit.SendTime = nowtime + CMPP_SEND_TIMEOUT;	//下一次发送时间(重发时间)
				else if(UpNode->Protocol==1)
					SUnit.SendTime = nowtime + SEND_TIMEOUT;	//下一次发送时间(重发时间)
				if(SUnit.iResent>0)
				{
					SUnit.iResent--;
					//放入已发送队列,(流水号--队列单元指针,映射哈希表)
					UpNode->SentQ.Add(PackSeq , SUnit);
					#ifdef DEBUG
					//printf("\n UpSender: A pack[%d = %x] is added to SentQ after sending! \n", PackSeq, PackSeq);
					#endif
				}
			//}
		}
		//else if( SUnit.iResent==-1 && SendFail==0)	//对于成功发送的应答包
		else if(SUnit.iResent==-1)	//对于成功发送的应答包(应答包只被发送一次便删除)
		{
			free(SUnit.Pack);	//删除包,释放内存
			#ifdef DEBUG
			if(UpNode->Protocol==0)
				printf("\n UpSender: A CMPP_Rsp pack[%x] is delete after sent successfully! \n", PackSeq);
			else if(UpNode->Protocol==1)
				printf("\n UpSender: A SMGP_Rsp pack[%x] is delete after sent successfully! \n", PackSeq);
			#endif
		}

		if(UpNode->State && UpNode->ToExit)	//系统收到了退出命令，且连接已断
		{
			UpNode->State=2;
			break;
		}
    	/*if(SendFail && UpNode->State)	//最后一次发送失败，且目前仍连接异常,发送线程退出
    	{
	    	#ifdef DEBUG
	    	printf("\n UpSender: EXIT! \n");
	    	#endif
    		break;
    	}*/
  }//while(1)
  
//UpSenderEnd:
  UpNode->WriteThr = 0;
  if(UpNode->ToExit)
  {
	UpNode->State = 2;	//节点状态为暂时不再连接
	sprintf(sSysEvent,"LOG: UpSender of UpNode[%s] is ended!",UpNode->NodeCode);
  }
  else
	sprintf(sSysEvent,"Warning: UpSender of UpNode[%s] QUIT!",UpNode->NodeCode);
  UpNode->WrSystemLog->WriteLog(sSysEvent,SYSTEMLOG);

  return (void *)NULL;
}