当前位置:网站首页>Tcpconnection class for source code analysis of Muduo Network Library

Tcpconnection class for source code analysis of Muduo Network Library

2020-12-07 17:44:38 osc_ oq0m8kxk

Used to manage a specific TCP Connect , For example, receiving and sending messages , Complete the user specified connection callback connectionCallback.

TcpConnection When constructing, the receiving parameters are TCP Connected sockfd, Server address localAddr, Client address peerAddr, And pass Socket encapsulation sockfd. And use Channel Management should sockfd, towards Channel Register for readability 、 Can write 、 close 、 Error callback function , be used for Poller After returning to the ready event Channel::handleEvent() Execute the callback of the corresponding event .
TcpConnection There are four states , A simple state diagram :
 Picture description here

TcpConnection There are a series of user specified event callback functions , such as TcpConnection::connectionCallbackmessageCallbackwriteCompleteCallback, These are users through TcpServer Pass to TcpConnection. When Poller return TcpConnection Corresponding Socket Ready event ,Channel::handleEvent() -> TcpConnection::handle* Series of functions -> These events call back functions ( Such as connectionCallback).

Corresponding to the above ,TcpConnection There is a series of be used for TcpServer Specify events for the connection callback Function of , such as TcpConnection::setConnectionCallbacksetCloseCallback wait , Is in TcpServer::newConncetion() Create a TcpConnection Object and set the callback function specified by the user ( It was said in the previous paragraph ) adopt TcpConnection::set*Callback The function is passed to TcpConnection.

TcpConncetion There are also a series of functions in the sockfd Event callback function on , such as TcpConnection::handleRead() Is in Poller When a readable event is returned by Channel::handleEvent() Called , Similar to that handleWrite()handleClose()handleError(), These functions call the user specified event callback function ( It was said in the last paragraph ), such as TcpConnection::handleRead() -> messageCallback(). These event callback functions are TcpConnection Created during construction sockfd Corresponding Channel After passage Channel::set*Callback Series of functions registered .
TcpConnection::handleRead() : When the connection corresponds to sockfd Called when a readable event occurs , It mainly reads data to Buffer in , Execute message callback function messageCallback_().
TcpConnection::handleWrite(): When the connection corresponds to sockfd Called when a writable event occurs , Mainly is to Buffer Send the data in , If one-time delivery is completed , The user specified callback is executed writeCompleteCallback_(), If one time is not sent out , muduo use LT Pattern , A writable event is triggered repeatedly , Next time there's a chance to send the rest of the data .
TcpConnection::handleClose(): Main execution Channel::disableAll() and TcpConnection::closeCallback().
TcpConnection::handleError(): Mainly in the log output error information .



TcpConnection::closeCallback() It's not for ordinary users , This is for TcpServer and TcpClient With , Used to notify them to remove the TcpConnectionPtr. It's bound to TcpServer::removeConnection(), Through the following relationship :Accetptor Accept one Tcp When the connection Channel::handleEvent() -> Acceptor::handleRead() -> TcpServer::newConnection() Create a new one in TcpConnection And pass TcpConnection::setCloseCallback(bind(&TcpServer::removeConnectionCallback, this, _1)) . When will this callback be called ? When TcpConnection::handleRead() in read return 0, perhaps sockfd happen POLLHUP When the event is ready , Will call TcpConnection::handleClose() -> TcpConnection::closeCallback_() -> TcpServer::removeConnection().

