处理新连接Acceptor

处理新连接Acceptor

前面已经完成了muduo这些模块的重写：

1. 日志、InetAddress、Socket的封装
2. 事件channel的封装
3. 多路分发器Poller的抽象接口及EPollPoller的具体实现
4. EventLoop事件循环分发
5. EventLoopThreadPool开启Thread执行EventLoop事件循环

Acceptor就是封装了对listenfd的操作，负责监听新用户的连接。连接到来后执行处理新连接到来的回调，这个回调所做的就是拿到跟客户端通信的connfd，并打包成channel，根据channel的分配策略选择一个subLoop唤醒处理。

Acceptor所属loop_一定是mainLoop，acceptSocket_成员是listenfd监听地址的封装对象，acceptChannel_成员是listenfd和其关注事件的打包，newConnectionCallback是新连接到来时的处理回调函数对象，listenning_成员是listenfd的状态。

重写Acceptor.h:

#pragma once

#include "noncopyable.h"
#include "Socket.h"
#include "Channel.h"

#include 

class EventLoop;
class InetAddress;

class Acceptor : noncopyable
{
public:
    using NewConnectionCallback = std::function<void(int sockfd, const InetAddress&)>;
    Acceptor(EventLoop* loop, const InetAddress& listenAddr, bool reuseport);
    ~Acceptor();

    void setNewConnectionCallback(const NewConnectionCallback& cb)
    {
        newConnectionCallback_ = cb;
    }

    bool listenning() const { return listenning_; }
    void listen();
private:
    void handleRead();

    EventLoop* loop_; //Acceptor用的就是用户定义的那个baseLoop，也称作mainLoop
    Socket acceptSocket_;
    Channel acceptChannel_;
    NewConnectionCallback newConnectionCallback_;
    bool listenning_;
};
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

重写Acceptor.cc:

#include "Acceptor.h"
#include "Logger.h"
#include "InetAddress.h"

#include 
#include 
#include 
#include 

static int createNonblocking()
{
    int sockfd = socket(AF_INET, SOCK_STREAM | SOCK_NONBLOCK | SOCK_CLOEXEC, 0);
    if(sockfd < 0)
    {
        LOG_FATAL("%s:%s:%d listen socket create err:%d \n", __FILE__, __FUNCTION__, __LINE__, errno);
    }
    return sockfd;
}

Acceptor::Acceptor(EventLoop* loop, const InetAddress& listenAddr, bool reuseport)
    :loop_(loop)
    ,acceptSocket_(createNonblocking())
    ,acceptChannel_(loop, acceptSocket_.fd())
    ,listenning_(false)
{
    acceptSocket_.setReuseAddr(true);
    acceptSocket_.setReusePort(true);
    acceptSocket_.bindAddress(listenAddr); //bind
    acceptChannel_.setReadCallback(std::bind(&Acceptor::handleRead, this));

}
Acceptor::~Acceptor()
{
    acceptChannel_.disableALL();
    acceptChannel_.remove();
}

void Acceptor::listen()
{
    listenning_ = true;
    acceptSocket_.listen();     //listen
    acceptChannel_.enableReading(); //acceptChannel_ => Poller
}

//listenfd有事件发生了，就是用新用户连接了
void Acceptor::handleRead()
{
    InetAddress peerAddr = InetAddress();
    int connfd = acceptSocket_.accept(&peerAddr);
    if(connfd >= 0)
    {
        if(newConnectionCallback_)
        {
            newConnectionCallback_(connfd, peerAddr);   //轮询找到subLoop，唤醒，分发当前的新客户端的Channel
        }
        else
        {
            ::close(connfd);
        }
    }
    else
    {
        LOG_ERROR("%s:%s:%d listen socket create err:%d \n", __FILE__, __FUNCTION__, __LINE__, errno);
        if(errno == EMFILE)
        {
            LOG_ERROR("%s:%s:%d socket reached limit! \n", __FILE__, __FUNCTION__, __LINE__);
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69