Boost beast::websocket 回调函数

Question

我正在试验 Boost beast::websocket websocket_client_async.cpp 示例，以及 websocket_server_async.cpp。

如给定的那样，client 示例只是建立一个连接，向服务器发送一个字符串（它只是回显），打印回复，关闭并存在。

我正在尝试修改客户端以使会话保持活动状态，以便我可以重复发送/接收字符串。因此，虽然示例代码的 on_handshake 函数会立即通过 ws_.async_write(...) 发送字符串，但我将其分离到自己的 write(...) 函数中。

这是我修改后的session 类：

using tcp = boost::asio::ip::tcp;
namespace websocket = boost::beast::websocket;

void fail(boost::system::error_code ec, char const* what)
{
    std::cerr << what << ": " << ec.message() << "\n";
}

// Sends a WebSocket message and prints the response
class session : public std::enable_shared_from_this<session>
{
    tcp::resolver resolver_;
    websocket::stream<tcp::socket> ws_;
    std::atomic<bool> io_in_progress_;
    boost::beast::multi_buffer buffer_;
    std::string host_;

public:
    // Resolver and socket require an io_context
    explicit session(boost::asio::io_context& ioc) : resolver_(ioc), ws_(ioc) {
        io_in_progress_ = false;
    }

    bool io_in_progress() const {
        return io_in_progress_;
    }

    // +---------------------+
    // | The "open" sequence |
    // +---------------------+
    void open(char const* host, char const* port)
    {
        host_ = host;

        // Look up the domain name
        resolver_.async_resolve(host, port,
            std::bind( &session::on_resolve, shared_from_this(),
                std::placeholders::_1, std::placeholders::_2 )
        );
    }

    void on_resolve(boost::system::error_code ec, tcp::resolver::results_type results)
    {
        if (ec)
            return fail(ec, "resolve");

        boost::asio::async_connect(
            ws_.next_layer(), results.begin(), results.end(),
            std::bind( &session::on_connect, shared_from_this(),
                std::placeholders::_1 )
        );
    }

    void on_connect(boost::system::error_code ec)
    {
        if (ec)
            return fail(ec, "connect");

        ws_.async_handshake(host_, "/",
            std::bind( &session::on_handshake, shared_from_this(),
                std::placeholders::_1 )
        );
    }

    void on_handshake(boost::system::error_code ec)
    {
        if (ec)
            return fail(ec, "handshake");
        else {
            std::cout << "Successful handshake with server.\n";
        }
    }

    // +---------------------------+
    // | The "write/read" sequence |
    // +---------------------------+
    void write(const std::string &text)
    {
        io_in_progress_ = true;
        ws_.async_write(boost::asio::buffer(text),
            std::bind( &session::on_write, shared_from_this(),
                std::placeholders::_1, std::placeholders::_2 )
        );
    }

    void on_write(boost::system::error_code ec, std::size_t bytes_transferred)
    {
        boost::ignore_unused(bytes_transferred);
        if (ec)
            return fail(ec, "write");

        ws_.async_read(buffer_,
            std::bind( &session::on_read, shared_from_this(),
                std::placeholders::_1, std::placeholders::_2 )
        );
    }

    void on_read(boost::system::error_code ec, std::size_t bytes_transferred)
    {
        io_in_progress_ = false; // end of write/read sequence
        boost::ignore_unused(bytes_transferred);
        if (ec)
            return fail(ec, "read");

        std::cout << boost::beast::buffers(buffer_.data()) << std::endl;
    }

    // +----------------------+
    // | The "close" sequence |
    // +----------------------+
    void close()
    {
        io_in_progress_ = true;
        ws_.async_close(websocket::close_code::normal,
            std::bind( &session::on_close, shared_from_this(),
                std::placeholders::_1)
        );
    }

    void on_close(boost::system::error_code ec)
    {
        io_in_progress_ = false; // end of close sequence
        if (ec)
            return fail(ec, "close");

        std::cout << "Socket closed successfully.\n";
    }
};

问题是，虽然连接工作正常并且我可以发送一个字符串，但 on_read 回调永远不会被击中（除非我做了下面描述的丑陋黑客攻击）。

我的main 看起来像这样：

void wait_for_io(std::shared_ptr<session> psession, boost::asio::io_context &ioc)
{
    // Continually try to run the ioc until the callbacks are finally
    // triggered (as indicated by the session::io_in_progress_ flag)
    while (psession->io_in_progress()) {
        std::this_thread::sleep_for(std::chrono::milliseconds(1));
        ioc.run();
    }
}

int main(int argc, char** argv)
{
    // Check command line arguments.
    if (argc != 3) {
        std::cerr << "usage info goes here...\n";
        return EXIT_FAILURE;
    }
    const char *host = argv[1], *port = argv[2];

    boost::asio::io_context ioc;
    std::shared_ptr<session> p = std::make_shared<session>(ioc);
    p->open(host, port);
    ioc.run(); // This works. Connection is established and all callbacks are executed.

    p->write("Hello world"); // String is sent & received by server,
                             // even before calling ioc.run()
                             // However, session::on_read callback is never
                             // reached.

    ioc.run();               // This seems to be ignored and returns immediately, so
    wait_for_io(p, ioc);     // <-- so this hack is necessary

    p->close();              // session::on_close is never reached
    ioc.run();               // Again, this seems to be ignored and returns immediately, so
    wait_for_io(p, ioc);     // <-- this is necessary

    return EXIT_SUCCESS;
}

如果我这样做：

p->write("Hello world");
while(1) {
    std::this_thread::sleep_for(std::chrono::milliseconds(1));
}

我可以确认该字符串已由服务器发送和接收¹，并且session::on_read 回调未到达。

p->close() 也会发生同样的事情。

但是，如果我添加我奇怪的 wait_for_io() 函数，一切正常。我很肯定这是一个可怕的黑客攻击，但我无法弄清楚发生了什么。

¹ 注意：我可以确认消息确实到达了服务器，因为我修改了服务器示例以将任何接收到的字符串打印到控制台。这是我做的唯一修改。回显到客户端的功能没有改变。

Answer 1

io_context::run 只会在没有更多待处理的工作时返回。如果您只是确保始终有对 websocket::stream::async_read 的挂起调用处于活动状态，那么 run 将永远不会返回，并且不需要黑客攻击。此外，您将收到服务器发送的所有消息。

Answer 2

第一次调用后对io_context::run() 的调用不起作用的原因（此处显示）：

boost::asio::io_context ioc;
std::shared_ptr<session> p = std::make_shared<session>(ioc);
p->open(host, port);
ioc.run(); // This works. Connection is established and all callbacks are executed.

是因为函数 io_context::restart() 必须在任何后续调用 io_context::run 之前调用。

From the documentation:

io_context::restart

重新启动 io_context 以准备后续的 run() 调用。

此函数必须在 run()、run_one()、 poll() 或 poll_one() 函数，当这些函数的先前调用由于 io_context 停止或耗尽工作而返回时。调用restart()后，io_context对象的stopped()函数会返回false。