listen_thread.c

一个C程序

#include "emv_global.h"
#include "server_info.h"

#include "listen_thread.h"

int giUdpSocket = -1;

static void *udp_listen_thread(void *arg);

int init_udp_server(void)
{
	giUdpSocket = udp_server(g_port_arppoll_udp);
	if (giUdpSocket < 0) {
		emv_error("udp_server(%d) failed", g_port_arppoll_udp);
		return -1;
	}

	return 0;
}

int start_listen_thread(void)
{
	pthread_t tid;
	int retcode;

	retcode = pthread_create(&tid, NULL, udp_listen_thread, NULL);
	if (retcode != 0) {
		emv_error("pthread_create failed: %s", strerror(retcode));
		return -1;
	}
	
	return 0;
}

static void *udp_listen_thread(void *arg)
{
	char buf[1024*64] = "\0";
	char sPeerIp[MAX_IP_LEN] = "\0";
	int buflen;
	int bodylen;
	int sid, did;
	int retcode;
	unsigned short nPeerPort;

	pthread_detach(pthread_self());
	emv_info("消息接收线程启动");
	
	while (giRunStatus != STOPPING) {
		sleep_time(0, THREAD_SCHEDULE_TIME);
		sched_yield();

		memset(sPeerIp, 0, sizeof(sPeerIp));
		nPeerPort = -1;
		buflen = sizeof(buf);
		memset(buf, 0, buflen);

		retcode = udp_recvfrom(giUdpSocket, sPeerIp, &nPeerPort, buf, &buflen);
		if (retcode < 0) {
			emv_error("udp_recvfrom failed");
			continue;
		}
		
		emv_debug("received msg from [%s:%d], buflen[%d]", sPeerIp, nPeerPort, buflen);
		if (buflen < sizeof(MsgHead)) {
			emv_warning("msg len is not valid: msglen=%d", buflen);
			continue;
		}
		
		debug_msg(buf);
		bodylen = get_msg_body_len(buf);
		if (buflen != sizeof(MsgHead) + bodylen) {
			emv_warning("收到的消息包的长度(%lu)不正好"
				"是一个消息的长度(%lu)",
				buflen, sizeof(MsgHead) + bodylen);
			continue;
		}

		did = get_msg_did(buf);
		sid = get_msg_sid(buf);

		if (get_major_flag()) {
			if (sid == SERVER_ID_APP && did == SERVER_ID_ARPPOLL) {
			} else if (sid == SERVER_ID_ARPPOLL && did == SERVER_ID_APP) {
			} else {
				emv_error("本机为主采集机, 错误的消息格式(sid:%d, did:%d)", 
					sid, did);
				continue;
			}
		} else {
			if (sid != SERVER_ID_APP && did != SERVER_ID_ARPPOLL) {
				emv_error("本机为从采集机, 错误的消息格式(sid:%d, did:%d)", 
					sid, did);
				continue;
			}
		}

		recv_msg_input(sPeerIp, nPeerPort, buf, buflen);
	}
	
	emv_info("消息接收线程终止");

	return NULL;
}

int udp_server(unsigned short port)
{
	int socketfd;
	struct sockaddr_in sin;
	int recvbuf_opt = 1024*64;
	#if 0
	int reuse = 1;
	#endif
	int retcode;

	socketfd = socket(PF_INET,SOCK_DGRAM,0);
	if(socketfd < 0) {
		emv_error("socket failed: %s", strerror(errno));
		return -1;
	}

	bzero((char*)&sin, sizeof(sin));
	sin.sin_family = AF_INET;
	sin.sin_addr.s_addr = htonl(INADDR_ANY);
	sin.sin_port = htons(port);
	#if 0
	retcode = setsockopt(socketfd, SOL_SOCKET, SO_REUSEADDR, &reuse, sizeof(reuse));
	if (retcode < 0) {
		emv_error("setsockopt failed: %s", strerror(errno));
		close(socketfd);
		return -1;
	}
	#endif

	retcode = bind(socketfd, (struct sockaddr *)&sin, sizeof(sin));
	if (retcode < 0) {
		emv_error("bind failed: %s", strerror(errno));
		close(socketfd);
		return -1;
	}

	retcode = setsockopt(socketfd, SOL_SOCKET, SO_RCVBUF, &recvbuf_opt, sizeof(recvbuf_opt));
	if (retcode < 0) {
		emv_error("setsockopt failed: %s", strerror(errno));
		close(socketfd);
		return -1;
	}

	return socketfd;
}

int udp_client(void)
{
	int sd;
	sd = socket(PF_INET, SOCK_DGRAM, 0);
	if (sd < 0) {
		emv_error("socket failed: %s", strerror(errno));
		return -1;
	}
	return sd;
}


int udp_recvfrom(int socketfd,char *ipaddr,unsigned short *pport,void *data,int *pdatalen)
{
	struct sockaddr_in peer;
	int addrlen = 0;
	int readbytes = 0;
	char buf[1024 * 8];
	char *peeraddr = 0;
	
	//saint check
	if(socketfd < 0)
	  return -1;
	
	//if parameters are null,return FAILURE
	if(!pdatalen)
	{
		return -1;
	}
	if(!data)
	  return -1;
	
	addrlen = sizeof(struct sockaddr_in);
	readbytes = recvfrom(socketfd,buf,1024 * 8,0,(struct sockaddr *)&peer,(socklen_t*)&addrlen);
	if(readbytes < 0) {
		emv_error("recvfrom failed: %s", strerror(errno));
		return -1;
	}
	
	peeraddr = inet_ntoa(peer.sin_addr);
	if(ipaddr && peeraddr)
		strcpy(ipaddr,peeraddr);
	
	if(pport)
		*pport = ntohs(peer.sin_port);
	
	*pdatalen = readbytes;
	memcpy(data,buf,*pdatalen);
	
	return(*pdatalen);
}


int udp_sendto(int socketfd,char *ipaddr,unsigned short port,void *data,int datalen)
{
	struct sockaddr_in peer;
	int addrlen = 0,writebytes = 0;
	
	if((socketfd < 0) || (!ipaddr) || (!data) || (datalen < 0)) {
		return -1;
	}
	
	bzero((char*)&peer, sizeof(peer));
	peer.sin_family = AF_INET;
	if((peer.sin_addr.s_addr = inet_addr(ipaddr)) == -1) {
		return -1;
	}
	peer.sin_port = htons(port);
	
	addrlen = sizeof(peer);
	writebytes = sendto(socketfd,data,datalen,0,(struct sockaddr *)&peer,addrlen);
	if(writebytes < 0)
		return -1;
	else
		return writebytes;
}

int udp_send(char *ipaddr, unsigned short port, void *data, int datalen)
{
	int sd;
	int retcode;

	sd = udp_client();
	if (sd < 0) {
		emv_error("udp_client failed");
		return -1;
	}

	retcode = udp_sendto(sd, ipaddr, port, data, datalen);
	close(sd);

	return retcode;
}