Acceptor+TcpConnection+TcpServer

Acceptor负责accpet一个TCP客户连接并执行相应的回调通知连接的使用者

TcpConnection是指一个TCP连接，执行相应的连接回调

TcpServer管理所有的TCP连接

#include<iostream>
#include<stdio.h>
#include<map>
#include<string>
#include<boost/any.hpp>
#include<boost/enable_shared_from_this.hpp>
#include<boost/noncopyable.hpp>
#include<boost/scoped_ptr.hpp>
#include<boost/shared_ptr.hpp>
#include<netinet/in.h>
#include<errno.h>
#include<fcntl.h>
#include<stdio.h>
#include<strings.h>
#include<sys/types.h>
#include<sys/socket.h>
#include<unistd.h>
#include<arpa/inet.h>
#include<endian.h>
#include<netinet/tcp.h>
#include<netinet/in.h>
#include<boost/noncopyable.hpp>
#include<boost/function.hpp>
#include<boost/static_assert.hpp>
#include<boost/bind.hpp>
#include<string>
#include<stdio.h>
#include<iostream>
#include"EventLoop.hpp"
namespace sockets{inline uint64_t hostToNetwork64(uint64_t host64)
{//主机字节序转为网络字节序return htobe64(host64);
}
inline uint32_t hostToNetwork32(uint32_t host32)
{return htonl(host32);
}
inline uint16_t hostToNetwork16(uint16_t host16)
{return htons(host16);
}
inline uint64_t networkToHost64(uint64_t net64)
{//网络字节序转为主机字节序return be64toh(net64);
}inline uint32_t networkToHost32(uint32_t net32)
{return ntohl(net32);
}
inline uint16_t networkToHost16(uint16_t net16)
{return ntohs(net16);
}typedef struct sockaddr SA;
const SA* sockaddr_cast(const struct sockaddr_in* addr){//强制转换return static_cast<const SA*>(implicit_cast<const void*>(addr));
}
SA* sockaddr_cast(struct sockaddr_in* addr){return static_cast<SA*>(implicit_cast<void*>(addr));
}
void setNonBlockAndCloseOnExec(int sockfd){//将描述符设置为非阻塞和O_CLOEXEC(close on exec)int flags = ::fcntl(sockfd, F_GETFL, 0);flags |= O_NONBLOCK;int ret = ::fcntl(sockfd, F_SETFL, flags);flags = ::fcntl(sockfd, F_GETFD, 0);flags |= FD_CLOEXEC;ret = ::fcntl(sockfd, F_SETFD, flags);
}
int createNonblockingOrDie()
{//socket()创建非阻塞的socket描述符#if VALGRINDint sockfd = ::socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);if (sockfd < 0) {printf("socket() error\n");}setNonBlockAndCloseOnExec(sockfd);#elseint sockfd = ::socket(AF_INET,SOCK_STREAM | SOCK_NONBLOCK | SOCK_CLOEXEC,IPPROTO_TCP);if (sockfd < 0){printf("socke() error\n");}#endifreturn sockfd;
}
void bindOrDie(int sockfd, const struct sockaddr_in& addr)
{//bind()int ret = ::bind(sockfd, sockaddr_cast(&addr), sizeof addr);if (ret < 0) {printf("bind() error\n");}
}
void listenOrDie(int sockfd){//listen()int ret = ::listen(sockfd, SOMAXCONN);if (ret < 0){printf("listen() error\n");}
}
int accept(int sockfd, struct sockaddr_in* addr)
{//accept()socklen_t addrlen = sizeof *addr;#if VALGRINDint connfd = ::accept(sockfd, sockaddr_cast(addr), &addrlen);setNonBlockAndCloseOnExec(connfd);#elseint connfd = ::accept4(sockfd, sockaddr_cast(addr),&addrlen, SOCK_NONBLOCK | SOCK_CLOEXEC);#endifif (connfd < 0){int savedErrno = errno;printf("accept error\n");switch (savedErrno){case EAGAIN:case ECONNABORTED:case EINTR:case EPROTO: // ???case EPERM:case EMFILE: // per-process lmit of open file desctiptor ???errno = savedErrno;break;case EBADF:case EFAULT:case EINVAL:case ENFILE:case ENOBUFS:case ENOMEM:case ENOTSOCK:case EOPNOTSUPP:printf("accept() fatal erro\n");break;default:printf("accpet() unknown error\n");break;}}return connfd;
}
void close(int sockfd){//close()if (::close(sockfd) < 0){printf("sockets::close\n");}
}
void toHostPort(char* buf, size_t size,const struct sockaddr_in& addr)
{//将IPv4地址转为IP和端口char host[INET_ADDRSTRLEN] = "INVALID";::inet_ntop(AF_INET, &addr.sin_addr, host, sizeof host);uint16_t port =networkToHost16(addr.sin_port);snprintf(buf, size, "%s:%u", host, port);
}
void fromHostPort(const char* ip, uint16_t port,struct sockaddr_in* addr)
{//将主机IP和端口转为IPv4地址addr->sin_family = AF_INET;addr->sin_port = hostToNetwork16(port);if (::inet_pton(AF_INET, ip, &addr->sin_addr) <= 0){printf("sockets::fromHostPort\n");}
}
sockaddr_in getLocalAddr(int sockfd)
{struct sockaddr_in localaddr;bzero(&localaddr, sizeof localaddr);socklen_t addrlen = sizeof(localaddr);if (::getsockname(sockfd, sockaddr_cast(&localaddr), &addrlen) < 0){printf("getsockname() error\n");}return localaddr;
}
}//end-namespaceclass InetAddress;
class Socket:noncopyable{//创建一个socket描述符fd并绑定sockaddr，监听fd功能public:explicit Socket(int sockfd):sockfd_(sockfd){}~Socket();int fd() const{return sockfd_;}void bindAddress(const InetAddress& addr);void listen();int accept(InetAddress* peeraddr);void setReuseAddr(bool on);private:const int sockfd_;
};
class InetAddress{//sockaddr地址的风黄钻public:explicit InetAddress(uint16_t port);InetAddress(const string& ip,uint16_t port);InetAddress(const struct sockaddr_in& addr):addr_(addr){}string toHostPort() const;const struct sockaddr_in& getSockAddrInet() const{return addr_;}void setSockAddrInet(const struct sockaddr_in& addr){addr_=addr;}private:struct sockaddr_in addr_;
};
BOOST_STATIC_ASSERT(sizeof(InetAddress)==sizeof(struct sockaddr_in));//编译时断言
class Acceptor:noncopyable{//接受TCP连接并执行相应的回调public:typedef function<void(int sockfd,const InetAddress&)> NewConnectionCallback;Acceptor(EventLoop* loop,const InetAddress& listenAddr);void setNewConnectionCallback(const NewConnectionCallback& cb){ newConnectionCallback_=cb;}bool listening() const{return listening_;}void listen();private:void handleRead();EventLoop* loop_;Socket acceptSocket_;Channel acceptChannel_;NewConnectionCallback newConnectionCallback_;bool listening_;};
/**Socket实现*/
Socket::~Socket()
{sockets::close(sockfd_);
}
void Socket::bindAddress(const InetAddress& addr)
{sockets::bindOrDie(sockfd_, addr.getSockAddrInet());
}
void Socket::listen()
{sockets::listenOrDie(sockfd_);
}
int Socket::accept(InetAddress* peeraddr)
{struct sockaddr_in addr;bzero(&addr, sizeof addr);int connfd = sockets::accept(sockfd_, &addr);if (connfd >= 0){peeraddr->setSockAddrInet(addr);}return connfd;
}
void Socket::setReuseAddr(bool on)
{int optval = on ? 1 : 0;::setsockopt(sockfd_, SOL_SOCKET, SO_REUSEADDR,&optval, sizeof optval);
}
/**InetAddress实现*/
static const in_addr_t kInaddrAny=INADDR_ANY;//任意的网络字节序IP地址为0
InetAddress::InetAddress(uint16_t port)
{bzero(&addr_, sizeof addr_);addr_.sin_family = AF_INET;addr_.sin_addr.s_addr = sockets::hostToNetwork32(kInaddrAny);addr_.sin_port = sockets::hostToNetwork16(port);
}
InetAddress::InetAddress(const std::string& ip, uint16_t port)
{bzero(&addr_, sizeof addr_);sockets::fromHostPort(ip.c_str(), port, &addr_);
}
string InetAddress::toHostPort() const
{char buf[32];sockets::toHostPort(buf, sizeof buf, addr_);return buf;
}
/**Acceptor实现*/
Acceptor::Acceptor(EventLoop* loop, const InetAddress& listenAddr): loop_(loop),acceptSocket_(sockets::createNonblockingOrDie()),acceptChannel_(loop, acceptSocket_.fd()),listening_(false)
{acceptSocket_.setReuseAddr(true);acceptSocket_.bindAddress(listenAddr);acceptChannel_.setReadCallBack(boost::bind(&Acceptor::handleRead, this));
}
void Acceptor::listen()
{loop_->assertInLoopThread();listening_ = true;acceptSocket_.listen();acceptChannel_.enableReading();
}
void Acceptor::handleRead()
{loop_->assertInLoopThread();InetAddress peerAddr(0);int connfd = acceptSocket_.accept(&peerAddr);if (connfd >= 0) {if (newConnectionCallback_) {newConnectionCallback_(connfd, peerAddr);} else {sockets::close(connfd);}}
}
/**测试代码*/
/*
void newConnection(int sockfd, const InetAddress& peerAddr)
{printf("newConnection(): accepted a new connection from %s\n",peerAddr.toHostPort().c_str());::write(sockfd, "How are you?\n", 13);sockets::close(sockfd);
}
int main()
{printf("main(): pid = %d\n", getpid());InetAddress listenAddr(9981);EventLoop loop;Acceptor acceptor(&loop, listenAddr);acceptor.setNewConnectionCallback(newConnection);acceptor.listen();InetAddress listenAddr1(12345);Acceptor acceptor1(&loop,listenAddr1);acceptor1.setNewConnectionCallback(newConnection);acceptor1.listen();loop.loop();
}
*/
using namespace std;
using namespace boost;
class TcpConnection;//表示一个TCP连接
typedef shared_ptr<TcpConnection> TcpConnectionPtr;//
/**TcpConnection*/
class TcpConnection:noncopyable,public enable_shared_from_this<TcpConnection>{public:typedef function<void(const TcpConnectionPtr&)> ConnectionCallback;//连接建立回调函数typedef function<void(const TcpConnectionPtr&,const char* data,ssize_t len)> MessagCallback;//消息回调函数TcpConnection(EventLoop* loop,const string& name,int sockfd,const InetAddress& localAddr,const InetAddress& peerAddr);~TcpConnection();EventLoop* getLoop() const{return loop_;}const string& name() const{return name_;}const InetAddress& localAddr(){return localAddr_;}const InetAddress& peerAddress(){return peerAddr_;}bool connected() const{return state_==kConnected;}void setConnectionCallback(const ConnectionCallback& cb){connectionCallback_=cb;}void setMessageCallback(const MessagCallback& cb){messageCallback_=cb;}void connectEstablished();private:enum StateE{kConnecting,kConnected,};void setState(StateE s){state_=s;}void handleRead();EventLoop* loop_;string name_;StateE state_;scoped_ptr<Socket> socket_;scoped_ptr<Channel> channel_;InetAddress localAddr_;InetAddress peerAddr_;ConnectionCallback connectionCallback_;MessagCallback messageCallback_;
};
/**Tcpserver*/
class TcpServer:noncopyable{//管理所有的TCP连接public:typedef function<void()> ConnectionCallback;typedef function<void()> MessagCallback;TcpServer(EventLoop* loop,const InetAddress& listenAddr);~TcpServer();void start();void setConnectionCallback(const ConnectionCallback& cb){connectionCallback_=cb;}void setMessageCallback(const MessagCallback& cb){messageCallback_=cb;}private:void newConnection(int sockfd,const InetAddress& peerAddr);typedef map<string,TcpConnectionPtr> ConnectionMap;EventLoop* loop_;const string name_;scoped_ptr<Acceptor> acceptor_;ConnectionCallback connectionCallback_;MessagCallback messageCallback_;bool started_;int nextConnId_;ConnectionMap connections_;
};
/* *TcpConnection实现*/
TcpConnection::TcpConnection(EventLoop* loop,const std::string& nameArg,int sockfd,const InetAddress& localAddr,const InetAddress& peerAddr): loop_(loop),name_(nameArg),state_(kConnecting),socket_(new Socket(sockfd)),channel_(new Channel(loop, sockfd)),localAddr_(localAddr),peerAddr_(peerAddr)
{channel_->setReadCallBack(boost::bind(&TcpConnection::handleRead, this));
}
TcpConnection::~TcpConnection()
{printf("TcpConnection::%s,fd=%d\n",name_.c_str(),channel_->fd());
}
void TcpConnection::connectEstablished()
{loop_->assertInLoopThread();assert(state_ == kConnecting);setState(kConnected);channel_->enableReading();connectionCallback_(shared_from_this());//连接建立回调函数
}
void TcpConnection::handleRead()
{char buf[65536];ssize_t n = ::read(channel_->fd(), buf, sizeof buf);messageCallback_(shared_from_this(), buf, n);
}
/**TcpServer实现*/
TcpServer::TcpServer(EventLoop* loop, const InetAddress& listenAddr): loop_(loop),name_(listenAddr.toHostPort()),acceptor_(new Acceptor(loop, listenAddr)),started_(false),nextConnId_(1)
{acceptor_->setNewConnectionCallback(boost::bind(&TcpServer::newConnection, this, _1, _2));
}
TcpServer::~TcpServer()
{
}
void TcpServer::start()
{if (!started_){started_ = true;}if (!acceptor_->listening()){loop_->runInLoop(boost::bind(&Acceptor::listen, get_pointer(acceptor_)));}//通过EventLoop监听服务端的listenfd
}
void TcpServer::newConnection(int sockfd, const InetAddress& peerAddr)
{//用于Acceptor接受一个连接后通过此回调通知使用者loop_->assertInLoopThread();char buf[32];snprintf(buf, sizeof buf, "#%d", nextConnId_);++nextConnId_;string connName = name_ + buf;InetAddress localAddr(sockets::getLocalAddr(sockfd));TcpConnectionPtr conn(new TcpConnection(loop_, connName, sockfd, localAddr, peerAddr));connections_[connName]=conn;conn->setConnectionCallback(connectionCallback_);//传递给TcpConnectionconn->setMessageCallback(messageCallback_);conn->connectEstablished();//调用ConnectionCallback
}
/**测试代码*/
void onConnection(const TcpConnectionPtr& conn)
{if (conn->connected()){printf("onConnection(): new connection [%s] from %s\n",conn->name().c_str(),conn->peerAddress().toHostPort().c_str());}else{printf("onConnection(): connection [%s] is down\n",conn->name().c_str());}
}void onMessage(const :TcpConnectionPtr& conn,const char* data,ssize_t len)
{printf("onMessage(): received %zd bytes from connection [%s]\n",len, conn->name().c_str());
}int main()
{printf("main(): pid = %d\n", getpid());InetAddress listenAddr(9981);EventLoop loop;TcpServer server(&loop, listenAddr);server.setConnectionCallback(onConnection);server.setMessageCallback(onMessage);server.start();loop.loop();
}

Acceptor+TcpConnection+TcpServer相关推荐

muduo网络库之Acceptor、TcpServer
本篇博客针对Acceptor类和TcpServer类做下小结. 博客代码来自于陈硕的muduo网络库,github地址https://github.com/chenshuo/muduo 学习笔记: A ...
muduo::Acceptor、TcpServer分析
Acceptor 类Acceptor用来listen.accept,并调用回调函数来处理新到的连接.Acceptor中封装了Socket fd,它是用RAII手法初始化.accept后,如果有新连接到 ...
muduo源码分析——TcpServer和Acceptor
这篇文章用于分析muduo的TcpServer类和Acceptor类,原本打算将TcpConnection也放到这里一起聊的,但是那个太多啦,一篇文章太长会让人读的很不舒服把. 当然我用的代码是其他大 ...
muduo网络库学习（七）用于创建服务器的类TcpServer
目前为止,涉及到的绝大多数操作都没有提及线程,EventLoop,Poller,Channel,Acceptor,TcpConnection,这些对象的执行都是在单独线程完成,并没有设计多线程的创建销 ...
muduo源码剖析 - Acceptor
说明 Acceptor 是 TcpServer的一部分, 用于接受Tcp连接,主要完成服务端的 create,bind, listen,accept这几个阶段. 成员变量重点关注: 1. loop ...
acceptor 大法好
boost asio中io_service类的几种使用 io_service类你应该已经发现大部分使用Boost.Asio编写的代码都会使用几个io_service的实例.io_service是这个 ...
muduo学习笔记-Acceptor类
Acceptor类一般由TCPServer创建,负责处理客户端发送的connect,它拥有一个acceptSocket_和acceptChannel_成员. 1.创建Acceptor : TcpSer ...
muduo网络库学习（八）事件驱动循环线程池EventLoopThreadPool
muduo是支持多线程的网络库,在muduo网络库学习(七)用于创建服务器的类TcpServer中也提及了TcpServer中有一个事件驱动循环线程池,线程池中存在大量线程,每个线程运行一个Event ...
长文梳理Muduo库核心代码及优秀编程细节剖析
一.前言: 代码地址: https://github.com/yyg192/Cpp11-Muduo-MultiReactor Muduo库是陈硕个人开发的Tcp网络编程库,支持Reactor模型.本 ...

Acceptor+TcpConnection+TcpServer

Acceptor+TcpConnection+TcpServer相关推荐

最新文章

热门文章