socket.c

CBS（小区广播）程序实现手稿

#include "..\sys\Include.h"

int WaitTimer(int nSocketId,long TimeSec,long TimeUsec);


void ReceCBEMessage();

int initsocket()
{
	struct sockaddr_in server;
	int errorNo;
	int s;
	WSADATA ws;

	WSAStartup(MAKEWORD(1,1),&ws);
	if ((s=socket(AF_INET, SOCK_STREAM, 0))<0)
	{
        errorNo=WSAGetLastError();
		perror("Socket()");
		return -1;
	}

	memset(&server,0,sizeof(server));
	server.sin_family = AF_INET;
	server.sin_port = htons(USERPORT);
	server.sin_addr.s_addr = INADDR_ANY;
	if (bind(s, (struct sockaddr *)&server, sizeof(server))<0)
	{
		perror("Bind()");
		return -1;
	}
	if (listen(s,5)!=0)
	{
		perror("Listen()");
		return -1;
	}

	printf("Socket register successfully.\n");
	return s;
}

void ListenUser(int *sockethandle)
{
	struct sockaddr_in client;
    int namelen;
	int ns;
	int s1;

    s1=*sockethandle;
	namelen = sizeof(client);
	while (1)
	{
    memset(&client,0,namelen);
	if ((ns = accept(s1, (struct sockaddr *)&client, &namelen)) ==-1)
	{
		perror("Accept()");
        break;  
	}
	else
	{
		if (s_RuntimeInfo.CBCProxyConnectStatus==SOCKET_DISCONNECTED)
		{
			printf("Socket Accept!\n");
			s_RuntimeInfo.CBCProxySocketHandle=ns;
			s_RuntimeInfo.CBCProxyConnectStatus=SOCKET_CONNECTED;
        	CreateThread((LPSECURITY_ATTRIBUTES)NULL,0,(LPTHREAD_START_ROUTINE)ReceCBEMessage,NULL,0,NULL);
	
		}
		else
		{
			printf("Socket already Connect,New Connection will stop!\n");
			closesocket(ns);
		}
	}
	}
	ExitThread(1);    
}



int RecvData(int nSocketId, char *sBuf )
{
    int nRecvLen = -1 ;
	int received = 0,total = 0 ;

#ifdef  OPEN_WARNING
	/* 告警描述字串 */
	unsigned char  cWarningDescription[WARNING_DESC_LEN];
#endif
  	
	if (( nRecvLen = recv( nSocketId ,sBuf , PACKHEADLEN , 0) ) == PACKHEADLEN ) 
    {
		nRecvLen = *(short *)sBuf;

		if ( nRecvLen > MAXDATALEN || nRecvLen <= 0 ) 
		{
#ifdef  OPEN_WARNING
			/* 告警 */
			memset(cWarningDescription,0,WARNING_DESC_LEN);
			sprintf(cWarningDescription,"IP包长错误,Len=%d",nRecvLen);
			SendWarningToOmp(LINK_WARNING,ERROR_PACK,cWarningDescription);			
#endif
			/* 使告警有可能发出去 */
			Sleep(2000);

			printf("packet length is too long or invalid.\r\n"); 
			return -1 ;                //actual usage
		}
    
		while(1)
		{
			if (WaitTimer(nSocketId,1,0)==FAILURE)
				return FAILURE;
			received = recv( nSocketId , sBuf+total+PACKHEADLEN , nRecvLen-total , 0 ) ;

			if (received <= 0) return -1;
			total += received ;
			if (total  >= nRecvLen) break;
		}
		
		return nRecvLen ;

	}

	return -1 ; 
}

int WaitTimer(int nSocketId,long TimeSec,long TimeUsec)
{
	fd_set FDSet;
	TIMEVAL TimeVal;
  	
	TimeVal.tv_sec = TimeSec;
	TimeVal.tv_usec = TimeUsec;

	FD_ZERO(&FDSet);
	FD_SET(nSocketId,&FDSet);

	if (select(0,&FDSet,NULL,NULL,&TimeVal)<=0)
		return FAILURE;
	return SUCCESS;
}

