main.c

基于UNIX的中间件编程,只能运行于UNIX操作系统

			printf(">>network thread recieve from cm: error\n");
			m_log=SetLogContent(L_Error,"recieve from cm error","when network thread recieve \
							from the cm an error occred");
			writeLog(&m_log);
		}
		else
		{
			printf(">>network thread recieve from cm success:%s\n",buf);
			if(SeparatePacket(buf,&m_struct))
			{
				printf(">>network thread recieve from cm: error\n");
				m_log=SetLogContent(L_Error,"recieve from cm error","when network thread recieve \
				      from the cm an error occred,the packet lenght is less than 29");
				writeLog(&m_log);
			}
			strcpy(packettype,m_struct.packettype );
			oprcode=atoi(m_struct.oprcode  );
			switch(atoi( packettype))
			{
			case 0:
				/*message  is the request packet*/

				fsin.sin_port =htons(atoi(QueueMsg));	/*change to the queue thread socket*/

				if(!CallApplication(m_struct))
					SendClearQueueMsg(Qsock,fsin,alen,buf);
				break;
			case 1:
				/*message is the ACk packet*/
				break;	

			case 2:
			/*message is the data packet and sent it to the app's passive socket*/
				memset(&fsin,0,sizeof(fsin));
				fsin.sin_family =AF_INET;
				fsin.sin_addr.s_addr =inet_addr(myname.IP);
				fsin.sin_port =htons(atoi(AppSocket));
				sendto(Ssock,buf,strlen(buf),0,(struct sockaddr *)&fsin,sizeof(fsin));
				break;
			case 3:
			/*set the MAXSN to 0*/
				printf("start to clear the MAXSN...\n");
				if(strcpy(myname.IP ,remote_ser[1].RemoteIP)!=0)
					if(SetMaxSN(MaxSN,MAXCLIENTCOUNT,m_struct.source,atol(m_struct.sn)))
					{
						printf("set MAXSN to 0 error\n");
						m_log=SetLogContent(L_Error,"set MAXSN to zero error","when Netthread recieve the manage messgae to set the MAXSN to zero,an error occured");
						writeLog(&m_log);
					}
					else
						sendto(Msock,buf,strlen(buf),0,(struct sockaddr *)&fsin,alen);
				break;
			default:
				/*the error packet type*/
				printf(">>network thread recieved an erro packet type\n");
				break;
			}
		}
	}
}

