muduo库源码分析:TcpConnection 类

一.TcpConnection 成员及方法

源码:

TcpConnection.h

1. // Copyright 2010, Shuo Chen.  All rights reserved.
2. // http://code.google.com/p/muduo/
3. //
4. // Use of this source code is governed by a BSD-style license
5. // that can be found in the License file.
7. // Author: Shuo Chen (chenshuo at chenshuo dot com)
8. //
9. // This is a public header file, it must only include public header files.
11. #ifndef MUDUO_NET_TCPCONNECTION_H
12. #define MUDUO_NET_TCPCONNECTION_H
14. #include "muduo/base/noncopyable.h"
15. #include "muduo/base/StringPiece.h"
16. #include "muduo/base/Types.h"
17. #include "muduo/net/Callbacks.h"
18. #include "muduo/net/Buffer.h"
19. #include "muduo/net/InetAddress.h"
21. #include <memory>
23. #include <boost/any.hpp>
25. // struct tcp_info is in <netinet/tcp.h>
26. struct tcp_info;
28. namespace muduo
29. {
30. namespace net
31. {
33. class Channel;
34. class EventLoop;
35. class Socket;
37. ///
38. /// TCP connection, for both client and server usage.
39. ///
40. /// This is an interface class, so don't expose too much details.
41. class TcpConnection : noncopyable,
42.                       public std::enable_shared_from_this<TcpConnection>
43. {
44. public:
45.   /// Constructs a TcpConnection with a connected sockfd
46.   ///
47.   /// User should not create this object.
48.   TcpConnection(EventLoop* loop,
49.                 const string& name,
50.                 int sockfd,
51.                 const InetAddress& localAddr,
52.                 const InetAddress& peerAddr);
53.   ~TcpConnection();
55.   EventLoop* getLoop() const { return loop_; }
56.   const string& name() const { return name_; }
57.   const InetAddress& localAddress() const { return localAddr_; }
58.   const InetAddress& peerAddress() const { return peerAddr_; }
59.   bool connected() const { return state_ == kConnected; }
60.   bool disconnected() const { return state_ == kDisconnected; }
61.   // return true if success.
62.   bool getTcpInfo(struct tcp_info*) const;
63.   string getTcpInfoString() const;
65.   // void send(string&& message); // C++11
66.   void send(const void* message, int len);
67.   void send(const StringPiece& message);
68.   // void send(Buffer&& message); // C++11
69.   void send(Buffer* message);  // this one will swap data
70.   void shutdown(); // NOT thread safe, no simultaneous calling
71.   // void shutdownAndForceCloseAfter(double seconds); // NOT thread safe, no simultaneous calling
72.   void forceClose();
73.   void forceCloseWithDelay(double seconds);
74.   void setTcpNoDelay(bool on);
75.   // reading or not
76.   void startRead();
77.   void stopRead();
78.   bool isReading() const { return reading_; }; // NOT thread safe, may race with start/stopReadInLoop
80.   void setContext(const boost::any& context)
81.   { context_ = context; }
83.   const boost::any& getContext() const
84.   { return context_; }
86.   boost::any* getMutableContext()
87.   { return &context_; }
89.   void setConnectionCallback(const ConnectionCallback& cb)
90.   { connectionCallback_ = cb; }
92.   void setMessageCallback(const MessageCallback& cb)
93.   { messageCallback_ = cb; }
95.   void setWriteCompleteCallback(const WriteCompleteCallback& cb)
96.   { writeCompleteCallback_ = cb; }
98.   void setHighWaterMarkCallback(const HighWaterMarkCallback& cb, size_t highWaterMark)
99.   { highWaterMarkCallback_ = cb; highWaterMark_ = highWaterMark; }
101.   /// Advanced interface
102.   Buffer* inputBuffer()
103.   { return &inputBuffer_; }
105.   Buffer* outputBuffer()
106.   { return &outputBuffer_; }
108.   /// Internal use only.
109.   void setCloseCallback(const CloseCallback& cb)
110.   { closeCallback_ = cb; }
112.   // called when TcpServer accepts a new connection
113.   void connectEstablished();   // should be called only once
114.   // called when TcpServer has removed me from its map
115.   void connectDestroyed();  // should be called only once
117. private:
118.   enum StateE { kDisconnected, kConnecting, kConnected, kDisconnecting };
119.   void handleRead(Timestamp receiveTime);
120.   void handleWrite();
121.   void handleClose();
122.   void handleError();
123.   // void sendInLoop(string&& message);
124.   void sendInLoop(const StringPiece& message);
125.   void sendInLoop(const void* message, size_t len);
126.   void shutdownInLoop();
127.   // void shutdownAndForceCloseInLoop(double seconds);
128.   void forceCloseInLoop();
129.   void setState(StateE s) { state_ = s; }
130.   const char* stateToString() const;
131.   void startReadInLoop();
132.   void stopReadInLoop();
134.   EventLoop* loop_;
135.   const string name_;
136.   StateE state_;  // FIXME: use atomic variable
137.   bool reading_;
138.   // we don't expose those classes to client.
139.   std::unique_ptr<Socket> socket_;
140.   std::unique_ptr<Channel> channel_;
141.   const InetAddress localAddr_;
142.   const InetAddress peerAddr_;
143.   ConnectionCallback connectionCallback_;
144.   MessageCallback messageCallback_;
145.   WriteCompleteCallback writeCompleteCallback_;
146.   HighWaterMarkCallback highWaterMarkCallback_;
147.   CloseCallback closeCallback_;
148.   size_t highWaterMark_;
149.   Buffer inputBuffer_;
150.   Buffer outputBuffer_; // FIXME: use list<Buffer> as output buffer.
151.   boost::any context_;
152.   // FIXME: creationTime_, lastReceiveTime_
153.   //        bytesReceived_, bytesSent_
154. };
156. typedef std::shared_ptr<TcpConnection> TcpConnectionPtr;
158. }  // namespace net
159. }  // namespace muduo
161. #endif  // MUDUO_NET_TCPCONNECTION_H