/* 将消息加入接收消息队列中	*/ 
void AppendCBERecvMsg(CBEMESSAGE_t *s_CBEMessage)
{
#ifdef  OPEN_WARNING
	unsigned char cWarningDescription[WARNING_DESC_LEN];
#endif

	if(Rece_CBE_Buf.iMsgCount==MSG_QUEUE_SIZE)
	{
#ifdef  OPEN_WARNING
		/* 告警 */
		memset(cWarningDescription,0,sizeof(WARNING_DESC_LEN));
     	sprintf(cWarningDescription,"接收消息队列已满");
    	SendWarningToOmp(LINK_WARNING,IP_INPUT_QUEUE,cWarningDescription);
#endif
	}

	WaitForSingleObject(Rece_CBE_Buf.MsgEmpty,INFINITE);
	WaitForSingleObject(Rece_CBE_Buf.MsgMutex,INFINITE);
	
	memcpy(&Rece_CBE_Buf.s_Message[Rece_CBE_Buf.iTail],s_CBEMessage,sizeof(CBEMESSAGE_t));

	if (++Rece_CBE_Buf.iTail==MSG_QUEUE_SIZE) 
	{
		Rece_CBE_Buf.iTail = 0;
		Rece_CBE_Buf.iMsgCount++;
	}
	ReleaseMutex(Rece_CBE_Buf.MsgMutex);
	ReleaseSemaphore(Rece_CBE_Buf.MsgFull,1,NULL);

}

/* 从接收消息缓冲区中取出一条消息信息 */
void GetCBERecvMsg(CBEMESSAGE_t *NewCBEMsg)
{
	WaitForSingleObject(Rece_CBE_Buf.MsgFull,INFINITE);
	WaitForSingleObject(Rece_CBE_Buf.MsgMutex,INFINITE);
	memcpy(NewCBEMsg, &Rece_CBE_Buf.s_Message[Rece_CBE_Buf.iHead],sizeof(CBEMESSAGE_t));
	if (++Rece_CBE_Buf.iHead==MSG_QUEUE_SIZE) 
	{
		Rece_CBE_Buf.iHead = 0;
		Rece_CBE_Buf.iMsgCount--;
	}
	ReleaseMutex(Rece_CBE_Buf.MsgMutex);
	ReleaseSemaphore(Rece_CBE_Buf.MsgEmpty,1,NULL);
}

/* 将消息加入发送消息队列中	*/
void AppendCBESendMessage(CBEMESSAGE_t *s_CBEMessage)
{

	WaitForSingleObject(Send_CBE_Buf.MsgEmpty,INFINITE);
	WaitForSingleObject(Send_CBE_Buf.MsgMutex,INFINITE);
	
	memcpy(&Send_CBE_Buf.s_Message[Send_CBE_Buf.iTail],s_CBEMessage,s_CBEMessage->iLen+2);

	if (++Send_CBE_Buf.iTail==MSG_QUEUE_SIZE) 
	{
		Send_CBE_Buf.iTail = 0;
		Send_CBE_Buf.iMsgCount++;
	}
	ReleaseMutex(Send_CBE_Buf.MsgMutex);
	ReleaseSemaphore(Send_CBE_Buf.MsgFull,1,NULL);

}

/* 从发送消息缓冲区中取出一条消息信息 */
void GetCBESendMsg(CBEMESSAGE_t *NewCBEMsg)
{
	WaitForSingleObject(Send_CBE_Buf.MsgFull,INFINITE);
	WaitForSingleObject(Send_CBE_Buf.MsgMutex,INFINITE);
	memcpy(NewCBEMsg, &Send_CBE_Buf.s_Message[Send_CBE_Buf.iHead],sizeof(CBEMESSAGE_t));
	if (++Send_CBE_Buf.iHead==MSG_QUEUE_SIZE) 
	{
		Send_CBE_Buf.iMsgCount--;
		Send_CBE_Buf.iHead = 0;
	}
	ReleaseMutex(Send_CBE_Buf.MsgMutex);
	ReleaseSemaphore(Send_CBE_Buf.MsgEmpty,1,NULL);
}

