#include <stdio.h>
#include <stdlib.h>
#include <errno.h>
#include <string.h>
#include <sys/types.h>
#include <netinet/in.h>
#include <sys/socket.h>
#include <sys/wait.h>
#include <unistd.h>
#include <arpa/inet.h>
#include <fcntl.h>
#include <sys/epoll.h>
#include <sys/time.h>
#include <sys/resource.h>
#include <pthread.h>
#include <assert.h>
#include <list>
#define HEADSIZE 5
#define MAXEPOLLSIZE 20
#define PORT 8888
#define _PKG_INIT 0
#define _PKG_RECVING 1
class subscriber
{
private:
std::list<char *> m_MsgRecvQueue;
int listener, epoll_udp_fd;
socklen_t len;
struct sockaddr_in subscriber_addr;
struct epoll_event event;
struct epoll_event new_event;
struct epoll_event events[MAXEPOLLSIZE]; //用于存放epoll_wait返回的可读事件
struct sub_connection_s
{
sub_connection_s():fd(-1),irecvlen(0),curStat(_PKG_INIT),precvbuf(NULL),precvMemPointer(NULL){}
~sub_connection_s()
{
//delete []precvbuf;
}
int fd;
int irecvlen;
unsigned char curStat;
char dataHeadInfo[HEADSIZE];
char *precvbuf;
char *precvMemPointer;
};
public:
subscriber()
{
int ret = 0;
int opt = 1;
/*创建一个UDP套接字*/
if((listener = socket(PF_INET, SOCK_DGRAM, 0)) == -1)
{
perror("socket");
exit(1);
}
if((ret = setsockopt(listener, SOL_SOCKET, SO_REUSEADDR, &opt,sizeof(opt))) != 0)
{
exit(1);
}
if((ret = setsockopt(listener, SOL_SOCKET, SO_REUSEPORT, &opt, sizeof(opt))) !=0)
{
exit(1);
}
bzero(&subscriber_addr, sizeof(subscriber_addr));
/*地址填充*/
subscriber_addr.sin_family = PF_INET;
subscriber_addr.sin_port = htons(PORT);
subscriber_addr.sin_addr.s_addr = INADDR_ANY;
/*将创建的UDP套接字和结构体地址绑定*/
if(bind(listener, (struct sockaddr *)&subscriber_addr, sizeof(struct sockaddr)) == -1)
{
perror("bind");
exit(1);
}
}
bool epoll_udp_init()
{
/*创建一个epoll对象*/
epoll_udp_fd = epoll_create(MAXEPOLLSIZE);
/*关注可读事件，默认模式为LT模式*/
event.events = EPOLLIN;
/*套接字赋值*/
event.data.fd = listener;
/*将该事件作为一个节点加入到epoll对象中来*/
if(epoll_ctl(epoll_udp_fd, EPOLL_CTL_ADD, listener, &event) < 0)
{
fprintf(stderr, "epoll set insertion error: fd=%dn", listener);
return false;
}
return true;
}
int epoll_udp_accept(int recvsocket, struct sockaddr_in new_addr)
{
int new_sd = -1, ret = 0,reuse = 1;
struct sockaddr_in peer_addr;
socklen_t cli_len = sizeof(peer_addr);
/*为新连接分配一个连接*/
sub_connection_s *pConn = new sub_connection_s();
memset(pConn, 0, sizeof(sub_connection_s));
/*收包头*/
if((ret = recvfrom(recvsocket, pConn->dataHeadInfo, HEADSIZE, 0, (struct sockaddr *)&peer_addr, &cli_len)) > 0)
{
printf("pConn->dataHeadInfo = %s\n", pConn->dataHeadInfo);
printf("ret = %d\n", ret);
if(ret == HEADSIZE)
{
pConn->irecvlen = atoi(pConn->dataHeadInfo);
pConn->precvbuf = new char[pConn->irecvlen];
pConn->precvMemPointer = pConn->precvbuf;
printf("pConn->irecvlen = %d\n", pConn->irecvlen);
}
}
else
{
printf("error \n");
}
/*创建一个UDP套接字*/
if((new_sd = socket(PF_INET, SOCK_DGRAM, 0)) == -1)
{
perror("child socket");
exit(1);
}
else
{
pConn->fd = new_sd;
}
/*设置地址复用功能*/
if((ret = setsockopt(new_sd, SOL_SOCKET, SO_REUSEADDR, &reuse, sizeof(reuse))) != 0)
{
exit(1);
}
/*设置端口复用功能*/
if((ret = setsockopt(new_sd, SOL_SOCKET, SO_REUSEPORT, &reuse, sizeof(reuse))) != 0)
{
exit(1);
}
/*以一个新的套接字重新绑定这个客户端，后面接收数据就从这个新的套接字接收了*/
if((ret = bind(new_sd, (struct sockaddr *)&new_addr, sizeof(struct sockaddr))) != 0)
{
perror("new bind");
exit(1);
}
/*设置地址协议族*/
peer_addr.sin_family = PF_INET;
/*发起连接*/
if(connect(new_sd, (struct sockaddr *) &peer_addr, sizeof(struct sockaddr)) == -1)
{
perror("new connect");
exit(1);
}
/*关注可读事件*/
new_event.events = EPOLLIN;
/*为新接入的客户端增加文件描述符*/
new_event.data.fd = new_sd;
/*为每一个连接建立一个连接*/
new_event.data.ptr = (void *)pConn;
/*把新连接作为一个节点加到红黑树当中*/
if(epoll_ctl(epoll_udp_fd, EPOLL_CTL_ADD, new_sd, &new_event) < 0)
{
fprintf(stderr, "epoll set insertion error: fd=%dn", new_sd);
return -1;
}
return 1;
}
void epoll_udp_loop()
{
for(;;)
{
/*等待新的事件来*/
int nfds = epoll_wait(epoll_udp_fd, events, MAXEPOLLSIZE, -1);
if(nfds == -1)
{
perror("epoll_wait");
break;
}
for(int n = 0; n < nfds; ++n)
{
if(events[n].data.fd == listener)
{
printf("listener:%d\n", n);
epoll_udp_accept(listener, subscriber_addr);
}
else
{
recv_data(events[n].data.ptr);
}
}
}
}
bool recv_data(void *pTempConn)
{
int reco = 0;
sub_connection_s *pConn = (sub_connection_s *)pTempConn;
if((reco = recvfrom(pConn->fd, pConn->precvbuf, pConn->irecvlen, 0, NULL, NULL)) > 0)
{
printf("read[%d]: %s from %d\n", reco, pConn->precvbuf, pConn->fd);
/*收包初始化状态*/
if(pConn->curStat == _PKG_INIT)
{
/*收包完毕*/
if(pConn->irecvlen == reco)
{
/*将收到的数据包存储到队列中*/
printf("6666\n");
printf("pConn->precvMemPointer = %s\n", pConn->precvMemPointer);
m_MsgRecvQueue.push_back(pConn->precvMemPointer);
/*恢复初始状态*/
pConn->precvMemPointer = NULL;
pConn->curStat = _PKG_INIT;
pConn->precvbuf = NULL;
pConn->irecvlen = 0;
}
else
{
/*此时接收包头处于接收状态*/
pConn->curStat = _PKG_RECVING;
/*收包缓冲区往后走*/
pConn->precvbuf = pConn->precvbuf + reco;
/*剩余的要接收包头的长度*/
pConn->irecvlen = pConn->irecvlen - reco;
}
}
/*收包当中*/
else if(pConn->curStat == _PKG_RECVING)
{
/*收包完毕*/
if(pConn->irecvlen == reco)
{
/*将收到的数据包存储到队列中*/
m_MsgRecvQueue.push_back(pConn->precvMemPointer);
printf("8888\n");
printf("pConn->precvMemPointer = %s\n", pConn->precvMemPointer);
/*恢复初始状态*/
pConn->precvMemPointer = NULL;
pConn->curStat = _PKG_INIT;
pConn->precvbuf = NULL;
pConn->irecvlen = 0;
}
else
{
/*收包缓冲区往后走*/
pConn->precvbuf = pConn->precvbuf + reco;
/*剩余的要接收包头的长度*/
pConn->irecvlen = pConn->irecvlen - reco;
}
}
}
return true;
}
~subscriber()
{
close(listener);
}
};
int main(int argc, char **argv)
{
subscriber obj;
obj.epoll_udp_init();
obj.epoll_udp_loop();
return 0;
}