TcpConnection There are two more functions in the TcpServer The use of .
TcpConnection::connectEstablishd(), yes TcpServer::newConnection() Created TcpConnection object , After the callback function is set, it is called. , Mainly called Channel::enableReading() take TcpConnection Corresponding sockfd Register to read Events , The specified user is then executed connectionCallback_(), And will TcpConnection The state is set to kConnected. How is this function executed ? Acceptor Held lfd A readable event occurred , Immediate connection request , here Channel::handleEvent() -> Acceptor::handleRead() -> TcpServer::newConnection() -> ……->TcpConnection::connectEstablished(). The part omitted in the middle is thread transfer , Transferred to the TcpConnection Where IO Threads execute ,TcpServer and TcpConnection May not be on the same thread .
TcpConnection::connectDestroyed(), This is a TcpConnection A member function that was last called before destructing , It notifies the user that the connection has been disconnected . The main function is setting TcpConnection The status of is kDisconnected; Stop listening for all events ; call connectionCallback_() Performs a user specified callback ; from epoll Monitoring fd Remove TcpConnection Corresponding sockfd. When will this function be called ? One is TcpServer Deconstruction time , In general, it is through TcpConnection::handleClose() -> TcpConnection::closeCallback_() -> TcpServer::removeConnection() -> ……->TcpConnection::connectDestroyed().

send A series of functions It can be called by the user or other threads , For sending messages . If not in IO Threads , It will transfer the actual work to IO Thread calls . First check TcpConnection Corresponding Socket Is a writable event registered , If a writable event is registered, the output buffer outputBuffer_ There is already data waiting to be sent in , In order to ensure no disorder , This time the data is just appended to the output buffer , adopt Channel::handleEvent() -> TcpConnection::handleWrite() To send the . If Socket No writable events are registered , Output buffer has no data , Then this time the news can be directly passed through write send out , If it is not sent all at once , that message The rest of the data is still needed append To outputBuffer in , And to Poller Register writable Events , When socket When it becomes writable ,Channel Would call TcpConnection::handleWrite() To send the outputBuffer_ Data piled up in , Stop listening for writable events immediately after sending , avoid busy loop. Whether it's sendInLoop() -> write() still Channel::handleEvent() -> handleWrite(), Just make sure it's finished message perhaps outputBuffer_ Data in , The callback specified by the user must be called writeCompleteCallback().
Use epoll Of LT Pattern , When socket Writable time , It's going to trigger socket The writable event of , How to solve this problem at this time ? The first way : You need to socket When writing data , Sign up for this socket The writable event of , After receiving a writable event , And then call write/send Writing data , When all the data is written , Stop observing immediately writable event , avoid busy loop.. The second kind , When data needs to be sent , Call directly write/send Write , If it's not finished , So open is monitoring this socket Of writable event , After receiving a writable event , call write/send send data , When all the data is written , Stop observing immediately writable event , avoid busy loop. muduo Adopted LT Pattern , It's the second way to solve this problem .

Besides TcpConnection There are also several small features , such as TcpConnection::setTcpNoDelay() and TcpConnection::setKeepAlive().TCP No Delay and TCP keepalive They are commonly used TCP Options , The function of the former is to prohibit Nagle Algorithm , Avoid delay in continuous contract awarding , This is important for writing low latency network services . The function of the latter is to explore regularly TCP Is the connection still there , Generally speaking, if there is an application layer heartbeat ,TCP keepalive It's not necessary .

Buffer class The design will be analyzed later .

TcpConnection.h

#ifndef MUDUO_NET_TCPCONNECTION_H
#define MUDUO_NET_TCPCONNECTION_H

#include <muduo/base/StringPiece.h>
#include <muduo/base/Types.h>
#include <muduo/net/Callbacks.h>
#include <muduo/net/Buffer.h>
#include <muduo/net/InetAddress.h>

#include <boost/any.hpp>
#include <boost/enable_shared_from_this.hpp>
#include <boost/noncopyable.hpp>
#include <boost/scoped_ptr.hpp>
#include <boost/shared_ptr.hpp>

// struct tcp_info is in <netinet/tcp.h>
struct tcp_info;

