TcpServer::TcpServer(EventLoop* loop,
const InetAddress& listenAddr,
const string& _name)
: loop_(CHECK_NOTNULL(loop)),
hostport_(listenAddr.toHostPort()),
name_(_name),
acceptor_(new Acceptor(loop, listenAddr)),
threadPool_(new EventLoopThreadPool(loop)),
connectionCallback_(defaultConnectionCallback),
messageCallback_(defaultMessageCallback),
started_(false),
nextConnId_(1)
{
acceptor_->setNewConnectionCallback(
boost::bind(&TcpServer::newConnection, this, _1, _2));
}

Acceptor::Acceptor(EventLoop* loop, const InetAddress& listenAddr, bool reuseport)
: loop_(loop),
acceptSocket_(sockets::createNonblockingOrDie()),      //create a listening socket
acceptChannel_(loop, acceptSocket_.fd()),                    //create listening channel
listenning_(false),
idleFd_(::open("/dev/null", O_RDONLY | O_CLOEXEC))

{
    acceptSocket_.setReuseAddr(true);
    acceptSocket_.bindAddress(listenAddr);
    acceptChannel_.setReadCallback(
    boost::bind(&Acceptor::handleRead, this));
 }

void Acceptor::handleRead()
{loop_->assertInLoopThread();InetAddress peerAddr(0);//FIXME loop until no moreint connfd = acceptSocket_.accept(&peerAddr);if (connfd >= 0){// string hostport = peerAddr.toHostPort();// LOG_TRACE << "Accepts of " << hostport;if (newConnectionCallback_){newConnectionCallback_(connfd, peerAddr);}else{sockets::close(connfd);}}
}

int sockets::accept(int sockfd, struct sockaddr_in* addr)
{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;LOG_SYSERR << "Socket::accept";switch (savedErrno){case EAGAIN:case ECONNABORTED:case EINTR:case EPROTO: // ???case EPERM:// expected errorsbreak;case EBADF:case EFAULT:case EINVAL:case EMFILE: // per-process lmit of open file desctiptor ???case ENFILE:case ENOBUFS:case ENOMEM:case ENOTSOCK:case EOPNOTSUPP:// unexpected errorsLOG_FATAL << "unexpected error of ::accept";break;default:LOG_FATAL << "unknown error of ::accept " << savedErrno;break;}}return connfd;
}

void TcpServer::start()
{if (!started_){started_ = true;threadPool_->start();}if (!acceptor_->listenning()){loop_->runInLoop(boost::bind(&Acceptor::listen, get_pointer(acceptor_)));}
}

void Acceptor::listen()
{loop_->assertInLoopThread();listenning_ = true;acceptSocket_.listen();acceptChannel_.enableReading();
}

void enableReading() { events_ |= kReadEvent; update(); }

void Channel::update()
{loop_->updateChannel(this);
}

void EventLoop::updateChannel(Channel* channel)
{assert(channel->getLoop() == this);assertInLoopThread();poller_->updateChannel(channel);
}

void EPollPoller::updateChannel(Channel* channel)
{Poller::assertInLoopThread();LOG_TRACE << "fd = " << channel->fd() << " events = " << channel->events();if (channel->index() < 0){// a new one, add with EPOLL_CTL_ADDint fd = channel->fd();assert(channels_.find(fd) == channels_.end());channel->set_index(1);channels_[fd] = channel;update(EPOLL_CTL_ADD, channel);}else{// update existing one with EPOLL_CTL_MODint fd = channel->fd();assert(channels_.find(fd) != channels_.end());assert(channels_[fd] == channel);assert(channel->index() == 1);update(EPOLL_CTL_MOD, channel);}
}

void EPollPoller::update(int operation, Channel* channel)
{struct epoll_event event;bzero(&event, sizeof event);event.events = channel->events();event.data.ptr = channel;if (::epoll_ctl(epollfd_, operation, channel->fd(), &event) < 0){LOG_SYSFATAL << "epoll_ctl op=" << operation;}
}