1.9.C++项目：仿muduo库实现并发服务器之Connection模块的设计

项目完整在：

一、Connection模块：这是一个对于通信连接进行整体管理的一个模块，对一个连接的操作都是通过这个模块来进行！

二、提供的功能

Connection模块，一个连接有任何的事件怎么处理都是有这个模块来进行处理的，因为组件的设计也不知道使用者要如何处理事件，因此只能是提供一些事件回调函数由使用者设置。

三、实现思想

（一）功能

1. 发送数据 —— 给用户提供的发生数据的接口，并不是真的发送接口，而只是把数据发送到发送缓冲区，然后启动写事件监控！
2. 关闭连接 —— 给用户提供的关闭链接接口，在实际释放连接之前，看看输入输出缓冲区是否有数据待处理！
3. 启动非活跃连接超时销毁功能
4. 取消非活跃连接超时销毁功能
5. 协议切换 —— 一个链接接受数据后如何进行业务处理，取决于上下文，以及数据的业务处理回调函数！

（二）意义

这个模块本身不是一个单独的功能模块，是一个对连接做管理的模块。

（三）功能设计

1. 套接字的管理，能够进行套接字的操作！
2. 连接事件的管理，可读，可写，错误，挂断，任意！
3. 缓冲区管理：把socket读取的数据放进缓冲区，要有输入缓冲区和输出缓冲区管理！
4. 协议上下文的管理，记录请求数据的处理过程！
5. 回调函数的管理
   因为连接收到数据之后该如何处理，需要由用户决定，必须要有业务处理函数！
   一个连接建立成功后，应该如何处理，由用户决定，因此必须有连接建立成功的回调函数！
   一个连接关闭前，该如何处理，有用户决定，因此必须有关闭连接回调函数！
   任何事件的产生，有没有某些处理，由用户决定，因此必须任意事件的回调函数！

场景：对连接进行操作的时候，对于连接以及被释放，导致内存访问错误，最终程序崩溃！
解决方案：使用智能指针share_ptr 对Connect 对象进行管理，这样可以保证任意一个地方对Connect对象进行操作的时候，
保存了一分share_ptr,因此就算其他地方进行了释放，也只是对share_ptr的计数器-1，而不会导致Connection的实际释放！

四、框架

DISCONECTED -- 连接关闭状态；   CONNECTING -- 连接建立成功-待处理状态
//CONNECTED -- 连接建立完成，各种设置已完成，可以通信的状态；  DISCONNECTING -- 待关闭状态
type enum { 
        // 连接关闭；
        // 连接建立成功 —— 待处理状态；
        // 连接设立完成，可以通信；
        // 待关闭状态；
        DISCONECTED,CONNECTING,CONNECTED,DISCONECTING} ConnStatu;