namespace muduo
{
namespace net
{

class Channel;
class EventLoop;
class Socket;

///
/// TCP connection, for both client and server usage.
///
/// This is an interface class, so don't expose too much details.
class TcpConnection : boost::noncopyable,
                      public boost::enable_shared_from_this<TcpConnection>
{
 public:
  /// Constructs a TcpConnection with a connected sockfd
  ///
  /// User should not create this object.
  TcpConnection(EventLoop* loop,
                const string& name,
                int sockfd,
                const InetAddress& localAddr,
                const InetAddress& peerAddr);
  ~TcpConnection();

  EventLoop* getLoop() const { return loop_; }
  const string& name() const { return name_; }
  const InetAddress& localAddress() const { return localAddr_; }
  const InetAddress& peerAddress() const { return peerAddr_; }
  bool connected() const { return state_ == kConnected; }
  bool disconnected() const { return state_ == kDisconnected; }
  // return true if success.
  bool getTcpInfo(struct tcp_info*) const;
  string getTcpInfoString() const;

  // void send(string&& message); // C++11
  void send(const void* message, int len);
  void send(const StringPiece& message);
  // void send(Buffer&& message); // C++11
  void send(Buffer* message);  // this one will swap data
  void shutdown(); // NOT thread safe, no simultaneous calling
  // void shutdownAndForceCloseAfter(double seconds); // NOT thread safe, no simultaneous calling
  void forceClose();
  void forceCloseWithDelay(double seconds);
  void setTcpNoDelay(bool on);
  // reading or not
  void startRead();
  void stopRead();
  bool isReading() const { return reading_; }; // NOT thread safe, may race with start/stopReadInLoop

  /*  Set up TCP Context  */ 
  void setContext(const boost::any& context)
  { context_ = context; }

  /*  obtain TCP Context  */
  const boost::any& getContext() const
  { return context_; }

  boost::any* getMutableContext()
  { return &context_; }

  /*  Set the callback function when the connection is established  */
  void setConnectionCallback(const ConnectionCallback& cb)
  { connectionCallback_ = cb; }

  /*  Set the callback function of message arrival  */
  void setMessageCallback(const MessageCallback& cb)
  { messageCallback_ = cb; }

  /*  Set the callback function when the data is successfully written to the other party  */
  void setWriteCompleteCallback(const WriteCompleteCallback& cb)
  { writeCompleteCallback_ = cb; }

  /*  Set the high water level callback function and high water level value , When the size achieve highWaterMark This request is triggered when  */
  void setHighWaterMarkCallback(const HighWaterMarkCallback& cb, size_t highWaterMark)
  { highWaterMarkCallback_ = cb; highWaterMark_ = highWaterMark; }

  /// Advanced interface
  /*  Return input buffer and output buffer pointer  */
  Buffer* inputBuffer()
  { return &inputBuffer_; }

  Buffer* outputBuffer()
  { return &outputBuffer_; }

  /// Internal use only.
  /*  Set up TCP Closed callback function , For internal use only , Used to remove held TcpConnectionPtr */
  void setCloseCallback(const CloseCallback& cb)
  { closeCallback_ = cb; }

  // called when TcpServer accepts a new connection
  /* TcpServer Use , Call after creating a new connection  */
  void connectEstablished();   // should be called only once
  // called when TcpServer has removed me from its map
  /* TcpServer Use , from  map  Called when deleted from  */
  void connectDestroyed();  // should be called only once

 private:
  /* TcpConnection  There are four states , From left to right : Disconnected   on connection   Connected   Disconnecting  */
  enum StateE { kDisconnected, kConnecting, kConnected, kDisconnecting };

  /*  Handle  connfd  Event callback function on  */
  void handleRead(Timestamp receiveTime);   //  Handling readable events 
  void handleWrite();   //  Handling writable events 
  void handleClose();   //  Handling shutdown events 
  void handleError();   //  Handling error events 

  /*  It can send messages safely through thread transfer operation  */
  // void sendInLoop(string&& message);
  void sendInLoop(const StringPiece& message);
  void sendInLoop(const void* message, size_t len);
  /*  Safe shutdown through thread transfer operation TCP Connect  */
  void shutdownInLoop();
  // void shutdownAndForceCloseInLoop(double seconds);
  /*  Active close connection  */
  void forceCloseInLoop();
  /*  Set up TCP The state of the connection  */
  void setState(StateE s) { state_ = s; }
  const char* stateToString() const;
  void startReadInLoop();
  void stopReadInLoop();

  EventLoop* loop_;     //  Handle the TCP Connected EventLoop, The EventLoop Inside epoll monitor TCP Connect the corresponding fd
  const string name_;   // TCP The name of the connection 
  StateE state_;  // FIXME: use atomic variable     //  This article TCP The state of the connection 
  bool reading_;        // 
  // we don't expose those classes to client.
  boost::scoped_ptr<Socket> socket_;    // TCP Connected fd Where socket object ,fd It's up to it to close 
  boost::scoped_ptr<Channel> channel_;  // TCP Connected fd  Corresponding  Channel
  const InetAddress localAddr_;         // TCP Connect to local address :ip port
  const InetAddress peerAddr_;          // TCP Address of connecting party :ip port
  ConnectionCallback connectionCallback_;   //  Callback function when connection is established 
  MessageCallback messageCallback_;         //  Callback function when message is received 
  WriteCompleteCallback writeCompleteCallback_;     //  The callback function when the message is written to the other buffer 
  HighWaterMarkCallback highWaterMarkCallback_;     //  High water level callback function 
  CloseCallback closeCallback_;         //  close TCP Connected callback function 
  size_t highWaterMark_;    //  High water mark 
  Buffer inputBuffer_;      // TCP Connected input buffer , Read the input from the connection and store it here 
  Buffer outputBuffer_;     // TCP Connected output buffer , The data to be sent is saved here 
  // FIXME: use list<Buffer> as output buffer.
  boost::any context_;      // TCP The context of the connection , It is generally used to deal with the situation that multiple messages are related to each other , For example, file sending 
  // FIXME: creationTime_, lastReceiveTime_
  //        bytesReceived_, bytesSent_
};

/*  Use  shared_ptr  To manage  TCP  The lifetime of the connection object  */
typedef boost::shared_ptr<TcpConnection> TcpConnectionPtr;

}
}