/*
--------------------------------------------------------
LocalThread :this is the local thread .It accept the app's
request and send the nessage to the server
------------------------------------------------------------
*/
void *LocalThread(void *arg)
{
	WSLOGDATA m_log;
	char errorno[4];
	char RequestType[2];
	struct timeval z_time;
		
/* 
----------------------------------------------
  create the app--cm  passive socket
----------------------------------------------
*/

	SOCKET Msock;/*the app--cm passive port*/
	SOCKET Ssock;/*connect to the other cm's Network passive port*/
	SOCKET Qsock;/*connect to the other cm's queue passive port*/
	
	char temp[34];
	char *port;
	int retry;
	

	struct sockaddr_in to;/*the other cm's address structure*/
	struct sockaddr_in from;/*the application's address structure*/
	int from_len;
	int to_len;
	
	char buf[MAXLEN];
	int j;
	int cc;
	char destination[20];

	cc=1;
	retry=0;
	

	port=temp;
	z_time.tv_sec=0;
	z_time.tv_usec=300;
	itoa(myname.APPPORT,temp,10);
	Msock=passiveUDP(myname.IP ,port);
	if(Msock==INVALID_SOCKET)
	{
		printf("cannot create the app--cm socket\n");
		m_log=SetLogContent(L_Error,"cannot create the socket","in local thread when call the Passive UDP functin ,the app-cm socket cannot be created");
		writeLog(&m_log);
		return ;
	}
	else
	{
		printf(" create the app--cm socket successfully\n");
		m_log=SetLogContent(L_Success," create the socket"," in local thread when call the Passive UDP functin ,the app-cm socke  be created successfully");
		writeLog(&m_log);
	}

	itoa(myname.MYMAINPORT,temp,10);
	Ssock=connectUDP(remote_ser[1].RemoteIP,temp);
	if(Ssock==INVALID_SOCKET)
	{
		printf("local thread:cannot connect to the other cm's passivesocket\n");
		m_log=SetLogContent(L_Error,"local thread:cannot connect to the socket","when call the connect UDP functin ,the  server is no response");
		writeLog(&m_log);
		return ;
	}
	else
	{
		printf("local thread:connect to the cm's socket successfully\n");
		m_log=SetLogContent(L_Success,"local thread:connect server successfully","when call the connect UDP functin ,the connection be created successfully");
		writeLog(&m_log);
	}

	Qsock=connectUDP(remote_ser[1].RemoteIP,QueueMsg);
	if(Qsock==INVALID_SOCKET)
	{
		printf("local thread:cannot connect to the other cm's Queue passivesocket\n");
		m_log=SetLogContent(L_Error,"local thread:cannot connect to the socket","when call the connect UDP functin ,the  server is no response");
		writeLog(&m_log);
		return ;
	}
	else
	{
		printf("local thread:connect to the cm's socket successfully\n");
		m_log=SetLogContent(L_Success,"local thread:connect server successfully","when call the connect UDP functin ,the connection be created successfully");
		writeLog(&m_log);
	}

	/*
	----------------------------------------------
	send message to set the maxsn to 0,if server 
	no response ,then retry n times
	----------------------------------------------
	*/
	if(strcmp(remote_ser[1].RemoteIP,"0.0.0.0")!=0)
		while(retry<TRYCOUNT)
		{
			struct sockaddr_in fsin;
			int alen=sizeof(fsin);
			fsin.sin_family =AF_INET;
			fsin.sin_port =htons(myname.MYMAINPORT);
			fsin.sin_addr.s_addr=inet_addr(remote_ser[1].RemoteIP);
			if(!SendSetMaxSNMessage(Ssock,fsin,alen))
				break;
			retry++;
		}
	if(retry==TRYCOUNT)
	{
		printf("cannot set manage packet to set the esrver's MAXSN to 0\n");
		m_log=SetLogContent(L_Error,"cannot send the manage packet to set the server's MAXSN to 0","Maybe the server shutdown");
		writeLog(&m_log);
	}
	
	while(1)
	{
		memset(destination,'\0',sizeof(destination));
		memset(&from,0,sizeof(from));
		from_len=sizeof(from);
		to_len=sizeof(to);					
		memset(&to,0,sizeof(to));						

		memset(buf,'\0',sizeof(buf));
		printf(">>local thread waiting for accept application's request\n");
		cc=recvfrom(Msock,buf,sizeof(buf),0,(struct sockaddr *)&from,&from_len);

		if(cc<=0)/*when recieve 意外错误*/
		{
			printf(">>local thread recieve from application with error No:%d\n",WSAGetLastError());
			m_log=SetLogContent(L_Error,"recieve from app error","when local thread recieve from the app an error occred");
			writeLog(&m_log);
			continue;
		}
		else
		{
		/*replace the sequence NO with the TransactionID*/
			int PacketType;
			int position;
			char TempID[10];
			printf(">>local thread recieve from application: success\n");
			TransactionID++;
			printf("the transaction sequence is:%d\n",TransactionID);
			sprintf(TempID,"%010d",TransactionID);
			for(position=0;position<=9;position++)
				buf[position+2]=TempID[position];

			if(SeparatePacket(buf,&m_struct))
			{
				printf(">>local thread recieve from cm: error\n");
				m_log=SetLogContent(L_Error,"recieve from cm error","when local thread recieve \
				      from the app an error occred,the packet lenght is less than 29");
				writeLog(&m_log);
			}
			printf("recieved application's message:%s\n",m_struct.message );		
			
			strcpy(destination,FormateHostName(m_struct.destination,destination ));
			to.sin_family =AF_INET;		
			to.sin_port=htons(myname.MYMAINPORT);
			to.sin_addr.s_addr =inet_addr(destination);			
			
			PacketType=atoi(m_struct.packettype );
			strcpy(RequestType,m_struct.requesttype );
			j=atoi(RequestType);

			if(PacketType==1)
				to.sin_port=htons(atoi(QueueMsg));
			else
				to.sin_port=htons(myname.MYMAINPORT);

			switch(j)
			{
				case 1:
				/*file transmit and not waiting for ACK*/				

				case 3:
				/*message transmit and not waiting for ACK*/
					SendNotWait(Msock,Ssock,Qsock,buf,from,sizeof(from),to,sizeof(to),PacketType);						
					break;

				case 2:
				/*file transmit and waitiong for ACK*/
				case 4:
				/*message transmit and waitong for ACK*/
					SendWait(Msock,Ssock,Qsock,buf,from,sizeof(from),to,sizeof(to),PacketType);
					break;
					
				default:
				/*error request type*/
					sprintf(errorno,"%04d0",E_Other);
					strncpy(buf+29,errorno,5);
					printf(">>local thread recieved the error reequest type from app\n");
					sendto(Msock,buf,strlen(buf),0,(struct sockaddr *)&from,sizeof(from));
					break;
			}/*end switch		*/
		}/*end else*/
	}/*end while*/
}

