-
Notifications
You must be signed in to change notification settings - Fork 0
/
TcpServer.cpp
129 lines (113 loc) · 4.26 KB
/
TcpServer.cpp
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
//=====================================================================
//
// TcpServer.cpp -
//
// Created by wwq on 2022/08/21
// Last Modified: 2022/08/21 14:50:38
//
//=====================================================================
#include "TcpServer.h"
#include "Acceptor.h"
#include "InetAddress.h"
#include "Logger.h"
#include "TcpConnection.h"
#include "SocketsOps.h"
#include <functional>
#include <string>
#include <string.h>
using namespace simple_muduo;
static EventLoop* CheckLoopNotNull(EventLoop* loop)
{
if(loop == nullptr)
{
LOG_ERROR("mainLoop is null\n");
}
return loop;
}
TcpServer::TcpServer(EventLoop* loop,
const InetAddress& listenAddr,
const std::string& nameArg,
Option option)
: loop_(CheckLoopNotNull(loop))
, ipPort_(listenAddr.toIpPort())
, name_(nameArg)
, acceptor_(new Acceptor(loop, listenAddr, option == kReusePort))
, threadPool_(new EventLoopThreadPool(loop, name_))
, connectionCallback_()
, messageCallback_()
, started_(0)
, nextConnId_(1)
{
// 当有新的客户端连接到server后,Acceptor负责管理,绑定的
// acceptChannel_会有可读事件发生,执行handleRead()调用TcpServer::newConnection回调函数
acceptor_->setNewConnectionCallback(
std::bind(&TcpServer::newConnection, this, std::placeholders::_1, std::placeholders::_2));
}
TcpServer::~TcpServer()
{
for(auto &item : connections_)
{
TcpConnectionPtr conn(item.second);
// 把原始的智能指针复位
// 让栈空间的TcpConnectionPtr conn指向该对象
// 当conn出了其作用域 即可释放智能指针指向的对象
item.second.reset();
// 销毁连接
conn->getLoop()->runInLoop(std::bind(&TcpConnection::connectDestroyed, conn));
}
}
void TcpServer::setThreadNum(int numThreads)
{
threadPool_->setThreadNum(numThreads);
}
void TcpServer::start()
{
if(started_++ == 0)
{
// 启动底层的loop线程池
loop_->runInLoop(std::bind(&Acceptor::listen, acceptor_.get()));
}
}
// 有一个新用户连接,acceptor会执行这个回调操作,负责将mainLoop接收到的请求连接(acceptChannel_会有读事件发生)通过回调轮询分发给subLoop去处理
void TcpServer::newConnection(int sockfd, const InetAddress& peerAddr)
{
// 轮询算法 选择一个subLoop 来管理connfd对应的channel
EventLoop *ioLoop = threadPool_->getNextLoop();
char buf[64] = {0};
snprintf(buf, sizeof buf, "-%s#%d", ipPort_.c_str(), nextConnId_);
++nextConnId_; // 这里没有设置为原子类是因为其只在mainloop中执行 不涉及线程安全问题
std::string connName = name_ + buf;
LOG_INFO("TcpServer::newConnection [%s] - new connection [%s] from %s\n",
name_.c_str(), connName.c_str(), peerAddr.toIpPort().c_str());
// 通过sockfd获取其绑定的本机的ip地址和端口信息
InetAddress localAddr(sockets::getLocalAddr(sockfd));
TcpConnectionPtr conn(new TcpConnection(ioLoop,
connName,
sockfd,
localAddr,
peerAddr));
connections_[connName] = conn;
// 下面的回调都是用户设置给TcpServer => TcpConnection的,至于Channel绑定的则是TcpConnection设置的四个,handleRead,handleWrite... 这下面的回调用于handlexxx函数中
conn->setConnectionCallback(connectionCallback_);
conn->setMessageCallback(messageCallback_);
conn->setWriteCompleteCallback(writeCompleteCallback_);
// 设置了如何关闭连接的回调
conn->setCloseCallback(
std::bind(&TcpServer::removeConnection, this, std::placeholders::_1));
ioLoop->runInLoop(
std::bind(&TcpConnection::connectEstablished, conn));
}
void TcpServer::removeConnection(const TcpConnectionPtr &conn)
{
loop_->runInLoop(
std::bind(&TcpServer::removeConnectionInLoop, this, conn));
}
void TcpServer::removeConnectionInLoop(const TcpConnectionPtr &conn)
{
LOG_INFO("TcpServer::removeConnectionInLoop [%s] - connection %s\n",
name_.c_str(), conn->name().c_str());
connections_.erase(conn->name());
EventLoop *ioLoop = conn->getLoop();
ioLoop->queueInLoop(
std::bind(&TcpConnection::connectDestroyed, conn));
}