TcpConnection回调

TcpConnection回调有哪些？

TcpConnection为应用程序提供了几个用于处理连接及数据的回调接口，主要包括 ConnectionCallback（连接建立回调）、MessageCallback（接收到消息回调）、WriteCompleteCallback（写完成回调）、HighWaterMarkCallback（高水位回调）、CloseCallback（连接关闭回调）。

通过以下几个成员实现存储、判断：

    ConnectionCallback connectionCallback_;       // 连接回调
    MessageCallback messageCallback_;             // 收到消息回调
    WriteCompleteCallback writeCompleteCallback_; // 写完成回调
    HighWaterMarkCallback highWaterMarkCallback_; // 高水位回调
    CloseCallback closeCallback_;                 // 关闭连接回调

    size_t highWaterMark_;                        // 高水位阈值
    Buffer inputBuffer_;                          // 应用层接收缓冲区
    Buffer outputBuffer_; // FIXME: use list<Buffer> as output buffer.　//　应用层发送缓冲区
    boost::any contex_;                           // 绑定一个用户自定义类型的上下文参数

connectionCallback_存放用户注册的连接回调， 会在以下几种情形回调：
1）Tcp连接确认建立时：TcpServer/TcpClient在Tcp连接建立时，通过newConenction新建TcpConnection对象时，回调TcpConnection::connectEstablished，进而回调connectionCallback_。
2）Tcp连接确认销毁时：TcpServer/TcpClient在移除Tcp连接时，通过removeConnectionInLoop销毁连接，回调connectionCallback_。
3）强制关闭Tcp连接时：TcpConnection强制销毁时，调用forceClose()/forceClose()，或者从sockfd read到数据量为0（对端关闭连接），进而调用handleClose，回调connectionCallback_。

messageCallback_ 存放用户注册的接收到消息回调，会在Channel调用handleRead处理读事件时，调用cpConnection::handleRead，从sockfd read到数据量>0，进而回调messageCallback_。

writeCompleteCallback_ 存放用户注册的写完成回调，会在数据发送完毕回调函数，所有的用户数据都已拷贝，outputBuffer_被清空也会回调该函数。

highWaterMarkCallback_ 存放用户注册的高水位回调，用户send数据时，会在应用发送缓存堆积数据过多（>highWaterMark_）时调用。

回调注册

注册回调接口：

    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; }

何时调用connectionCallback_？

调用connectionCallback_ 分为多种情况。connectionCallback_ 通常不应该为空，如果用户没有特殊设置，应该设为默认的defaultConnectionCallback。

1）Tcp连接建立时

/**
* TcpServer/TcpClient 调用newTcpConnection创建TcpConnection对象后, 用来做一些连接以建立的事后工作
*/
void TcpConnection::connectEstablished()
{
    loop_->assertInLoopThread();
    assert(state_ == kConnecting);
    setState(kConnected);
    channel_->tie(shared_from_this()); // tie this TcpConnection object to channel_
    channel_->enableReading(); // 使能监听读事件, 连接上有读事件发生时, 如收到对端数据, 会触发channel_::handleRead


    // FIXME: add by Martin
    assert(connectionCallback_ != nullptr); // connectionCallback_通常不允许为空, TcpServer/TcpClient可以设置为缺省值defaultConnectionCallback


    connectionCallback_(shared_from_this()); // 回调connectionCallback_
}

2）Tcp连接销毁时，通常由TcpServer/TcpClient主动移除通道事件触发

/**
* TcpServer/TcpClient移除通道事件时, 调用该函数销毁tcp连接
* @note 只有处于已连接状态(kConnected)的tcp连接, 才需要先更新状态, 关闭通道事件监听
*/
void TcpConnection::connectDestroyed()
{
    loop_->assertInLoopThread();
    if (state_ == kConnected) // 只有kConnected的连接, 才有必要采取断开连接动作
    {
        setState(kDisconnected);
        channel_->disableAll(); // 关闭通道事件监听


        connectionCallback_(shared_from_this()); // 调用连接回调
    }
    channel_->remove(); // 从EventLoop和Poller中移除监听通道事件
}

3）对端关闭连接（从sockfd read返回0），本地调用handleClose被动关闭

/**
* 处理Tcp连接关闭
* @details 更新状态为kDisconnected, 清除所有事件通道监听.
* @note 必须在所属loop线程中运行.
*/
void TcpConnection::handleClose()
{
    loop_->assertInLoopThread(); // 确保在所属loop线程中运行
    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); // 更新Tcp连接状态
    channel_->disableAll();     // 停止监听所有通道事件(读写事件)

    TcpConnectionPtr guardThis(shared_from_this());
    connectionCallback_(guardThis); // 连接回调
    // must be the last line
    closeCallback_(guardThis);      //　关闭连接回调
}