/*
--------------------------------------------------------
sm_LocalThread :this is the local thread .It accept the app's
request and send the nessage to the server(use the shared
memeory)
------------------------------------------------------------
*/
void *SM_LocalThread(void * arg)
{
	WSLOGDATA m_log;
	char errorno[4];
	char RequestType[2];
	struct 	timeval z_time;
	SOCKET Ssock;/*connect to the other cm's Network passive port*/
	SOCKET Qsock;/*connect to the other cm's queue passive port*/
	char temp[34];
	char *port;
	struct sockaddr_in to;/*the other cm's address structure*/
	struct sockaddr_in from;/*the application's address structure*/
	int from_len;
	int to_len;
	char *buf;
	char tempbuf[MAXLEN+1];
	int j;
	int cc;
	char destination[20];

	sm_TID mytid;
	int partions;
	char *packettype;
	char *sn;
	char *requesttype;
	char *oprcode;
	char *source;	
	char *TIDdestination;
	char *service;	
	partions=2;

	packettype="02";
	sn="0000000000";
	requesttype="04";
	oprcode="000007";
	source="202.115.17.134";	
	TIDdestination="202.115.17.134";
	service="8000";	

	memset(&mytid,'\0',sizeof(mytid));
	sprintf(mytid.packettype,"%s",packettype);
	sprintf(mytid.ID ,"%s",sn);	
	sprintf(mytid.requesttype,"%s",requesttype);
	sprintf(mytid.oprcode ,"%s",oprcode);
	mytid.partionnumbers=partions;
	sprintf(mytid.source,"%s",source);
	sprintf(mytid.destination ,"%s",TIDdestination);
	mytid.appport=atoi(service);
	mytid.timer =3;



	z_time.tv_sec=0;
	z_time.tv_usec=300;
	cc=1;
	port=temp;
	

/* 
----------------------------------------------
  create the app--cm  passive socket
----------------------------------------------
*/
	
	itoa(myname.MYMAINPORT,temp,10);
	Ssock=connectUDP(remote_ser[1].RemoteIP,temp);
	if(Ssock==INVALID_SOCKET)
	{
		printf("local thread:cannot connect to the other cm's passivesocket\n");
		m_log=SetLogContent(L_Error,"local thread:cannot connect to the socket","when call the connect UDP functin ,the  server is no response");
		writeLog(&m_log);
		return ;
	}
	else
	{
		printf("local thread:connect to the cm's socket successfully\n");
		m_log=SetLogContent(L_Success,"local thread:connect server successfully","when call the connect UDP functin ,the connection be created successfully");
		writeLog(&m_log);
	}

	Qsock=connectUDP(remote_ser[1].RemoteIP,QueueMsg);
	if(Qsock==INVALID_SOCKET)
	{
		printf("local thread:cannot connect to the other cm's Queue passivesocket\n");
		m_log=SetLogContent(L_Error,"local thread:cannot connect to the socket","when call the connect UDP functin ,the  server is no response");
		writeLog(&m_log);
		return ;
	}
	else
	{
		printf("local thread:connect to the cm's socket successfully\n");
		m_log=SetLogContent(L_Success,"local thread:connect server successfully","when call the connect UDP functin ,the connection be created successfully");
		writeLog(&m_log);
	}
	
	buf=tempbuf;	
	while(1)
	{

		memset(destination,'\0',sizeof(destination));
		memset(&from,0,sizeof(from));
		from_len=sizeof(from);
		to_len=sizeof(to);					
		memset(&to,0,sizeof(to));						
		memset(buf,'\0',sizeof(buf));
		printf(">>Shared memroy local thread waiting for accept application's request\n");

		SemaphoreP(InputSemaphoreID,atoi(mytid.oprcode));
		sm_RecieveMsg(1,&buf,sizeof(buf),NULL,&mytid);

		if(mytid.errorno!=0)
		{
			printf("SM_LocalThread recieve from shm err:[%d}\n",GetLastError());
		}
		else
		{
		/*replace the sequence NO with the TransactionID*/
			int PacketType;
			int position;
			char TempID[10];

			printf(">>Shm local thread recieve from application: success:[%s]\n",buf);
			TransactionID++;
			printf("the transaction sequence is:%d\n",TransactionID);
			sprintf(TempID,"%010d",TransactionID);
			for(position=0;position<=9;position++)
				buf[position+2]=TempID[position];
printf("add the sn's content:[%s]\n",buf);
			if(SeparatePacket(buf,&m_struct))
			{
				printf(">>local thread recieve from cm: error\n");
				m_log=SetLogContent(L_Error,"recieve from cm error","when local thread recieve \
				      from the app an error occred,the packet lenght is less than 29");
				writeLog(&m_log);
			}
			printf("the sn:%s\n",m_struct.sn);
			printf("the packettye:%s\n",m_struct.packettype);
			printf("the source:%s\n",m_struct.source);
			printf("the requesttype:%s\n",m_struct.requesttype);
			printf("the opecode:%s\n",m_struct.oprcode);
			printf("recieved application's message:%s\n",m_struct.message );		
			strcpy(destination,FormateHostName(m_struct.destination,destination ));
			printf("the destination:%d.%d.%.d.%d\n",m_struct.destination[0],m_struct.destination[1],m_struct.destination[2],m_struct.destination[3]);

			to.sin_family =AF_INET;		
			to.sin_port=(myname.MYMAINPORT);
			to.sin_addr.s_addr =inet_addr(destination);			
			
			PacketType=atoi(m_struct.packettype );
			strcpy(RequestType,m_struct.requesttype );
			j=atoi(RequestType);
			
			switch(j)
			{
				case 1:
				/*file transmit and not waiting for ACK*/				
				case 3:
				/*message transmit and not waiting for ACK*/
					sm_SendNotWait(Ssock,Qsock,buf,from,sizeof(from),to,sizeof(to),PacketType);						
					break;

				case 2:
				/*file transmit and waitiong for ACK*/
				case 4:
				/*message transmit and waitong for ACK*/
					sm_SendWait(Ssock,Qsock,buf,from,sizeof(from),to,sizeof(to),PacketType);
					break;
					
				default:
				/*error request type*/
					sprintf(errorno,"%04d0",E_Other);
					strncpy(buf+29,errorno,5);
					printf(">>local thread recieved the error reequest type from app\n");
					break;
			}/*end switch*/
		}/*