multiple reactors(35 -1 )(四)
CE << "[1] usecount=" << conn.use_count();
connections_[connName] = conn;
LOG_TRACE << "[2] usecount=" << conn.use_count();
conn->setConnectionCallback(connectionCallback_);
conn->setMessageCallback(messageCallback_);
conn->setCloseCallback(
boost::bind(&TcpServer::removeConnection, this, _1));
// conn->connectEstablished(); 这个表示直接在当前线程中调用
ioLoop->runInLoop(boost::bind(&TcpConnection::connectEstablished, conn));
LOG_TRACE << "[5] usecount=" << conn.use_count();
}
void TcpServer::removeConnection(const TcpConnectionPtr& conn)
{
/*
loop_->assertInLoopThread();
LOG_INFO << "TcpServer::removeConnectionInLoop [" << name_
<< "] - connection " << conn->name();
LOG_TRACE << "[8] usecount=" << conn.use_count();
size_t n = connections_.erase(conn->name());
LOG_TRACE << "[9] usecount=" << conn.use_count();
(void)n;
assert(n == 1);
loop_->queueInLoop(
boost::bind(&TcpConnection::connectDestroyed, conn));
LOG_TRACE << "[10] usecount=" << conn.use_count();
*/
loop_->runInLoop(boost::bind(&TcpServer::removeConnectionInLoop, this, conn));
}
void TcpServer::removeConnectionInLoop(const TcpConnectionPtr& conn)
{
loop_->assertInLoopThread();
LOG_INFO << "TcpServer::removeConnectionInLoop [" << name_
<< "] - connection " << conn->name();
LOG_TRACE << "[8] usecount=" << conn.use_count();
size_t n = connections_.erase(conn->name());
LOG_TRACE << "[9] usecount=" << conn.use_count();
(void)n;
assert(n == 1);
EventLoop* ioLoop = conn->getLoop();
ioLoop->queueInLoop(
boost::bind(&TcpConnection::connectDestroyed, conn));
//loop_->queueInLoop(
// boost::bind(&TcpConnection::connectDestroyed, conn));
LOG_TRACE << "[10] usecount=" << conn.use_count();
}
测试程序
[cpp]
#include
#include
#include
#include
#include
using namespace muduo;
using namespace muduo::net;
class TestServer
{
public:
TestServer(EventLoop* loop,
const InetAddress& listenAddr, int numThreads)
: loop_(loop),
server_(loop, listenAddr, "TestServer"),
numThreads_(numThreads)
{
server_.setConnectionCallback(
boost::bind(&TestServer::onConnection, this, _1));
server_.setMessageCallback(
boost::bind(&TestServer::onMessage, this, _1, _2, _3));
server_.setThreadNum(numThreads);
}
void start()
{
server_.start();
}
private:
void onConnection(const TcpConnectionPtr& conn)
{
if (conn->connected())
{
printf("onConnection(): new connection [%s] from %s\n",
conn->name().c_str(),
conn->peerAddress().toIpPort().c_str());
}
else
{
printf("onConnection(): connection [%s] is down\n",
conn->name().c_str());
}
}
void onMessage(const TcpConnectionPtr& conn,
const char* data,
ssize_t len)
{
printf("onMessage(): received %zd bytes from connection [%s]\n",
len, conn->name().c_str());
}
EventLoop* loop_;
TcpServer server_;
int numThreads_;
};
int main()
{
printf("main(): pid = %d\n", getpid());
InetAddress listenAddr(8888);
EventLoop loop;
TestServer server(&loop, listenAddr,4);
server.start();
loop.loop();
}