using PreConnection = std::shared_ptr<Connection>;
class Connection {
        private:
                uint64_t _conn_id; //连接的唯一ID，便于连接的管理和查找
                bool _enable_inactive_release;  // 连接是否启动非活跃销毁的判断标志，默认为false
                int _sockfd; // 连接关联的文件描述符
                ConnStatu _statu; // 
                Socket _socket; // 套接字操作管理
                Channel _channel;  // 连接二点事件管理
                Buffer _in_buffer; // 输入缓冲区 —— 存放从socket中读到的数据
                buffer _out_buffer; // 输出缓冲区 ——  发送给对端的是数据，等到描述符事件可写，再发！
                Any _context; // 请求的接受处理上下文
                /*这四个回调函数，是让服务器模块来设置的（其实服务器模块的处理回调也是组件使用者设置的）*/
                /*换句话说，这几个回调都是组件使用者使用的*/
                using ConnectCallback = std::function<void(const PreConnection&)>;
                using MessageCallback = std::function<void(const PtrConnection&, Buffer *)>;
                using ClosedCallback = std::function<void(const PtrConnection&)>;
                using AnyEventCallback = std::function<void(const PtrConnection&)>;
                ConnectedCallback _connected_callback;
                MessageCallback _message_callback;
                ClosedCallback _closed_callback;
                AnyEventCallback _event_callback;
                /*组件内的连接关闭回调--组件内设置的，因为服务器组件内会把所有的连接管理起来，一旦某个连接要关闭*/
                /*就应该从管理的地方移除掉自己的信息*/
                ClosedCallback _server_closed_callback;
        private:
        // /*五个channel的事件回调函数*/
        //描述符可读事件触发后调用的函数，接收socket数据放到接收缓冲区中，然后调用_message_callback
                void HandleRead() {
                }
                void HandleRead() {
                }
                void HandleClose() {
                }
                void HandleError() {
                }
                //描述符触发任意事件: 1. 刷新连接的活跃度--延迟定时销毁任务；  2. 调用组件使用者的任意事件回调
                void HandleEvent() { 
                }
                //连接获取之后，所处的状态下要进行各种设置（启动读监控,调用回调函数）
                void EstablishedInLoop() { 
                }
                //这个接口才是实际的释放接口
                void ReleaseInLoop() {
                }
                //这个接口并不是实际的发送接口，而只是把数据放到了发送缓冲区，启动了可写事件监控
                void SendInLoop(Buffer &buf) {
                }
                //这个关闭操作并非实际的连接释放操作，需要判断还有没有数据待处理，待发送
                void ShutdownInLoop() {
                }
                //启动非活跃连接超时释放规则
                void EnableInactiveReleaseInLoop(int sec) {
                }
                void CancelInactiveReleaseInLoop() {
                }
                void UpgradeInLoop(const Any &context, 
                        const ConnectedCallback &conn, 
                        const MessageCallback &msg, 
                        const ClosedCallback &closed, 
                        const AnyEventCallback &event) {
                        _context = context;
                        _connected_callback = conn;
                        _message_callback = msg;
                        _closed_callback = closed;
                        _event_callback = event;
                }
        public:
                Connection(EventLoop* loop,uint64_t _conn_id,int sockfd) :  _sockfd(sockfd),
                        _enable_inactive_release(false), _loop(loop), _statu(CONNECTING), _socket(_sockfd), _channel(loop, _sockfd) {
                        _channel.SetCloseCallback(std::bind(&Connection::HandleClose, this));
                        _channel.SetEventCallback(std::bind(&Connection::HandleEvent, this));
                        _channel.SetReadCallback(std::bind(&Connection::HandleRead, this));
                        _channel.SetWriteCallback(std::bind(&Connection::HandleWrite, this));
                        _channel.SetErrorCallback(std::bind(&Connection::HandleError, this));
                }
                 ~Connection() { DBG_LOG("RELEASE CONNECTION:%p", this); }
                  //获取管理的文件描述符
                 int Fd() {return _sockfd; }
                 // 获取连接ID
                 int Id() {return _conn_id; }
                 // 是否处于CONNECTED状态
                 bool Connected() { return (_statu == CONNECTED); }
                 //设置上下文--连接建立完成时进行调用
                void SetContext(const Any &context) { _context = context; }
                 //获取上下文，返回的是指针
                Any *GetContext() { return &_context; }
                void SetConnectedCallback(const ConnectedCallback&cb) { _connected_callback = cb; }
                void SetMessageCallback(const MessageCallback&cb) { _message_callback = cb; }
                void SetClosedCallback(const ClosedCallback&cb) { _closed_callback = cb; }
                void SetAnyEventCallback(const AnyEventCallback&cb) { _event_callback = cb; }
                void SetSrvClosedCallback(const ClosedCallback&cb) { _server_closed_callback = cb; }
                //连接建立就绪后，进行channel回调设置，启动读监控，调用_connected_callback
                void Established() {}
                _loop->RunInLoop(std::bind(&Connection::EstablishedInLoop, this));
                // 发送数据，将数据放到发送缓冲区，启动写事件监控
                void Send(const char *data, size_t len) {}
                    //提供给组件使用者的关闭接口--并不实际关闭，需要判断有没有数据待处理
                void Shutdown() {}
                void Release() {}
                //启动非活跃销毁，并定义多长时间无通信就是非活跃，添加定时任务
                void EnableInactiveRelease(int sec) { }
                 //取消非活跃销毁
                void CancelInactiveRelease() {}
                void Upgrade(const Any &context, const ConnectedCallback &conn, const MessageCallback &msg, 
                        const ClosedCallback &closed, const AnyEventCallback &event) {
                        _loop->AssertInLoop();
                        _loop->RunInLoop(std::bind(&Connection::UpgradeInLoop, this, context, conn, msg, closed, event));
        }

 };
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
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119