#endif  // MUDUO_NET_TCPCONNECTION_H

TcpConnection.cc

#include <muduo/net/TcpConnection.h>

#include <muduo/base/Logging.h>
#include <muduo/base/WeakCallback.h>
#include <muduo/net/Channel.h>
#include <muduo/net/EventLoop.h>
#include <muduo/net/Socket.h>
#include <muduo/net/SocketsOps.h>

#include <boost/bind.hpp>

#include <errno.h>

using namespace muduo;
using namespace muduo::net;

/*  The default callback function when the connection is established and disconnected  */
void muduo::net::defaultConnectionCallback(const TcpConnectionPtr& conn)
{
  LOG_TRACE << conn->localAddress().toIpPort() << " -> "
            << conn->peerAddress().toIpPort() << " is "
            << (conn->connected() ? "UP" : "DOWN");
  // do not call conn->forceClose(), because some users want to register message callback only.
}

/*  The default callback function for receiving messages  */
void muduo::net::defaultMessageCallback(const TcpConnectionPtr&,
                                        Buffer* buf,
                                        Timestamp)
{
  buf->retrieveAll();   //  By default, all data in the buffer is taken out   
}

TcpConnection::TcpConnection(EventLoop* loop,
                             const string& nameArg,
                             int sockfd,
                             const InetAddress& localAddr,
                             const InetAddress& peerAddr)
  : loop_(CHECK_NOTNULL(loop)),
    name_(nameArg),         // connection  Name 
    state_(kConnecting),    //  The initial state is   on connection 
    reading_(true),
    socket_(new Socket(sockfd)),    //  establish sockfd  Corresponding Socket
    channel_(new Channel(loop, sockfd)), //  establish sockfd Corresponding Channel
    localAddr_(localAddr),
    peerAddr_(peerAddr),    
    highWaterMark_(64*1024*1024)    //  The default high water level is 64K
{
    /*  Callback function for registering events , Called when the corresponding event occurs  */
  channel_->setReadCallback(
      boost::bind(&TcpConnection::handleRead, this, _1));
  channel_->setWriteCallback(
      boost::bind(&TcpConnection::handleWrite, this));
  channel_->setCloseCallback(
      boost::bind(&TcpConnection::handleClose, this));
  channel_->setErrorCallback(
      boost::bind(&TcpConnection::handleError, this));
  LOG_DEBUG << "TcpConnection::ctor[" <<  name_ << "] at " << this
            << " fd=" << sockfd;
  socket_->setKeepAlive(true);
}