复制代码

TcpConnection.cc

1. // Copyright 2010, Shuo Chen.  All rights reserved.
2. // http://code.google.com/p/muduo/
3. //
4. // Use of this source code is governed by a BSD-style license
5. // that can be found in the License file.
7. // Author: Shuo Chen (chenshuo at chenshuo dot com)
9. #include "muduo/net/TcpConnection.h"
11. #include "muduo/base/Logging.h"
12. #include "muduo/base/WeakCallback.h"
13. #include "muduo/net/Channel.h"
14. #include "muduo/net/EventLoop.h"
15. #include "muduo/net/Socket.h"
16. #include "muduo/net/SocketsOps.h"
18. #include <errno.h>
20. using namespace muduo;
21. using namespace muduo::net;
23. void muduo::net::defaultConnectionCallback(const TcpConnectionPtr& conn)
24. {
25.   LOG_TRACE << conn->localAddress().toIpPort() << " -> "
26.             << conn->peerAddress().toIpPort() << " is "
27.             << (conn->connected() ? "UP" : "DOWN");
28.   // do not call conn->forceClose(), because some users want to register message callback only.
29. }
31. void muduo::net::defaultMessageCallback(const TcpConnectionPtr&,
32.                                         Buffer* buf,
33.                                         Timestamp)
34. {
35.   buf->retrieveAll();
36. }
38. TcpConnection::TcpConnection(EventLoop* loop,
39.                              const string& nameArg,
40.                              int sockfd,
41.                              const InetAddress& localAddr,
42.                              const InetAddress& peerAddr)
43.   : loop_(CHECK_NOTNULL(loop)),
44.     name_(nameArg),
45.     state_(kConnecting),
46.     reading_(true),
47.     socket_(new Socket(sockfd)),
48.     channel_(new Channel(loop, sockfd)),
49.     localAddr_(localAddr),
50.     peerAddr_(peerAddr),
51.     highWaterMark_(64*1024*1024)
52. {
53.   channel_->setReadCallback(
54.       std::bind(&TcpConnection::handleRead, this, _1));
55.   channel_->setWriteCallback(
56.       std::bind(&TcpConnection::handleWrite, this));
57.   channel_->setCloseCallback(
58.       std::bind(&TcpConnection::handleClose, this));
59.   channel_->setErrorCallback(
60.       std::bind(&TcpConnection::handleError, this));
61.   LOG_DEBUG << "TcpConnection::ctor[" <<  name_ << "] at " << this
62.             << " fd=" << sockfd;
63.   socket_->setKeepAlive(true);
64. }
66. TcpConnection::~TcpConnection()
67. {
68.   LOG_DEBUG << "TcpConnection::dtor[" <<  name_ << "] at " << this
69.             << " fd=" << channel_->fd()
70.             << " state=" << stateToString();
71.   assert(state_ == kDisconnected);
72. }
74. bool TcpConnection::getTcpInfo(struct tcp_info* tcpi) const
75. {
76.   return socket_->getTcpInfo(tcpi);
77. }
79. string TcpConnection::getTcpInfoString() const
80. {
81.   char buf[1024];
82.   buf[0] = '\0';
83.   socket_->getTcpInfoString(buf, sizeof buf);
84.   return buf;
85. }
87. void TcpConnection::send(const void* data, int len)
88. {
89.   send(StringPiece(static_cast<const char*>(data), len));
90. }
92. void TcpConnection::send(const StringPiece& message)
93. {
94.   if (state_ == kConnected)
95.   {
96.     if (loop_->isInLoopThread())
97.     {
98.       sendInLoop(message);
99.     }
100.     else
101.     {
102.       void (TcpConnection::*fp)(const StringPiece& message) = &TcpConnection::sendInLoop;
103.       loop_->runInLoop(
104.           std::bind(fp,
105.                     this,     // FIXME
106.                     message.as_string()));
107.                     //std::forward<string>(message)));
108.     }
109.   }
110. }
112. // FIXME efficiency!!!
113. void TcpConnection::send(Buffer* buf)
114. {
115.   if (state_ == kConnected)
116.   {
117.     if (loop_->isInLoopThread())
118.     {
119.       sendInLoop(buf->peek(), buf->readableBytes());
120.       buf->retrieveAll();
121.     }
122.     else
123.     {
124.       void (TcpConnection::*fp)(const StringPiece& message) = &TcpConnection::sendInLoop;
125.       loop_->runInLoop(
126.           std::bind(fp,
127.                     this,     // FIXME
128.                     buf->retrieveAllAsString()));
129.                     //std::forward<string>(message)));
130.     }
131.   }
132. }
134. void TcpConnection::sendInLoop(const StringPiece& message)
135. {
136.   sendInLoop(message.data(), message.size());
137. }
139. void TcpConnection::sendInLoop(const void* data, size_t len)
140. {
141.   loop_->assertInLoopThread();
142.   ssize_t nwrote = 0;
143.   size_t remaining = len;
144.   bool faultError = false;
145.   if (state_ == kDisconnected)
146.   {
147.     LOG_WARN << "disconnected, give up writing";
148.     return;
149.   }
150.   // if no thing in output queue, try writing directly
151.   if (!channel_->isWriting() && outputBuffer_.readableBytes() == 0)
152.   {
153.     nwrote = sockets::write(channel_->fd(), data, len);
154.     if (nwrote >= 0)
155.     {
156.       remaining = len - nwrote;
157.       if (remaining == 0 && writeCompleteCallback_)
158.       {
159.         loop_->queueInLoop(std::bind(writeCompleteCallback_, shared_from_this()));
160.       }
161.     }
162.     else // nwrote < 0
163.     {
164.       nwrote = 0;
165.       if (errno != EWOULDBLOCK)
166.       {
167.         LOG_SYSERR << "TcpConnection::sendInLoop";
168.         if (errno == EPIPE || errno == ECONNRESET) // FIXME: any others?
169.         {
170.           faultError = true;
171.         }
172.       }
173.     }
174.   }
176.   assert(remaining <= len);
177.   if (!faultError && remaining > 0)
178.   {
179.     size_t oldLen = outputBuffer_.readableBytes();
180.     if (oldLen + remaining >= highWaterMark_
181.         && oldLen < highWaterMark_
182.         && highWaterMarkCallback_)
183.     {
184.       loop_->queueInLoop(std::bind(highWaterMarkCallback_, shared_from_this(), oldLen + remaining));
185.     }
186.     outputBuffer_.append(static_cast<const char*>(data)+nwrote, remaining);
187.     if (!channel_->isWriting())
188.     {
189.       channel_->enableWriting();
190.     }
191.   }
192. }
194. void TcpConnection::shutdown()
195. {
196.   // FIXME: use compare and swap
197.   if (state_ == kConnected)
198.   {
199.     setState(kDisconnecting);
200.     // FIXME: shared_from_this()?
201.     loop_->runInLoop(std::bind(&TcpConnection::shutdownInLoop, this));
202.   }
203. }
205. void TcpConnection::shutdownInLoop()
206. {
207.   loop_->assertInLoopThread();
208.   if (!channel_->isWriting())
209.   {
210.     // we are not writing
211.     socket_->shutdownWrite();
212.   }
213. }
215. // void TcpConnection::shutdownAndForceCloseAfter(double seconds)
216. // {
217. //   // FIXME: use compare and swap
218. //   if (state_ == kConnected)
219. //   {
220. //     setState(kDisconnecting);
221. //     loop_->runInLoop(std::bind(&TcpConnection::shutdownAndForceCloseInLoop, this, seconds));
222. //   }
223. // }
225. // void TcpConnection::shutdownAndForceCloseInLoop(double seconds)
226. // {
227. //   loop_->assertInLoopThread();
228. //   if (!channel_->isWriting())
229. //   {
230. //     // we are not writing
231. //     socket_->shutdownWrite();
232. //   }
233. //   loop_->runAfter(
234. //       seconds,
235. //       makeWeakCallback(shared_from_this(),
236. //                        &TcpConnection::forceCloseInLoop));
237. // }
239. void TcpConnection::forceClose()
240. {
241.   // FIXME: use compare and swap
242.   if (state_ == kConnected || state_ == kDisconnecting)
243.   {
244.     setState(kDisconnecting);
245.     loop_->queueInLoop(std::bind(&TcpConnection::forceCloseInLoop, shared_from_this()));
246.   }
247. }
249. void TcpConnection::forceCloseWithDelay(double seconds)
250. {
251.   if (state_ == kConnected || state_ == kDisconnecting)
252.   {
253.     setState(kDisconnecting);
254.     loop_->runAfter(
255.         seconds,
256.         makeWeakCallback(shared_from_this(),
257.                          &TcpConnection::forceClose));  // not forceCloseInLoop to avoid race condition
258.   }
259. }
261. void TcpConnection::forceCloseInLoop()
262. {
263.   loop_->assertInLoopThread();
264.   if (state_ == kConnected || state_ == kDisconnecting)
265.   {
266.     // as if we received 0 byte in handleRead();
267.     handleClose();
268.   }
269. }
271. const char* TcpConnection::stateToString() const
272. {
273.   switch (state_)
274.   {
275.     case kDisconnected:
276.       return "kDisconnected";
277.     case kConnecting:
278.       return "kConnecting";
279.     case kConnected:
280.       return "kConnected";
281.     case kDisconnecting:
282.       return "kDisconnecting";
283.     default:
284.       return "unknown state";
285.   }
286. }
288. void TcpConnection::setTcpNoDelay(bool on)
289. {
290.   socket_->setTcpNoDelay(on);
291. }
293. void TcpConnection::startRead()
294. {
295.   loop_->runInLoop(std::bind(&TcpConnection::startReadInLoop, this));
296. }
298. void TcpConnection::startReadInLoop()
299. {
300.   loop_->assertInLoopThread();
301.   if (!reading_ || !channel_->isReading())
302.   {
303.     channel_->enableReading();
304.     reading_ = true;
305.   }
306. }
308. void TcpConnection::stopRead()
309. {
310.   loop_->runInLoop(std::bind(&TcpConnection::stopReadInLoop, this));
311. }
313. void TcpConnection::stopReadInLoop()
314. {
315.   loop_->assertInLoopThread();
316.   if (reading_ || channel_->isReading())
317.   {
318.     channel_->disableReading();
319.     reading_ = false;
320.   }
321. }
323. void TcpConnection::connectEstablished()
324. {
325.   loop_->assertInLoopThread();
326.   assert(state_ == kConnecting);
327.   setState(kConnected);
328.   channel_->tie(shared_from_this());
329.   channel_->enableReading();
331.   connectionCallback_(shared_from_this());
332. }
334. void TcpConnection::connectDestroyed()
335. {
336.   loop_->assertInLoopThread();
337.   if (state_ == kConnected)
338.   {
339.     setState(kDisconnected);
340.     channel_->disableAll();
342.     connectionCallback_(shared_from_this());
343.   }
344.   channel_->remove();
345. }
347. void TcpConnection::handleRead(Timestamp receiveTime)
348. {
349.   loop_->assertInLoopThread();
350.   int savedErrno = 0;
351.   ssize_t n = inputBuffer_.readFd(channel_->fd(), &savedErrno);
352.   if (n > 0)
353.   {
354.     messageCallback_(shared_from_this(), &inputBuffer_, receiveTime);
355.   }
356.   else if (n == 0)
357.   {
358.     handleClose();
359.   }
360.   else
361.   {
362.     errno = savedErrno;
363.     LOG_SYSERR << "TcpConnection::handleRead";
364.     handleError();
365.   }
366. }
368. void TcpConnection::handleWrite()
369. {
370.   loop_->assertInLoopThread();
371.   if (channel_->isWriting())
372.   {
373.     ssize_t n = sockets::write(channel_->fd(),
374.                                outputBuffer_.peek(),
375.                                outputBuffer_.readableBytes());
376.     if (n > 0)
377.     {
378.       outputBuffer_.retrieve(n);
379.       if (outputBuffer_.readableBytes() == 0)
380.       {
381.         channel_->disableWriting();
382.         if (writeCompleteCallback_)
383.         {
384.           loop_->queueInLoop(std::bind(writeCompleteCallback_, shared_from_this()));
385.         }
386.         if (state_ == kDisconnecting)
387.         {
388.           shutdownInLoop();
389.         }
390.       }
391.     }
392.     else
393.     {
394.       LOG_SYSERR << "TcpConnection::handleWrite";
395.       // if (state_ == kDisconnecting)
396.       // {
397.       //   shutdownInLoop();
398.       // }
399.     }
400.   }
401.   else
402.   {
403.     LOG_TRACE << "Connection fd = " << channel_->fd()
404.               << " is down, no more writing";
405.   }
406. }
408. void TcpConnection::handleClose()
409. {
410.   loop_->assertInLoopThread();
411.   LOG_TRACE << "fd = " << channel_->fd() << " state = " << stateToString();
412.   assert(state_ == kConnected || state_ == kDisconnecting);
413.   // we don't close fd, leave it to dtor, so we can find leaks easily.
414.   setState(kDisconnected);
415.   channel_->disableAll();
417.   TcpConnectionPtr guardThis(shared_from_this());
418.   connectionCallback_(guardThis);
419.   // must be the last line
420.   closeCallback_(guardThis);
421. }
423. void TcpConnection::handleError()
424. {
425.   int err = sockets::getSocketError(channel_->fd());
426.   LOG_ERROR << "TcpConnection::handleError [" << name_
427.             << "] - SO_ERROR = " << err << " " << strerror_tl(err);
428. }

复制代码

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。