libsocketserver.c

这是封装的socket库

#include "tcpsocketserver.h"
#include <sys/socket.h>
#include <sys/types.h>
#include <netinet/in.h>
#include <sys/un.h>
#include <unistd.h>
#include <pthread.h>
#include <poll.h>
#include <fcntl.h>
#include <errno.h>

void * server_thread(void *arg)
{
	int ret,delay = 1000;
	client_list_t *tmp,*tmp_next;
	struct pollfd poll_table[100],*poll_entry;
	server_ctx_t *ser_ctx = (server_ctx_t*)arg;
	fprintf(stderr,"server_fd:%d.\n",ser_ctx->server_fd);
	fprintf(stderr,"server thread start......................\n");
	while(1)
	{
		poll_entry = poll_table;
		poll_entry->fd = ser_ctx->server_fd;
		poll_entry->events = POLLIN;
		poll_entry++;

		list_for_each_entry(tmp,&(ser_ctx->client_list.list),list)
		{
			tmp->poll_entry = poll_entry;
			poll_entry->fd = tmp->client_fd;
			poll_entry->events = POLLIN;
			poll_entry++;
		}
		do{
			ret = poll(poll_table, poll_entry-poll_table, delay);
			if((ret < 0)&& (errno == EINTR) && (errno == EAGAIN))
			{
				return ;
			}
		}while(ret < 0);
		list_for_each_entry_safe(tmp, tmp_next, &(ser_ctx->client_list.list), list)
		{
			/*handle client*/
			ret = handle_connection(tmp);
			if(ret < 0)
			{
				close_connection(tmp);
//				list_for_each_entry(tmp,&(ser_ctx->client_list.list), list)
//				{
		//			fprintf(stderr,"line:%d client_fd:%d\n",__LINE__,tmp->client_fd);
//				}
			}
		}
		poll_entry = poll_table;
//		fprintf(stderr,"line:%d poll_entry->fd:%d,poll_entry->events:%0x,poll_entry->revents:%0x\n",__LINE__,poll_entry->fd,poll_entry->events,poll_entry->revents);
		if(poll_entry->revents & POLLIN)
		{
			new_connection(ser_ctx);
		}


	}
}

static int new_connection(server_ctx_t *server_ctx)
{
	struct sockaddr_in from_addr;
	int fd;
	socklen_t len;
	client_list_t *tmp;
	tmp = (client_list_t*)malloc(sizeof(client_list_t));
	if(tmp == NULL)
	{
		fprintf(stderr,"Failed to malloc client_list var.\n");
		return -1;
	}

	len = sizeof(struct sockaddr_in);
	fd = accept(server_ctx->server_fd, (struct sockaddr*)&from_addr, &len);
	if(fd < 0)
	{
		perror("accept failed\n");
		fprintf(stderr,"accept failed. server_fd:%d\n",server_ctx->server_fd);
		return -1;
	}
	fprintf(stderr,"accetp fd %d.\n",fd);
	set_nonblock(fd);
	tmp->client_fd = fd;
	tmp->server_ctx = server_ctx;
	tmp->poll_entry = NULL;
	tmp->timeout = get_cur_time();
//	fprintf(stderr,"tmp->timeout:%d client_fd:%d\n",tmp->timeout,tmp->client_fd);
//	fprintf(stderr,"client_fd:%d\n",tmp->client_fd);
	list_add(&(tmp->list), &(server_ctx->client_list.list));

}

static int handle_connection(client_list_t *tmp)
{
	int len,ret;
	unsigned int timeout;

	timeout = get_cur_time();
//	fprintf(stderr,"client_timeout:%d\n",CLIENT_TIME_OUT);
//	fprintf(stderr,"timeout:%d tmp->timeout:%d tmp->client_fd:%d\n",timeout,tmp->timeout,tmp->client_fd);
//	fprintf(stderr,"timeout-tmp->timeout:%d client_timeout:%d\n",timeout-(tmp->timeout),CLIENT_TIME_OUT);
	if((timeout - (tmp->timeout)) > CLIENT_TIME_OUT)
	{
		fprintf(stderr,"fd timeout.\n");
		if(tmp->server_ctx->interface.disconnect_callback!=NULL)	
		{
			tmp->server_ctx->interface.disconnect_callback(tmp->server_ctx->disconnect_param,tmp->client_fd);
		}
		return -1;
	}
	if((tmp->poll_entry->revents)& (POLLERR | POLLHUP))
	{
		perror("poll");
		return -1;
	}
	/*this socket for server_send*/
	tmp->server_ctx->client_fd = tmp->client_fd;
	if((tmp->poll_entry->revents) & POLLIN)
	{
		tmp->timeout = timeout;
		len = recv(tmp->client_fd,tmp->server_ctx->buf,8192,0);
		if(len < 0)
		{

			return -1;
		}
		else if(len == 0)
		{
			return -1;
		}
		else
		{
			if(tmp->server_ctx->interface.recv_callback!=NULL)
			{
				tmp->server_ctx->interface.recv_callback(tmp->server_ctx->param,tmp->client_fd, tmp->server_ctx->buf,len, 0);
			}
		}
	}
	return 0;	

}

