318-重写C++muduo库(TcpConnection)

• 打包成功连接服务器的客户端得到一条通信链路的；
• 描述建立连接的客户端和服务器之间的联系的！

mainloop拿到连接成功的通信connfd，通过轮询算法拿到一个subloop，最终将connfd打包成TcpConnection，注册到subloop；（TcpServer::newConnection函数就是做这个事的）

TcpConnection连接只是保存了客户端通信的所有的数据；（需要一个Socket）

但是最终subloop中的事件分发器Poller主要监听的还是一个Channel，所以TcpConnection的部分数据 + Socket打包成一个Channel注册到Poller中，Poller监听到有事件发生后，就会执行Channel中的回调操作：

这些回调都是Channel从TcpConnection中拿到的！

可以看到在TcpConnection中设置了相应的回调：

回调函数由用户给了TcpServer，TcpSever给到TcpConnection，TcpConnection最终将这些回调操作设置给Channel，Channel就可以加入Poller，Poller监听到事件之后，Channel就会调用相应的回调。

HighWaterMarkCallback：

用于控制两端的收发速率，设置了“水位线”

std::enable_shared_from_this：得到当前对象的一个智能指针！（https://blog.csdn.net/QIANGWEIYUAN/article/details/88973735）

TcpConnection都是在subloop中管理的，所以TcpConnection中的成员变量loop肯定是subloop，是mainloop根据轮询算出来的！

• TcpSever通过Acceptor，有一个新用户的连接，通过accept函数拿到新用户的连接connfd，然后打包成TcpConnection，TcpSever给设置相应的回调；
• 然后TcpConnection将相应的回调设置到Channel中，Channel注册到Poller上，Poller监听到事件以后，就会调用Poller的回调操作

Channel绑定的回调操作，执行的就是在TcpConnection中预置的回调操作！

Acceptor只绑定了读事件的回调操作：

• 只绑定了setReadCallback回调函数，主要作用是监听新用户连接，拿到connfd，返回Socket，打包成TcpConnection，注册相应的回调，这些都是在TcpServer中实现的
  

用户会给TcpSever注册一个MessageCallback回调函数，就是已建立连接有读写事件的时候，MessageCallback会响应，MessageCallback处理完业务代码后，会通过这个send方法给客户端用户返回信息！

send函数就是直接向outputBuffer_中发送数据，因为直接发送的数据太快，需要一个缓冲区，并设置一个水位线，防止发送数据过快（容易丢数据）！

我们对Socket的读事件感兴趣：

tie_用一个弱智能指针记录着！

用一个弱智能指针绑定TcpConnection资源，监视TcpConnection资源作用！防止TcpConnection资源被释放；防止TcpConnection被remove掉之后，Channel底层调用的回调函数就没了！（强弱智能指针搭配使用，既可以监控对象的状态，还可以解决多线程访问共享对象的安全问题！）

#pragma once

#include "noncopyable.h"
#include "InetAddress.h"
#include "Callbacks.h"
#include "Buffer.h"
#include "Timestamp.h"

#include 
#include 
#include 

class Channel;
class EventLoop;
class Socket;

 
class TcpConnection : noncopyable, public std::enable_shared_from_this
{
public:
    TcpConnection(EventLoop *loop, 
                const std::string &name, 
                int sockfd,
                const InetAddress& localAddr,
                const InetAddress& peerAddr);
    ~TcpConnection();

    EventLoop* getLoop() const { return loop_; }
    const std::string& name() const { return name_; }
    const InetAddress& localAddress() const { return localAddr_; }
    const InetAddress& peerAddress() const { return peerAddr_; }

    bool connected() const { return state_ == kConnected; }//查看是否成功了

    //发送数据
    void send(const std::string &buf);
    //关闭连接
    void shutdown();

    void setConnectionCallback(const ConnectionCallback& cb)
    { connectionCallback_ = cb; }

    void setMessageCallback(const MessageCallback& cb)
    { messageCallback_ = cb; }

    void setWriteCompleteCallback(const WriteCompleteCallback& cb)
    { writeCompleteCallback_ = cb; }

    void setHighWaterMarkCallback(const HighWaterMarkCallback& cb, size_t highWaterMark)
    { highWaterMarkCallback_ = cb; highWaterMark_ = highWaterMark; }

    void setCloseCallback(const CloseCallback& cb)
    { closeCallback_ = cb; }

    //连接建立
    void connectEstablished();
    //连接销毁
    void connectDestroyed();
private:
    enum StateE {kDisconnected, kConnecting, kConnected, kDisconnecting};
    void setState(StateE state) { state_ = state; }

    void handleRead(Timestamp receiveTime);
    void handleWrite();
    void handleClose();
    void handleError();

    void sendInLoop(const void* message, size_t len);
    void shutdownInLoop();

    EventLoop *loop_;//这里绝对不是baseLoop， 因为TcpConnection都是在subLoop里面管理的
    const std::string name_;
    std::atomic_int state_;
    bool reading_;