/* 接收CBE消息 */
void ReceCBEMessage()
{
//	int sHanlde;
	CBEMESSAGE_t s_CBEMessage;

	memset(&s_CBEMessage,0,sizeof(CBEMESSAGE_t));

//	sHanlde=*sockethandle;

	for (;;)
	{
		if (RecvData(s_RuntimeInfo.CBCProxySocketHandle,(char *)&s_CBEMessage)<0)
		{
			break;
		}
//		s_CBEMessage.SocketHandle=sHanlde;
		AppendCBERecvMsg(&s_CBEMessage);
	    memset(&s_CBEMessage,0,sizeof(CBEMESSAGE_t));
	}
	closesocket(s_RuntimeInfo.CBCProxySocketHandle);
    s_RuntimeInfo.CBCProxyConnectStatus=SOCKET_DISCONNECTED;
	printf("Client Connect exit.\n");
	ExitThread(1);    
}

/* 发送函数 */
int SendData(int nSocketId,unsigned char *sBuf,int nLen )
{
	fd_set FDSet;
	TIMEVAL TimeVal;
  	
	TimeVal.tv_sec = 10;
	TimeVal.tv_usec = 0;

	FD_ZERO(&FDSet);
	FD_SET(nSocketId,&FDSet);
	if (select(0,NULL,&FDSet,NULL,&TimeVal)<=0)  return -1;
	return  send(nSocketId, sBuf, nLen, 0);
}


//发送CBE消息
void SendCBEMessage()
{
	CBEMESSAGE_t s_CBEMessage;
//	int nSocketId;
	
	memset(&s_CBEMessage,0,sizeof(CBEMESSAGE_t));

	for (;;)
	{
		GetCBESendMsg(&s_CBEMessage);
//		nSocketId=s_CBEMessage.SocketHandle;
		
		if (SendData(s_RuntimeInfo.CBCProxySocketHandle,(char *)&s_CBEMessage,(s_CBEMessage.iLen+PACKHEADLEN) )<0)
		{
			closesocket(s_RuntimeInfo.CBCProxySocketHandle);

			s_RuntimeInfo.CBCProxyConnectStatus=SOCKET_DISCONNECTED;
		}
	    memset(&s_CBEMessage,0,sizeof(CBEMESSAGE_t));
	}

//	printf("Client Connect exit.\n");
//	ExitThread(1);    
}

//CBCtoCBE
void ReceCBCMsg(int *sockethandle)
{
	struct sockaddr_in client;
    int namelen;
	int ns;
	int s1;

    s1=*sockethandle;
	namelen = sizeof(client);
	while (1)
	{
		memset(&client,0,namelen);
	if ((ns = accept(s1, (struct sockaddr *)&client, &namelen)) ==-1)
	{
		perror("Accept()");
        break;  
	}
	else
	{
		printf("Socket Accept!\n");
    	CreateThread((LPSECURITY_ATTRIBUTES)NULL,0,(LPTHREAD_START_ROUTINE)ReceCBEMessage,&ns,0,NULL);
	}
	}
	ExitThread(1);    
}

//客户端Socket
int TcpInit()
{
    WSADATA ws;
    return WSAStartup(MAKEWORD(1,1),&ws);
}

int OpenTcp()
{
    return socket(AF_INET, SOCK_STREAM,0);
}

int SetSocketBlock(int nId , int nFlag)
{
    static long nBlock  =0;
	static long nNoBlock=1;

    if ( nFlag == 1 )
         return ioctlsocket(nId,FIONBIO,&nBlock);
	else
         return ioctlsocket(nId,FIONBIO,&nNoBlock);
}

int ConnectSocket(int nSocketId , int nIp,int nPort)
{
    struct sockaddr_in rServer;

    memset(&rServer , 0 , sizeof(rServer));
    rServer.sin_family=AF_INET;
    rServer.sin_port=htons((short)nPort);
    rServer.sin_addr.s_addr= nIp;
    return connect(nSocketId,(struct sockaddr *)&rServer,sizeof(rServer));
}

int ConnectToServer( char *sAddr , int nPort )
{
	int nSocketId ;

	TcpInit();

	if ( (nSocketId = OpenTcp()) <= 0 ) 
	{
		printf(" fgdf register socketId Fail ....\n");
		return -1 ;
	}
	 
  	if ( ConnectSocket(nSocketId ,inet_addr( sAddr ) ,nPort) == 0 )
	{

		SetSocketBlock( nSocketId  , 1);	//阻塞方式
		printf("socket connect Succ ...\n");

		return nSocketId ;
	}
	return -1;
}