TcpConnection::~TcpConnection()
{
  LOG_DEBUG << "TcpConnection::dtor[" <<  name_ << "] at " << this
            << " fd=" << channel_->fd()
            << " state=" << stateToString();
  assert(state_ == kDisconnected);
}

bool TcpConnection::getTcpInfo(struct tcp_info* tcpi) const
{
  return socket_->getTcpInfo(tcpi);
}

string TcpConnection::getTcpInfoString() const
{
  char buf[1024];
  buf[0] = '\0';
  socket_->getTcpInfoString(buf, sizeof buf);
  return buf;
}

void TcpConnection::send(const void* data, int len)
{
  send(StringPiece(static_cast<const char*>(data), len));
}

void TcpConnection::send(const StringPiece& message)
{
  if (state_ == kConnected)
  {
      /*  If it's in loop Threads , Send the data directly  */
    if (loop_->isInLoopThread())
    {
      sendInLoop(message);
    }
    else
    {
        /*  otherwise , Transferred to the loop Threads execute  */
      loop_->runInLoop(
          boost::bind(&TcpConnection::sendInLoop,
                      this,     // FIXME
                      message.as_string()));
                    //std::forward<string>(message)));
    }
  }
}

// FIXME efficiency!!!
void TcpConnection::send(Buffer* buf)
{
  if (state_ == kConnected)
  {
    if (loop_->isInLoopThread())
    {
      sendInLoop(buf->peek(), buf->readableBytes());
      buf->retrieveAll();
    }
    else
    {
      loop_->runInLoop(
          boost::bind(&TcpConnection::sendInLoop,
                      this,     // FIXME
                      buf->retrieveAllAsString()));
                    //std::forward<string>(message)));
    }
  }
}

void TcpConnection::sendInLoop(const StringPiece& message)
{
  sendInLoop(message.data(), message.size());
}


/*
 *  Try sending data directly first , If one send is completed, it will not be enabled WriteCallback
 * 
 *  If only part of the data is sent , Put the remaining data into the output buffer ,
 *  And focus on writable event , In the future handleWrite() Send the remaining data in 
 * 
 *  If the current output buffer already has data to send , Then you can't try sending first 
 *  It will cause data disorder 
 */
void TcpConnection::sendInLoop(const void* data, size_t len)
{
    /*  The assertion ensures that the loop Threads  */
  loop_->assertInLoopThread();
  ssize_t nwrote = 0;
  size_t remaining = len;
  bool faultError = false;
  /*  If the connection is closed , Then give up writing data   */
  if (state_ == kDisconnected)
  {
    LOG_WARN << "disconnected, give up writing";
    return;
  } 
  // if no thing in output queue, try writing directly
  /*  If there is no data in the output buffer , You can go directly to fd Writing data , No disorder  */
  if (!channel_->isWriting() && outputBuffer_.readableBytes() == 0)
  { 
      /*  Go to fd send data  */
    nwrote = sockets::write(channel_->fd(), data, len);
    /*  Send successfully , See if there's any data left  */
    if (nwrote >= 0)
    {
        /*  Calculate the remaining quantity  */
      remaining = len - nwrote;
      /*  All sent , perform writeCompleteCallback_ Callback  */
      if (remaining == 0 && writeCompleteCallback_)
      {
        loop_->queueInLoop(boost::bind(writeCompleteCallback_, shared_from_this()));
      }
    }
    /*  send out   Failure  */
    else // nwrote < 0
    {
      nwrote = 0;
      if (errno != EWOULDBLOCK)
      {
        LOG_SYSERR << "TcpConnection::sendInLoop";
        if (errno == EPIPE || errno == ECONNRESET) // FIXME: any others?
        {
          faultError = true;
        }
      }
    }
  } // end if

  assert(remaining <= len);
  if (!faultError && remaining > 0)
  {
      /*  The remaining bytes of the output buffer  */
    size_t oldLen = outputBuffer_.readableBytes();
    /*  If the number of bytes to be sent now reaches a high level , Not before  */
    if (oldLen + remaining >= highWaterMark_
        && oldLen < highWaterMark_
        && highWaterMarkCallback_)
    {
        /*  stay  loop  The high water level callback function is executed in the thread  */
      loop_->queueInLoop(boost::bind(highWaterMarkCallback_, shared_from_this(), oldLen + remaining));
    }

    /*  Will not be sent data Put the data in the output buffer  */
    outputBuffer_.append(static_cast<const char*>(data)+nwrote, remaining);
    /*  If the corresponding Channel Not listening write event  */
    if (!channel_->isWriting())
    {
        /*  Register writable Events  */
      channel_->enableWriting();
    }
  }
}

