Boost awaitable：写入套接字并等待特定响应 好的，问题来了

如何解决Boost awaitable：写入套接字并等待特定响应 好的，问题来了

这个问题可能有点复杂。我会尽量解释最好的情况，以及我想出的工具来解决我的问题。

我正在编写一个套接字应用程序，它可以写入套接字并期待响应。该协议以一种简单的方式实现了这一点：每个请求都有一个“命令 ID”，它将被转发回响应中，因此我们可以拥有对特定请求做出反应的代码。

为简单起见，我们假设所有通信都是在套接字中使用 json 完成的。

首先，让我们假设这种会话类型：

using json = /* assume any json lib */;

struct socket_session {
    auto write(json data) -> boost::awaitable<void>;
    auto read() -> boost::awaitable<json>;

private:
    boost::asio::ip::tcp::socket socket;
};

通常，我会使用一个（非常）大致像这样的回调系统。

using command_it_t = std::uint32_t;

// global incrementing command id
command_it_t command_id = 0;

// All callbacks associated with commands
std::unordered_map<command_id_t,std::function<void(json)>> callbacks;

void write_command_to_socket(
    boost::io_context& ioc,socket_session& session,json command,std::function<void(json)> callback
) {
    boost::co_spawn(ioc,session->write(command),asio::detached);
    callbacks.emplace(command_id++,callback);
}

// ... somewhere in the read loop,we call this:
void call_command(json response) {
    if (auto const& command_id = response["command"]; command_id.is_integer()) {
        if (auto const it = callbacks.find(command_id_t{command_id}); it != callbacks.end()) {
            // We found the callback for this command,call it!
            auto const& [id,callback] = *it;
            callback(response["payload"]);
            callbacks.erase(it);
        }
    }
}

它会像这样使用：

write_command_to_socket(ioc,session,json_request,[](json response) {
    // do stuff
});

---

当我开始越来越多地将协程用于异步代码时，我注意到这是在那种系统中使用它们的绝佳机会。

它不会向 write 函数发送回调，而是返回一个包含响应负载的 boost::awaitable<json>，我想象它有点像这样：

auto const json_response = co_await write_command_to_socket(session,json_request);

好的，问题来了

所以第一步是像这样转换我的代码：

void write_command_to_socket(socket_session& session,json command) {
    co_await session->write(command);
    co_return /* response data from the read loop?? */
}

我注意到我没有任何等待响应的意思，因为它在另一个异步循环中。我能够想象一个看起来像我想要的系统，但我不知道如何使用协程将我自己的心智模型转换为 asio。

// Type from my mental model: an async promise
template<typename T>
struct promise {
    auto get_value() -> boost::awaitable<T>;
    auto write_value(T value);
};

// Instead of callbacks,my mental model needs promises structured in a similar way:
std::unordered_map<command_id_t,promise<json>> promises;

void write_command_to_socket(socket_session& session,json command) {
    auto const [it,inserted] = promises.emplace(session_id++,promise<json>{});
    auto const [id,promise] = *it;

    co_await session->write(command);

    // Here we awaits until the reader loop sets the value
    auto const response_json = co_await promise.get_value();
    co_return response_json;
}

// ... somewhere in the read loop

void call_command(json response) {
    if (auto const& command_id = response["command"]; command_id.is_integer()) {
        if(auto const it = promises.find(command_id_t{command_id}); it != promises.end()) {
            auto const& [id,promise] = *it;

            // Effectively calls the write_command_to_socket coroutine to continue
            promise.write_value(response["payload"]);
            promise.erase(it);
        }
    }
}

据我所知，我在这里写的作为例子的“promise类型”在boost中是不存在的。没有这种类型，我真的很难我的命令系统如何存在。我需要为那种系统编写自己的协程类型吗？有没有办法使用 boost 的协程类型来逃避？

解决方法

对于 asio，正如我所说，“承诺类型”不存在。 Asio 改为使用延续处理程序，这是一种回调，可以实际调用回调或恢复协程。

要创建这样的延续处理程序，必须首先启动异步操作。如果需要，异步操作可以由另一个恢复，或者由许多异步操作组成。这是通过 asio::async_initiate 函数完成的，该函数采用一些参数来保护延续的形式：

// the completion token type can be a callback,// could be `asio::use_awaitable_t const&` or even `asio::detached_t const&`
return asio::async_initiate<CompletionToken,void(json)>(
    [self = shared_from_this()](auto&& handler) {
        // HERE! `handler` is a callable that resumes the coroutine!
        // We can register it somewhere
        callbacks.emplace(command_id,std::forward<decltype(handler)>(handler));
    }
);

要恢复异步操作，您只需调用延续处理程序：

void call_command(json response) {
    if (auto const& command_id = response["command"]; command_id.is_integer()) {
        if (auto const it = callbacks.find(command_id_t{command_id}); it != callbacks.end()) {
            // We found the continuation handler for this command,call it!
            // It resumes the coroutine with the json as its result
            auto const& [id,callback] = *it;
            callback(response["payload"]);
            callbacks.erase(it);
        }
    }
}

这是系统的其余部分，它的外观（非常粗略）：

using command_it_t = std::uint32_t;

// global incrementing command id
command_it_t command_id = 0;

// All callbacks associated with commands
std::unordered_map<command_id_t,moveable_function<void(json)>> callbacks;

void write_command_to_socket(
    boost::io_context& ioc,socket_session session,json command
) -> boost::asio::awaitable<json> {
    return asio::async_initiate<boost::asio::use_awaitable_t<> const&,void(json)>(
        [&ioc,session](auto&& handler) {
            callbacks.emplace(command_id,std::forward<decltype(handler)>(handler));
            boost::asio::co_spawn(ioc,session.write(command),asio::detached);
        }
    );
}