-
Notifications
You must be signed in to change notification settings - Fork 0
/
TcpServer.cpp
101 lines (90 loc) · 2.96 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
#include"TcpServer.h"
#include"Acceptor.h"
#include"EventLoopThreadPool.h"
#include"base/Logging.h"
TcpServer::TcpServer(EventLoop* base,
const InetAddress& listenAddr,
const std::string& name,
bool option)
:loop_(base),
name_(name),
ipPort_(listenAddr.toIpPort()),
acceptor_(new Acceptor(loop_,listenAddr)),
threadPool_(new EventLoopThreadPool(loop_,name_)),
connections_(),
started_(false),
nextConnId_(1)
{
acceptor_->setNewConnectionCallback(std::bind(&TcpServer::newConnection,this,
std::placeholders::_1,
std::placeholders::_2));
}
TcpServer::~TcpServer()
{
loop_->assertInLoopThread();
Log<<"TcpServer::~TcpServer() ["<<name_<<"] destroying\n";
for (auto &item : connections_)
{
//for guard
TcpConnectionPtr ptr(item.second);
item.second.reset();
//in ioloop destroy conn
ptr->getLoop()->runInLoop(std::bind(&TcpConnection::connectDestroyed,ptr));
}
}
void TcpServer::setThreadNum(int numThreads)
{
assert(numThreads>=0);
threadPool_->setThreadNum(numThreads);
}
void TcpServer::start()
{
if(!started_)
{
started_=true;
threadPool_->start(initCallback_);
}
assert(!acceptor_->isListening());
loop_->runInLoop(std::bind(&Acceptor::listen,acceptor_.get()));
}
void TcpServer::newConnection(int sockfd,const InetAddress& peer)
{
loop_->assertInLoopThread();
EventLoop *ioloop=threadPool_->getNextLoop();
char buf[64];
InetAddress localAddr(sockets::getLocalAddr(sockfd));
snprintf(buf,sizeof buf,"%s%u",ipPort_.c_str(),nextConnId_);
++nextConnId_;
std::string name=name_+buf;
Log<<"TcpServer::newConnection ["<<name_
<<"] -new connection["<<name<<" ] from "<<peer.toIpPort()<<" \n";
TcpConnectionPtr conn(new TcpConnection(ioloop,name,sockfd,localAddr,peer));
conn->setConnectionCallback(connectioncallback_);
conn->setMessageCallback(messageCallback_);
conn->setCloseCallback(std::bind(&TcpServer::removeConnection,
this,
std::placeholders::_1));
//add setwritecompletecallback
connections_[name]=conn;
ioloop->runInLoop(std::bind(&TcpConnection::connectEstablished,conn));
}
void TcpServer::removeConnection(const TcpConnectionPtr& conn)
{
// if(loop_->isInLoopThread())
// {
// removeConnectionInLoop(conn);
// }
// else
// {
loop_->runInLoop(std::bind(&TcpServer::removeConnectionInLoop,this,conn));
// }
}
void TcpServer::removeConnectionInLoop(const TcpConnectionPtr& conn)
{
loop_->assertInLoopThread();
Log<<"TcpServer::removeConnectionInLoop ["<<conn->name()<<" ] \n";
size_t n=connections_.erase(conn->name());
assert(n==1);
EventLoop *ioloop=conn->getLoop();
ioloop->queueInLoop(std::bind(&TcpConnection::connectDestroyed,conn));
}