void TcpConnection::shutdown()
{
  // FIXME: use compare and swap
  if (state_ == kConnected)     //  perform shutdown It must be under normal connection 
  {
    setState(kDisconnecting);
    // FIXME: shared_from_this()?
    loop_->runInLoop(boost::bind(&TcpConnection::shutdownInLoop, this));
  }
}

void TcpConnection::shutdownInLoop()
{
  loop_->assertInLoopThread();  //  stay IO Execute in thread 
  if (!channel_->isWriting())   //  No write events are currently monitored 
  {
    // we are not writing
    socket_->shutdownWrite();   //  Write end is closed 
  }
}
/*  Active close connection  */
void TcpConnection::forceClose()
{
  // FIXME: use compare and swap
  if (state_ == kConnected || state_ == kDisconnecting)
  {
    setState(kDisconnecting);
    loop_->queueInLoop(boost::bind(&TcpConnection::forceCloseInLoop, shared_from_this()));
  }
}

void TcpConnection::forceCloseWithDelay(double seconds)
{
  if (state_ == kConnected || state_ == kDisconnecting)
  {
    setState(kDisconnecting);
    loop_->runAfter(
        seconds,
        makeWeakCallback(shared_from_this(),
                         &TcpConnection::forceClose));  // not forceCloseInLoop to avoid race condition
  }
}

void TcpConnection::forceCloseInLoop()
{
  loop_->assertInLoopThread();
  if (state_ == kConnected || state_ == kDisconnecting)
  {
    // as if we received 0 byte in handleRead();
    handleClose(); //  Also called  handleClose()
  }
}

const char* TcpConnection::stateToString() const
{
  switch (state_)
  {
    case kDisconnected:
      return "kDisconnected";
    case kConnecting:
      return "kConnecting";
    case kConnected:
      return "kConnected";
    case kDisconnecting:
      return "kDisconnecting";
    default:
      return "unknown state";
  }
}

/*  Ban nagle Algorithm , Reduce network latency  */
void TcpConnection::setTcpNoDelay(bool on)
{
  socket_->setTcpNoDelay(on);
}

void TcpConnection::startRead()
{
  loop_->runInLoop(boost::bind(&TcpConnection::startReadInLoop, this));
}

void TcpConnection::startReadInLoop()
{
  loop_->assertInLoopThread();
  if (!reading_ || !channel_->isReading())
  {
    channel_->enableReading();
    reading_ = true;
  }
}

void TcpConnection::stopRead()
{
  loop_->runInLoop(boost::bind(&TcpConnection::stopReadInLoop, this));
}

void TcpConnection::stopReadInLoop()
{
  loop_->assertInLoopThread();
  if (reading_ || channel_->isReading())
  {
    channel_->disableReading();
    reading_ = false;
  }
}