    //这里和Acceptor类似   Acceptor=》mainLoop    TcpConenction=》subLoop
    std::unique_ptr socket_;
    std::unique_ptr channel_;

    const InetAddress localAddr_;//当前主机IP地址端口号 
    const InetAddress peerAddr_;//对端IP地址端口号 

    ConnectionCallback connectionCallback_;//有新连接时的回调
    MessageCallback messageCallback_;//有读写消息时的回调
    WriteCompleteCallback writeCompleteCallback_;//消息发送完成以后的回调
    HighWaterMarkCallback highWaterMarkCallback_;//水位 
    CloseCallback closeCallback_;
    size_t highWaterMark_;//水位标志

    Buffer inputBuffer_;//接收数据的缓冲区
    Buffer outputBuffer_;//发送数据的缓冲区
};

#include "TcpConnection.h"
#include "Logger.h"
#include "Socket.h"
#include "Channel.h"
#include "EventLoop.h"

#include 
#include 
#include          
#include 
#include 
#include 
#include 
#include 

static EventLoop* CheckLoopNotNull(EventLoop *loop)
{
    if (loop == nullptr)
    {
        LOG_FATAL("%s:%s:%d TcpConnection Loop is null! n", __FILE__, __FUNCTION__, __LINE__);
    }
    return loop;
}

TcpConnection::TcpConnection(EventLoop *loop, //构造函数 
                const std::string &nameArg, 
                int sockfd,
                const InetAddress& localAddr,
                const InetAddress& peerAddr)
    : loop_(CheckLoopNotNull(loop))
    , name_(nameArg)
    , state_(kConnecting)
    , reading_(true)
    , socket_(new Socket(sockfd))
    , channel_(new Channel(loop, sockfd))
    , localAddr_(localAddr)
    , peerAddr_(peerAddr)
    , highWaterMark_(64*1024*1024) //超过64M就到水位线了，要停止发送 
{
    //下面给channel设置相应的回调函数，poller给channel通知感兴趣的事件发生了，channel会回调相应的操作函数
    channel_->setReadCallback(
        std::bind(&TcpConnection::handleRead, this, std::placeholders::_1)
    );
    channel_->setWriteCallback(
        std::bind(&TcpConnection::handleWrite, this)
    );
    channel_->setCloseCallback(
        std::bind(&TcpConnection::handleClose, this)
    );
    channel_->setErrorCallback(
        std::bind(&TcpConnection::handleError, this)
    );

    LOG_INFO("TcpConnection::ctor[%s] at fd=%dn", name_.c_str(), sockfd);
    socket_->setKeepAlive(true);//启动Tcp Socket的保活机制
}


TcpConnection::~TcpConnection()//析构函数 
{
    LOG_INFO("TcpConnection::dtor[%s] at fd=%d state=%d n", 
        name_.c_str(), channel_->fd(), (int)state_);
}

void TcpConnection::send(const std::string &buf)
{
    if (state_ == kConnected)
    {
        if (loop_->isInLoopThread())//当前loop是不是在对应的线程 
        {
            sendInLoop(buf.c_str(), buf.size());
        }
        else
        {
            loop_->runInLoop(std::bind(
                &TcpConnection::sendInLoop,
                this,
                buf.c_str(),
                buf.size()
            ));
        }
    }
}

 
void TcpConnection::sendInLoop(const void* data, size_t len)
{
    ssize_t nwrote = 0;
    size_t remaining = len;
    bool faultError = false;

    //之前调用过该connection的shutdown，不能再进行发送了
    if (state_ == kDisconnected)
    {
        LOG_ERROR("disconnected, give up writing!");
        return;
    }

    //表示channel_第一次开始写数据，而且缓冲区没有待发送数据 (Channel向缓冲区写数据，给客户端应用响应)
    if (!channel_->isWriting() && outputBuffer_.readableBytes() == 0)
    {
        nwrote = ::write(channel_->fd(), data, len);
        if (nwrote >= 0)
        {
            remaining = len - nwrote;
            if (remaining == 0 && writeCompleteCallback_)
            {
                //既然在这里数据全部发送完成，就不用再给channel设置epollout事件了
                loop_->queueInLoop(
                    std::bind(writeCompleteCallback_, shared_from_this())
                );
            }
        }
        else//nwrote < 0
        {
            nwrote = 0;
            if (errno != EWOULDBLOCK)
            {
                LOG_ERROR("TcpConnection::sendInLoop");
                if (errno == EPIPE || errno == ECONNRESET) // SIGPIPE  RESET, 表示对端的socket的重置
                {
                    faultError = true;
                }
            }
        }
    }

    //说明当前这一次write，并没有把数据全部发送出去，剩余的数据需要保存到缓冲区当中，然后给channel
    //注册epollout事件，poller发现tcp的发送缓冲区有空间，会通知相应的sock-channel，调用writeCallback_回调方法
    //也就是调用TcpConnection::handleWrite方法，把发送缓冲区中的数据全部发送完成
    if (!faultError && remaining > 0) 
    {
        //目前发送缓冲区剩余的待发送数据的长度
        size_t oldLen = outputBuffer_.readableBytes();
        if (oldLen + remaining >= highWaterMark_
            && oldLen < highWaterMark_
            && highWaterMarkCallback_)
        {
            loop_->queueInLoop(
                std::bind(highWaterMarkCallback_, shared_from_this(), oldLen+remaining)
            );
        }
        outputBuffer_.append((char*)data + nwrote, remaining);
        if (!channel_->isWriting())
        {
            channel_->enableWriting();//这里一定要注册channel的写事件，否则poller不会给channel通知epollout
        }
    }
}

