1. 头文件
#ifndef __TCP_SERVER_H__ #define __TCP_SERVER_H__ #include <unistd.h> #include <stdio.h> #include <stdlib.h> #include <string.h> #include <errno.h> #include <fcntl.h> #include <sys/socket.h> #include <sys/epoll.h> #include <sys/eventfd.h> #include <netinet/tcp.h> #include <netinet/in.h> #include <arpa/inet.h> #include <netdb.h> #include <iostream> #include <memory> #include <thread> #include <string> #include <atomic> using namespace std; class tcp_server { public: class tcp_notify { public: virtual size_t on_recv_data(const unsigned int clientid, const char* buf, const size_t len) = 0; }; public: tcp_server(tcp_notify& notify); virtual ~tcp_server(); public: void start(const string& port); void stop(); private: int create_and_bind (const char *port); int make_socket_non_blocking (int sfd); void thread_func(); private: tcp_notify& m_notify; int sfd; int efd; static const int MAXEVENTS = 64; struct epoll_event *events; atomic<bool> m_thread_state_; shared_ptr<thread> m_thread_func_; }; #endif /* TCP_SERVER_H_ */
2. 定义文件
#include "tcp_server.h" tcp_server::tcp_server(tcp_notify& notify) : m_notify(notify), efd(::epoll_create1(EPOLL_CLOEXEC)), m_thread_state_(true), m_thread_func_(nullptr) { } tcp_server::~tcp_server() { stop(); } void tcp_server::start(const string& port) { int ret; struct epoll_event event; sfd = create_and_bind(port.c_str()); if (sfd == -1) abort(); ret = make_socket_non_blocking(sfd); if (ret == -1) abort(); ret = listen(sfd, SOMAXCONN); if (ret == -1) { perror("listen"); abort(); } event.data.fd = sfd; event.events = EPOLLIN | EPOLLET; ret = epoll_ctl(efd, EPOLL_CTL_ADD, sfd, &event); if (ret == -1) { perror("epoll_ctl"); abort(); } /* Buffer where events are returned */ events = (struct epoll_event*)calloc(MAXEVENTS, sizeof(event)); m_thread_func_ = make_shared < thread > ( bind(&tcp_server::thread_func, this)); m_thread_func_->join(); } void tcp_server::stop() { m_thread_state_ = false; m_thread_func_->join(); free(events); close(sfd); } int tcp_server::create_and_bind(const char *port) { struct addrinfo hints; struct addrinfo *result, *rp; int ret, sfd; memset(&hints, 0, sizeof(struct addrinfo)); hints.ai_family = AF_UNSPEC; /* Return IPv4 and IPv6 choices */ hints.ai_socktype = SOCK_STREAM; /* We want a TCP socket */ hints.ai_flags = AI_PASSIVE; /* All interfaces */ ret = getaddrinfo(NULL, port, &hints, &result); if (ret != 0) { fprintf(stderr, "getaddrinfo: %s ", gai_strerror(ret)); return -1; } for (rp = result; rp != NULL; rp = rp->ai_next) { sfd = socket(rp->ai_family, rp->ai_socktype, rp->ai_protocol); if (sfd == -1) continue; ret = bind(sfd, rp->ai_addr, rp->ai_addrlen); if (ret == 0) { /* We managed to bind successfully! */ break; } close(sfd); } if (rp == NULL) { fprintf(stderr, "Could not bind "); return -1; } freeaddrinfo(result); return sfd; } int tcp_server::make_socket_non_blocking(int sfd) { int flags, ret; flags = fcntl(sfd, F_GETFL, 0); if (flags == -1) { perror("fcntl"); return -1; } flags |= O_NONBLOCK; ret = fcntl(sfd, F_SETFL, flags); if (ret == -1) { perror("fcntl"); return -1; } return 0; } void tcp_server::thread_func() { int ret; struct epoll_event event; /* The event loop */ while (m_thread_state_) { int n, i; n = epoll_wait(efd, events, MAXEVENTS, -1); for (i = 0; i < n; i++) { if ((events[i].events & EPOLLERR) || (events[i].events & EPOLLHUP) || (!(events[i].events & EPOLLIN))) { /* An error has occured on this fd, or the socket is not ready for reading (why were we notified then?) */ fprintf(stderr, "epoll error "); close(events[i].data.fd); continue; } else if (sfd == events[i].data.fd) { /* We have a notification on the listening socket, which means one or more incoming connections. */ while (1) { struct sockaddr in_addr; socklen_t in_len; int infd; char hbuf[NI_MAXHOST], sbuf[NI_MAXSERV]; in_len = sizeof in_addr; infd = accept(sfd, &in_addr, &in_len); if (infd == -1) { if ((errno == EAGAIN) || (errno == EWOULDBLOCK)) { /* We have processed all incoming connections. */ break; } else { perror("accept"); break; } } ret = getnameinfo(&in_addr, in_len, hbuf, sizeof hbuf, sbuf, sizeof sbuf, NI_NUMERICHOST | NI_NUMERICSERV); if (ret == 0) { printf("Accepted connection on descriptor %d " "(host=%s, port=%s) ", infd, hbuf, sbuf); } /* Make the incoming socket non-blocking and add it to the list of fds to monitor. */ ret = make_socket_non_blocking(infd); if (ret == -1) abort(); event.data.fd = infd; event.events = EPOLLIN | EPOLLET; ret = epoll_ctl(efd, EPOLL_CTL_ADD, infd, &event); if (ret == -1) { perror("epoll_ctl"); abort(); } } continue; } else { /* We have data on the fd waiting to be read. Read and display it. We must read whatever data is available completely, as we are running in edge-triggered mode and won't get a notification again for the same data. */ int done = 0; while (1) { ssize_t count; char buf[512]; count = read(events[i].data.fd, buf, sizeof buf); if (count == -1) { /* If errno == EAGAIN, that means we have read all data. So go back to the main loop. */ if (errno != EAGAIN) { perror("read"); done = 1; } break; } else if (count == 0) { /* End of file. The remote has closed the connection. */ done = 1; break; } /* Write the buffer to standard output */ // ret = write(1, buf, count); // if (ret == -1) { // perror("write"); // abort(); // } m_notify.on_recv_data(events[i].data.fd, buf, count); } if (done) { printf("Closed connection on descriptor %d ", events[i].data.fd); /* Closing the descriptor will make epoll remove it from the set of descriptors which are monitored. */ close(events[i].data.fd); } } // rec end }// for }//while(1) }
3.main.cpp
#include <iostream> #include "tcp_server.h" using namespace std; class notify : public tcp_server::tcp_notify { virtual size_t on_recv_data(const unsigned int clientid, const char* buf, const size_t len) { string str(buf, len); cout << "on_recv_data:" << str << endl; return len; } }; int main(int argc,char *argv[]) { notify notify_inst; tcp_server test(notify_inst); test.start("9587"); cout << "Helloworld!" << endl; return 0; }