void TcpConnection::handleRead(Timestamp receiveTime)
{
    loop_->assertInLoopThread();
    int savedErrno = 0;
    ssize_t n = inputBuffer_.readFd(channel_->fd(), &savedErrno); // 从指定fd读取数据到内部缓冲
    if (n > 0)
    {
        messageCallback_(shared_from_this(), &inputBuffer_, receiveTime);
    }
    else if (n == 0)
    {
        handleClose();
    }
    ...
}

4）本地主动强制关闭（forceClose）Tcp连接，当然，connectionCallback_回调跟3）一样，还是在handleClose中发生的

/**
* 在所属loop循环中强制关闭连接, 只对连接为kConnected或kDisconnecting状态才有效
*/
void TcpConnection::forceCloseInLoop()
{
    loop_->assertInLoopThread();
    if (state_ == kConnected || state_ == kDisconnecting)
    {
        // as if we received 0 byte in handleRead()
        handleClose();
    }
}

/**
* 强制关闭连接, 只对连接为kConnected或kDisconnecting状态才有效
* @details 为防止意外, 动作应该放到loop末尾去做
*/
void TcpConnection::forceClose()
{
    // FIXME: use compare and swap
    if (state_ == kConnected || state_ == kDisconnecting)
    {
        setState(kDisconnecting);
        loop_->queueInLoop(std::bind(&TcpConnection::forceCloseInLoop, shared_from_this()));
    }
}

何时调用messageCallback_？

前面在handleRead()中已经看到，当从sockfd read返回>0时，就会回调messageCallback_，交给应用层处理从对端接收来的数据（存放到inputBuffer_中）。

void TcpConnection::handleRead(Timestamp receiveTime)
{
    loop_->assertInLoopThread();
    int savedErrno = 0;
    ssize_t n = inputBuffer_.readFd(channel_->fd(), &savedErrno); // 从指定fd读取数据到内部缓冲
    if (n > 0)
    {
        messageCallback_(shared_from_this(), &inputBuffer_, receiveTime);
    }
    ...
}

何时调用writeCompleteCallback_，highWaterMarkCallback_？

TcpConnection::send()发送数据时，会转为在IO线程中调用TcpConnection::sendInLoop。将要发送的数据写到内核缓冲区，然后回调writeCompleteCallback_；如果一次没有写完，剩余数据量较大，超过highWaterMark_（高水位），就回调highWaterMarkCallback_。

调用顺序：
send() => sendInLoop() => writeCompleteCallback_(), highWaterMarkCallback_() => handleWrite()

void TcpConnection::sendInLoop(const char *data, size_t len)
{
    string s;
    std::reverse(s.begin(), s.end());
    loop_->assertInLoopThread();
    ssize_t nwrote = 0;
    size_t remaining = len;
    bool faultError = false;
    if (state_ == kDisconnected) // 如果已经断开连接(kDisconnected), 就无需发送, 打印log(LOG_WARN)
    {
        LOG_WARN << "disconnected, give up writing";
        return;
    }

    // write一次, 往对端发送数据, 后面再看是否发生错误, 是否需要高水位回调
    // if no thing output queue, try writing directly
    if (!channel_->isWriting() && outputBuffer_.readableBytes() == 0)
    { // 如果通道没有使能监听写事件, 并且outputBuffer　没有待发送数据, 就直接通过socket写
        nwrote = sockets::write(channel_->fd(), data, len);
        if (nwrote >= 0)
        {
            remaining = len - nwrote;
            // 写完了data[len]到到内核缓冲区, 就回调writeCompleteCallback_
            if (remaining == 0 && writeCompleteCallback_)
            {
                loop_->queueInLoop(std::bind(writeCompleteCallback_, shared_from_this()));
            }
        }
        else // nwrote < 0, error
        {
            nwrote = 0;
            if (errno != EWOULDBLOCK) // EWOULDBLOCK: 发送缓冲区已满, 且fd已设为nonblocking
            { // O_NONBLOCK fd, write block but return EWOULDBLOCK error
                LOG_SYSERR << "TcpConnection::sendInLoop";
                if (errno == EPIPE || errno == ECONNRESET) // FIXME: any others?
                { // EPIPE: reading end is closed; ECONNRESET: connection reset by peer
                    faultError = true;
                }
            }
        }
    }

    // 处理剩余待发送数据
    // 没有错误, 并且还有没有写完的数据, 说明内涵发送缓冲区满, 要将未写完的数据添加到output buffer中
    assert(remaining <= len);
    if (!faultError && remaining > 0) // 没有故障, 并且还有待发送数据, 可能是发送太快, 对方来不及接收
    { // no error and data remaining to be written
        size_t oldLen = outputBuffer_.readableBytes(); // Buffer中待发送数据(readable空间数据)

        // 如果readable空间数据 + 未写完数据, 超过highWaterMark_(高水位), 回调highWaterMarkCallback_
        if (oldLen + remaining >= highWaterMark_
        && oldLen < highWaterMark_
        && highWaterMarkCallback_)
        {
            loop_->queueInLoop(std::bind(highWaterMarkCallback_, shared_from_this(), oldLen + remaining));
        }
        // append data to be written to the output buffer
        outputBuffer_.append(static_cast<const char*>(data) + nwrote, remaining);
        // enable write event for channel_
        if (!channel_->isWriting())
        { // 如果没有在监听通道写事件, 就使能通道写事件
            channel_->enableWriting();
        }
    }
}