//关闭连接
void TcpConnection::shutdown()
{
    if (state_ == kConnected)
    {
        setState(kDisconnecting);
        loop_->runInLoop(
            std::bind(&TcpConnection::shutdownInLoop, this)
        );
    }
}

void TcpConnection::shutdownInLoop()
{
    if (!channel_->isWriting())//说明outputBuffer中的数据已经全部发送完成
    {
        socket_->shutdownWrite();//关闭写端
    }
}

//连接建立
void TcpConnection::connectEstablished()
{
    setState(kConnected);
    channel_->tie(shared_from_this());
    channel_->enableReading();//向poller注册channel的epollin事件

    //新连接建立，执行回调
    connectionCallback_(shared_from_this());
}

//连接销毁
void TcpConnection::connectDestroyed()
{
    if (state_ == kConnected)
    {
        setState(kDisconnected);
        channel_->disableAll(); // 把channel的所有感兴趣的事件，从poller中del掉
        connectionCallback_(shared_from_this());
    }
    channel_->remove();//把channel从poller中删除掉
}

//表示fd有数据可读
void TcpConnection::handleRead(Timestamp receiveTime)
{
    int savedErrno = 0;
    ssize_t n = inputBuffer_.readFd(channel_->fd(), &savedErrno);
    if (n > 0)
    {
        //已建立连接的用户，有可读事件发生了，调用用户传入的回调操作onMessage
        messageCallback_(shared_from_this(), &inputBuffer_, receiveTime);//shared_from_this:获取当前TcpConnection对象的一个智能指针
    }
    else if (n == 0)//客户端断开连接
    {
        handleClose();
    }
    else
    {
        errno = savedErrno;
        LOG_ERROR("TcpConnection::handleRead");
        handleError();
    }
}

void TcpConnection::handleWrite()
{
    if (channel_->isWriting())
    {
        int savedErrno = 0;
        ssize_t n = outputBuffer_.writeFd(channel_->fd(), &savedErrno);
        if (n > 0)
        {
            outputBuffer_.retrieve(n);
            if (outputBuffer_.readableBytes() == 0)
            {
                channel_->disableWriting();
                if (writeCompleteCallback_)
                {
                    //唤醒loop_对应的thread线程，执行回调
                    loop_->queueInLoop(
                        std::bind(writeCompleteCallback_, shared_from_this())
                    );
                }
                if (state_ == kDisconnecting)
                {
                    shutdownInLoop();
                }
            }
        }
        else
        {
            LOG_ERROR("TcpConnection::handleWrite");
        }
    }
    else
    {
        LOG_ERROR("TcpConnection fd=%d is down, no more writing n", channel_->fd());
    }
}

//poller => channel::closeCallback => TcpConnection::handleClose
void TcpConnection::handleClose()
{
    LOG_INFO("TcpConnection::handleClose fd=%d state=%d n", channel_->fd(), (int)state_);
    setState(kDisconnected);
    channel_->disableAll();

    TcpConnectionPtr connPtr(shared_from_this());
    connectionCallback_(connPtr);执行连接关闭的的回调（connectionCallback_回调函数 连接成功和关闭都会响应）
    closeCallback_(connPtr);//关闭连接的回调  执行的是TcpServer::removeConnection回调方法
}

void TcpConnection::handleError()
{
    int optval;
    socklen_t optlen = sizeof optval;
    int err = 0;
    if (::getsockopt(channel_->fd(), SOL_SOCKET, SO_ERROR, &optval, &optlen) < 0)
    {
        err = errno;
    }
    else
    {
        err = optval;
    }
    LOG_ERROR("TcpConnection::handleError name:%s - SO_ERROR:%d n", name_.c_str(), err);
}

TcpConnection主要关注：

• 连接建立： connectEstablished函数
• 连接断开： connectDestroyed函数
• handleRead、handleWrite、handleClose、handleError 4个回调函数！
  在TcpConnection构造函数中，会将这4个回调函数（本身也是从TcpServer传下来的！）传给Channel:
  
  下面是TcpServer将用户传的回调函数给TcpConnection的handleRead、handleWrite、handleClose、handleError 4个函数
  
  用户使用时，向TcpServer传入回调函数：
  
• send函数： 发送数据
• shutdown函数： 关闭连接

OK的！