/*  Connection is established ,TcpServer::newconnection()  call  */
void TcpConnection::connectEstablished()
{
  loop_->assertInLoopThread();
  assert(state_ == kConnecting);
  setState(kConnected);     //  Set the status to established 
  channel_->tie(shared_from_this());
  channel_->enableReading();    //  Register to read Events 

  /*  Execute the logic when the user establishes the connection  */
  connectionCallback_(shared_from_this());
}

/*  Connection is closed , Provide to TcpServer Use  TcpServer::removeConnectionInLoop() */
void TcpConnection::connectDestroyed()
{ 
  loop_->assertInLoopThread();
  if (state_ == kConnected)
  {
    setState(kDisconnected);
    channel_->disableAll();     //  Stop listening for all events 

    /*  Execute the user's shutdown logic  */
    connectionCallback_(shared_from_this());
  }
  channel_->remove();   //  from  epoll  Remove connfd
} 

/*  Handle read Events  */
void TcpConnection::handleRead(Timestamp receiveTime)
{ 
  loop_->assertInLoopThread();
  int savedErrno = 0;
  /*  from fd Read data into input buffer , Returns the number of bytes read successfully  */
  ssize_t n = inputBuffer_.readFd(channel_->fd(), &savedErrno);
  if (n > 0)
  {
      /*  Execute message callback function  */
    messageCallback_(shared_from_this(), &inputBuffer_, receiveTime);
  }
  else if (n == 0)
  {
      /*  Connection is closed , Execute the close callback function  */
    handleClose();
  }
  else
  {
    errno = savedErrno;
    LOG_SYSERR << "TcpConnection::handleRead";
    handleError();  
  }
}

/*  Handling writable events   Go to connfd in write */
void TcpConnection::handleWrite()
{
  loop_->assertInLoopThread();
  /*  If Channel Monitoring  write  event  */
  if (channel_->isWriting())
  {
    ssize_t n = sockets::write(channel_->fd(),
                               outputBuffer_.peek(),
                               outputBuffer_.readableBytes());
    if (n > 0)
    {
        /*  Delete sent data from the output buffer  */
      outputBuffer_.retrieve(n);
      /*  All data to be sent has occurred  */
      if (outputBuffer_.readableBytes() == 0)
      {
          /*  Stop listening fd Write events for */
        channel_->disableWriting();
        /*  Call the callback specified by the user  */
        if (writeCompleteCallback_)
        {
          loop_->queueInLoop(boost::bind(writeCompleteCallback_, shared_from_this()));
        }
        if (state_ == kDisconnecting)
        {
          shutdownInLoop();
        }
      }
    }
    else
    {
      LOG_SYSERR << "TcpConnection::handleWrite";
      // if (state_ == kDisconnecting)
      // {
   
   
      //   shutdownInLoop();
      // }
    }
  }
  else
  {
    LOG_TRACE << "Connection fd = " << channel_->fd()
              << " is down, no more writing";
  }
}

void TcpConnection::handleClose()
{
  loop_->assertInLoopThread();
  LOG_TRACE << "fd = " << channel_->fd() << " state = " << stateToString();
  assert(state_ == kConnected || state_ == kDisconnecting);
  // we don't close fd, leave it to dtor, so we can find leaks easily.
  setState(kDisconnected);
  channel_->disableAll();   // Channel Stop listening for all events 

  TcpConnectionPtr guardThis(shared_from_this());
  connectionCallback_(guardThis);   //  Execute the user's close connection logic 
  // must be the last line
  /* 
   * closeCallback_, Bound to the  TcpServer::removeConnection() 
   *  Execute it inside connectDestroyed()
   */
  closeCallback_(guardThis);
}

/*  Output error information in log  */
void TcpConnection::handleError()
{
  int err = sockets::getSocketError(channel_->fd());
  LOG_ERROR << "TcpConnection::handleError [" << name_
            << "] - SO_ERROR = " << err << " " << strerror_tl(err);
}

版权声明
本文为[osc_ oq0m8kxk]所创,转载请带上原文链接,感谢
https://chowdera.com/2020/12/20201207173937187e.html