/**
* 内核发送缓冲区有空间了, 回调该函数.
* @details 在send()之后触发. send()通常由用户主动调用, handleWrite由epoll_wait/poll监听,
* 由EventLoop对应IO线程回调.
*/
void TcpConnection::handleWrite()
{
    loop_->assertInLoopThread();
    if (channel_->isWriting())
    { // 只有通道在监听写事件时, 处理write事件才有意义
        // 继续往内核缓冲区写readable空间的待发送数据
        ssize_t n = sockets::write(channel_->fd(),
                                   outputBuffer_.peek(),
                                   outputBuffer_.readableBytes());
        if (n > 0)
        {
            outputBuffer_.retrieve(n);
            if (outputBuffer_.readableBytes() == 0)
            { // 发送缓冲区已清空
                channel_->disableWriting();
                if (writeCompleteCallback_)
                { // 应用层发送缓冲区已被清空, 就回调writeCompleteCallback_
                    loop_->queueInLoop(std::bind(writeCompleteCallback_, shared_from_this()));
                }
                if (state_ == kDisconnecting)
                { // 发送缓冲区已清空, 并且连接状态是kDisconnecting, 要关闭连接
                    shutDownInLoop(); // 关闭连接 写方向
                }
            }
        }
        else
        {
            LOG_SYSERR << "TcpConnection::handleWrite";
        }
    }
}

上下文数据传递

input buffer、output buffer是Tcp连接中，伴随着一次或几次请求、应答的数据，是与对端进行沟通的数据。如果想要一下数据，伴随着整个Tcp连接的声明周期，该怎么办？
可以使用TcpConnection::contex_，在连接建立时（回调onConnection），创建boost::any对象（可变类型），然后让TcpConnection::contex_指向该对象。这样，在整个TcpConnection生命周期内，都可以直接通过contex_访问该对象；直到沟通完成，或者连接关闭时，reset contex_。

public:
    ...
    void setContext(const boost::any& contex)
    { contex_ = contex; }

    const boost::any& getContext() const
    { return contex_; }

    boost::any* getMutableContext()
    { return &contex_; }
    ...
private:
    boost::any contex_;                           // 用户自定义参数
    ...

比如，对于一个HttpSever，可以中连接回调onConnection中存放一个HttpContext对象；在收到对端消息，回调onMessage时，可以获得该HttpContext对象；在确认解析完HTTP请求后，可以reset 重置context上下文。

void HttpServer::onConnection(const TcpConnectionPtr &conn)
{
    if (conn->connected()) // 确认已处于连接建立状态
    {
        conn->setContext(HttpContext()); // 让contex_ 指向一个HttpContext对象
    }
}

void HttpServer::onMessage(const TcpConnectionPtr &conn, Buffer *buf, Timestamp receivedTime)
{
    // 从TcpConnection获取contex_ 上下文，然后将内容转型为所需的HttpContext类型
    HttpContext* context = boost::any_cast<HttpContext>(conn->getMutableContext()); 

    if (!context->parseRequest(buf, receivedTime))
    { // parse request failure
        conn->send("HTTP/1.1 400 Bad Request\r\n\r\n");
        conn->shutdown();
    }

    if (context->gotAll())
    {
        onRequest(conn, context->request());
        context->reset();
    }
}

知识点

boost::any 类型

可变类型boost::any可以对任意类型安全存储、安全取回。boost::any相当于C语言里面的void*，但在C++中不是类型安全的。中STL中存放不同类型的方法，如vectorboost::any。

当需要用boost::any进行存储时，可以利用boost::any指向要存储的对象；

当需要取出内容时，可以用boost::any_cast将取出的内容转型为存放时代对象类型。

#include <boost/any.hpp>
#include <iostream>
#include <string>
#include <vector>
using namespace std;
int main()
{
    vector<boost::any> someValues;
    someValues.emplace_back(10);                       // 存放一个int数据
    someValues.emplace_back(string("test"));           // 存放一个string对象

    int& a = boost::any_cast<int&>(someValues.front());      // 取出时转型为int&
    cout << a << endl; // 打印10
    string& s = boost::any_cast<string&>(someValues.back()); // 取出时转型为string&
    cout << s << endl; // 打印test
    return 0;
}