五、代码

class Connection;
//DISCONECTED -- 连接关闭状态；   CONNECTING -- 连接建立成功-待处理状态
//CONNECTED -- 连接建立完成，各种设置已完成，可以通信的状态；  DISCONNECTING -- 待关闭状态
typedef enum { DISCONNECTED, CONNECTING, CONNECTED, DISCONNECTING}ConnStatu;
using PtrConnection = std::shared_ptr<Connection>;
class Connection : public std::enable_shared_from_this<Connection> {
    private:
        uint64_t _conn_id;  // 连接的唯一ID，便于连接的管理和查找
        //uint64_t _timer_id;   //定时器ID，必须是唯一的，这块为了简化操作使用conn_id作为定时器ID
        int _sockfd;        // 连接关联的文件描述符
        bool _enable_inactive_release;  // 连接是否启动非活跃销毁的判断标志，默认为false
        EventLoop *_loop;   // 连接所关联的一个EventLoop
        ConnStatu _statu;   // 连接状态
        Socket _socket;     // 套接字操作管理
        Channel _channel;   // 连接的事件管理
        Buffer _in_buffer;  // 输入缓冲区---存放从socket中读取到的数据
        Buffer _out_buffer; // 输出缓冲区---存放要发送给对端的数据
        Any _context;       // 请求的接收处理上下文