static int close_connection(client_list_t *tmp)
{
	fprintf(stderr,"close connection fd:%d\n",tmp->client_fd);
	close(tmp->client_fd);
	list_del(&(tmp->list));
	free(tmp);
	tmp == NULL;
	return 0;
}
server_ctx_t* server_new()
{
	server_ctx_t * server_ctx = (server_ctx_t*)malloc(sizeof(server_ctx_t));
	if(server_ctx == NULL)
	{
		fprintf(stderr,"server_new malloc faile\n");
		return NULL;
	}
	return server_ctx;
}

static int server_start(interface_t * this, char *ip, int port)
{
	int ret;
	int server_fd;
	int addr_len;
	pthread_t thread_id;
	struct sockaddr_in server_addr,client_addr;
	server_ctx_t *server_ctx = (server_ctx_t*)this->parent;

	
//	memset(&ser_ctx, 0, sizeof(ser_ctx));
	INIT_LIST_HEAD(&(server_ctx->client_list.list));
	
	server_fd = socket(AF_INET, SOCK_STREAM, 0);
	if( server_fd == -1)
	{
		fprintf(stderr,"Failed to create socket.\n");
		return -1;
	}
	
	addr_len = sizeof(struct sockaddr_in);

	server_addr.sin_family = AF_INET;
	server_addr.sin_port = htons(port);
	server_addr.sin_addr.s_addr = INADDR_ANY;

	ret = 1;
	setsockopt(server_fd, SOL_SOCKET, SO_REUSEADDR, (void*)&ret, sizeof(ret));

	ret = bind(server_fd, (struct sockaddr*)&server_addr, addr_len);
	if(ret == -1)
	{
		fprintf(stderr,"Failed to bind address.\n");
		return -1;

	}

	ret = listen(server_fd,64);
	if(ret == -1)
	{
		fprintf(stderr,"Failed to listen socket.\n");
		return -1;
	}

	set_nonblock(server_fd);
	server_ctx->server_fd = server_fd;	
	ret = pthread_create(&thread_id, NULL, server_thread, this);
	if(ret != 0)
	{
		fprintf(stderr,"Failed to create thread.\n");
		return -1;
	}
	
	return 0;
}

static int set_recv_callback(interface_t * this,void (*callback)(void *param, int socket, char *buf, int len, int flags), void *param)
{
	server_ctx_t *server_ctx = (server_ctx_t*)this->parent;
	this->recv_callback = callback;
	server_ctx->param = param;
}

static int set_disconnect_callback(interface_t * this,void (*callback)(void *param, int socket), void *param)
{
	server_ctx_t *server_ctx = (server_ctx_t*)this->parent;
	this->disconnect_callback = callback;
	server_ctx->disconnect_param = param;
}


static int server_send(interface_t *this, char *buf, int len, int flags)
{
	int ret;
	int delay = 1000;
	struct pollfd poll_fd;
	server_ctx_t *server_ctx = (server_ctx_t*)this->parent;
	poll_fd.fd = server_ctx->client_fd;
	poll_fd.events = POLLOUT;
	while(len > 0)
	{
		do{
			ret = poll(&poll_fd, 1, delay);
			if((ret < 0)&& (errno == EINTR) && (errno == EAGAIN))
			{
				return ;
			}
		}while(ret < 0);

		if((poll_fd.revents)& (POLLERR | POLLHUP))
		{
			perror("poll");
			return -1;
		}

		if(poll_fd.revents & POLLOUT)
		{
			ret = send(server_ctx->client_fd,buf,len,flags);
			buf += len;
			len = len - ret;
		}
	}
	return 0;
}
int server_init(server_ctx_t *server_ctx)
{
	if(server_ctx == NULL)
		return -1;

	server_ctx->interface.parent = (void *)server_ctx;
	server_ctx->interface.interface_start = server_start;
	server_ctx->interface.set_recv_callback = set_recv_callback;
	server_ctx->interface.set_disconnect_callback = set_disconnect_callback;
	server_ctx->interface.interface_send = server_send;

	return 0;
}

void inline set_nonblock(int fd)
{
	int ret;
	
	ret = fcntl(fd, F_GETFL);
	fcntl(fd, F_SETFL, ret|O_NONBLOCK);
	
	return ;
}

void inline set_block(int fd)
{
	int ret;

	ret = fcntl(fd, F_GETFL);
	ret &= ~O_NONBLOCK;
	fcntl(fd, F_SETFL, ret);

	return ;
}
unsigned int get_cur_time()
{
	struct timeval tv;
	gettimeofday(&tv,NULL);
	return tv.tv_sec;
}