        /*这四个回调函数，是让服务器模块来设置的（其实服务器模块的处理回调也是组件使用者设置的）*/
        /*换句话说，这几个回调都是组件使用者使用的*/
        using ConnectedCallback = std::function<void(const PtrConnection&)>;
        using MessageCallback = std::function<void(const PtrConnection&, Buffer *)>;
        using ClosedCallback = std::function<void(const PtrConnection&)>;
        using AnyEventCallback = std::function<void(const PtrConnection&)>;
        ConnectedCallback _connected_callback;
        MessageCallback _message_callback;
        ClosedCallback _closed_callback;
        AnyEventCallback _event_callback;
        /*组件内的连接关闭回调--组件内设置的，因为服务器组件内会把所有的连接管理起来，一旦某个连接要关闭*/
        /*就应该从管理的地方移除掉自己的信息*/
        ClosedCallback _server_closed_callback;
    private:
        /*五个channel的事件回调函数*/
        //描述符可读事件触发后调用的函数，接收socket数据放到接收缓冲区中，然后调用_message_callback
        void HandleRead() {
            //1. 接收socket的数据，放到缓冲区
            char buf[65536];
            ssize_t ret = _socket.NonBlockRecv(buf, 65535);
            if (ret < 0) {
                //出错了,不能直接关闭连接
                return ShutdownInLoop();
            }
            //这里的等于0表示的是没有读取到数据，而并不是连接断开了，连接断开返回的是-1
            //将数据放入输入缓冲区,写入之后顺便将写偏移向后移动
            _in_buffer.WriteAndPush(buf, ret);
            //2. 调用message_callback进行业务处理
            if (_in_buffer.ReadAbleSize() > 0) {
                //shared_from_this--从当前对象自身获取自身的shared_ptr管理对象
                return _message_callback(shared_from_this(), &_in_buffer);
            }
        }
        //描述符可写事件触发后调用的函数，将发送缓冲区中的数据进行发送
        void HandleWrite() {
            //_out_buffer中保存的数据就是要发送的数据
            ssize_t ret = _socket.NonBlockSend(_out_buffer.ReadPosition(), _out_buffer.ReadAbleSize());
            if (ret < 0) {
                //发送错误就该关闭连接了，
                if (_in_buffer.ReadAbleSize() > 0) {
                    _message_callback(shared_from_this(), &_in_buffer);
                }
                return Release();//这时候就是实际的关闭释放操作了。
            }
            _out_buffer.MoveReadOffset(ret);//千万不要忘了，将读偏移向后移动
            if (_out_buffer.ReadAbleSize() == 0) {
                _channel.DisableWrite();// 没有数据待发送了，关闭写事件监控
                //如果当前是连接待关闭状态，则有数据，发送完数据释放连接，没有数据则直接释放
                if (_statu == DISCONNECTING) {
                    return Release();
                }
            }
            return;
        }
        //描述符触发挂断事件
        void HandleClose() {
            /*一旦连接挂断了，套接字就什么都干不了了，因此有数据待处理就处理一下，完毕关闭连接*/
            if (_in_buffer.ReadAbleSize() > 0) {
                _message_callback(shared_from_this(), &_in_buffer);
            }
            return Release();
        }
        //描述符触发出错事件
        void HandleError() {
            return HandleClose();
        }
        //描述符触发任意事件: 1. 刷新连接的活跃度--延迟定时销毁任务；  2. 调用组件使用者的任意事件回调
        void HandleEvent() {
            if (_enable_inactive_release == true)  {  _loop->TimerRefresh(_conn_id); }
            if (_event_callback)  {  _event_callback(shared_from_this()); }
        }
        //连接获取之后，所处的状态下要进行各种设置（启动读监控,调用回调函数）
        void EstablishedInLoop() {
            // 1. 修改连接状态；  2. 启动读事件监控；  3. 调用回调函数
            assert(_statu == CONNECTING);//当前的状态必须一定是上层的半连接状态
            _statu = CONNECTED;//当前函数执行完毕，则连接进入已完成连接状态
            // 一旦启动读事件监控就有可能会立即触发读事件，如果这时候启动了非活跃连接销毁
            _channel.EnableRead();
            if (_connected_callback) _connected_callback(shared_from_this());
        }
        //这个接口才是实际的释放接口
        void ReleaseInLoop() {
            //1. 修改连接状态，将其置为DISCONNECTED
            _statu = DISCONNECTED;
            //2. 移除连接的事件监控
            _channel.Remove();
            //3. 关闭描述符
            _socket.Close();
            //4. 如果当前定时器队列中还有定时销毁任务，则取消任务
            if (_loop->HasTimer(_conn_id)) CancelInactiveReleaseInLoop();
            //5. 调用关闭回调函数，避免先移除服务器管理的连接信息导致Connection被释放，再去处理会出错，因此先调用用户的回调函数
            if (_closed_callback) _closed_callback(shared_from_this());
            //移除服务器内部管理的连接信息
            if (_server_closed_callback) _server_closed_callback(shared_from_this());
        }
        //这个接口并不是实际的发送接口，而只是把数据放到了发送缓冲区，启动了可写事件监控
        void SendInLoop(Buffer &buf) {
            if (_statu == DISCONNECTED) return ;
            _out_buffer.WriteBufferAndPush(buf);
            if (_channel.WriteAble() == false) {
                _channel.EnableWrite();
            }
        }
        //这个关闭操作并非实际的连接释放操作，需要判断还有没有数据待处理，待发送
        void ShutdownInLoop() {
            _statu = DISCONNECTING;// 设置连接为半关闭状态
            if (_in_buffer.ReadAbleSize() > 0) {
                if (_message_callback) _message_callback(shared_from_this(), &_in_buffer);
            }
            //要么就是写入数据的时候出错关闭，要么就是没有待发送数据，直接关闭
            if (_out_buffer.ReadAbleSize() > 0) {
                if (_channel.WriteAble() == false) {
                    _channel.EnableWrite();
                }
            }
            if (_out_buffer.ReadAbleSize() == 0) {
                Release();
            }
        }
        //启动非活跃连接超时释放规则
        void EnableInactiveReleaseInLoop(int sec) {
            //1. 将判断标志 _enable_inactive_release 置为true
            _enable_inactive_release = true;
            //2. 如果当前定时销毁任务已经存在，那就刷新延迟一下即可
            if (_loop->HasTimer(_conn_id)) {
                return _loop->TimerRefresh(_conn_id);
            }
            //3. 如果不存在定时销毁任务，则新增
            _loop->TimerAdd(_conn_id, sec, std::bind(&Connection::Release, this));
        }
        void CancelInactiveReleaseInLoop() {
            _enable_inactive_release = false;
            if (_loop->HasTimer(_conn_id)) { 
                _loop->TimerCancel(_conn_id); 
            }
        }
        void UpgradeInLoop(const Any &context, 
                    const ConnectedCallback &conn, 
                    const MessageCallback &msg, 
                    const ClosedCallback &closed, 
                    const AnyEventCallback &event) {
            _context = context;
            _connected_callback = conn;
            _message_callback = msg;
            _closed_callback = closed;
            _event_callback = event;
        }
    public:
        Connection(EventLoop *loop, uint64_t conn_id, int sockfd):_conn_id(conn_id), _sockfd(sockfd),
            _enable_inactive_release(false), _loop(loop), _statu(CONNECTING), _socket(_sockfd),
            _channel(loop, _sockfd) {
            _channel.SetCloseCallback(std::bind(&Connection::HandleClose, this));
            _channel.SetEventCallback(std::bind(&Connection::HandleEvent, this));
            _channel.SetReadCallback(std::bind(&Connection::HandleRead, this));
            _channel.SetWriteCallback(std::bind(&Connection::HandleWrite, this));
            _channel.SetErrorCallback(std::bind(&Connection::HandleError, this));
        }
        ~Connection() { DBG_LOG("RELEASE CONNECTION:%p", this); }
        //获取管理的文件描述符
        int Fd() { return _sockfd; }
        //获取连接ID
        int Id() { return _conn_id; }
        //是否处于CONNECTED状态
        bool Connected() { return (_statu == CONNECTED); }
        //设置上下文--连接建立完成时进行调用
        void SetContext(const Any &context) { _context = context; }
        //获取上下文，返回的是指针
        Any *GetContext() { return &_context; }
        void SetConnectedCallback(const ConnectedCallback&cb) { _connected_callback = cb; }
        void SetMessageCallback(const MessageCallback&cb) { _message_callback = cb; }
        void SetClosedCallback(const ClosedCallback&cb) { _closed_callback = cb; }
        void SetAnyEventCallback(const AnyEventCallback&cb) { _event_callback = cb; }
        void SetSrvClosedCallback(const ClosedCallback&cb) { _server_closed_callback = cb; }
        //连接建立就绪后，进行channel回调设置，启动读监控，调用_connected_callback
        void Established() {
            _loop->RunInLoop(std::bind(&Connection::EstablishedInLoop, this));
        }
        //发送数据，将数据放到发送缓冲区，启动写事件监控
        void Send(const char *data, size_t len) {
            //外界传入的data，可能是个临时的空间，我们现在只是把发送操作压入了任务池，有可能并没有被立即执行
            //因此有可能执行的时候，data指向的空间有可能已经被释放了。
            Buffer buf;
            buf.WriteAndPush(data, len);
            _loop->RunInLoop(std::bind(&Connection::SendInLoop, this, std::move(buf)));
        }
        //提供给组件使用者的关闭接口--并不实际关闭，需要判断有没有数据待处理
        void Shutdown() {
            _loop->RunInLoop(std::bind(&Connection::ShutdownInLoop, this));
        }
        void Release() {
            _loop->QueueInLoop(std::bind(&Connection::ReleaseInLoop, this));
        }
        //启动非活跃销毁，并定义多长时间无通信就是非活跃，添加定时任务
        void EnableInactiveRelease(int sec) {
            _loop->RunInLoop(std::bind(&Connection::EnableInactiveReleaseInLoop, this, sec));
        }
        //取消非活跃销毁
        void CancelInactiveRelease() {
            _loop->RunInLoop(std::bind(&Connection::CancelInactiveReleaseInLoop, this));
        }
        //切换协议---重置上下文以及阶段性回调处理函数 -- 而是这个接口必须在EventLoop线程中立即执行
        //防备新的事件触发后，处理的时候，切换任务还没有被执行--会导致数据使用原协议处理了。
        void Upgrade(const Any &context, const ConnectedCallback &conn, const MessageCallback &msg, 
                     const ClosedCallback &closed, const AnyEventCallback &event) {
            _loop->AssertInLoop();
            _loop->RunInLoop(std::bind(&Connection::UpgradeInLoop, this, context, conn, msg, closed, event));
        }
};
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
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227