如何将回调函数转换为Senders模型及实现二者协同？

一、将回调风格函数转换为Senders模型 #

回调式异步操作（比如ASIO的async_read_some）核心是「发起操作+完成后调用回调」，而Senders模型通过Sender-Receiver协议传递异步结果。要实现转换，本质是把回调逻辑封装成符合P2300标准的Sender类型。

以ASIO的async_read_some为例，具体实现步骤如下：

1. 定义自定义Sender类型 #

Sender需要持有异步操作的上下文（如socket、缓冲区），实现connect方法对接Receiver，同时提供完成签名（completion_signatures）和执行器（executor），满足Sender概念的基本要求。

2. 对接Receiver与回调逻辑 #

在Sender的operation对象的start方法中发起ASIO异步操作，在ASIO的回调里根据结果调用Receiver的set_value（成功）、set_error（失败）或set_stopped（取消）方法。

代码示例（基于stdexec） #

#include <asio.hpp>
#include <stdexec/execution.hpp>

template <typename Socket, typename Buffer>
struct read_some_sender {
    Socket& socket_;
    Buffer buffer_;
    asio::any_io_executor executor_;

    // 对接Receiver的操作对象
    template <stdexec::receiver Receiver>
    struct operation {
        Socket& socket_;
        Buffer buffer_;
        Receiver receiver_;

        void start() noexcept {
            // 发起ASIO异步读操作
            socket_.async_read_some(
                buffer_,
                [this](std::error_code ec, std::size_t bytes_read) {
                    // 根据ASIO回调结果触发Receiver的对应逻辑
                    if (ec == asio::error::operation_aborted) {
                        stdexec::set_stopped(std::move(receiver_));
                    } else if (ec) {
                        stdexec::set_error(std::move(receiver_), ec);
                    } else {
                        stdexec::set_value(std::move(receiver_), bytes_read);
                    }
                }
            );
        }
    };

    // Sender核心方法：连接Receiver
    template <stdexec::receiver Receiver>
    auto connect(Receiver&& r) noexcept {
        return operation<Socket, Buffer>{socket_, buffer_, std::forward<Receiver>(r)};
    }

    // 定义完成签名，告知Receiver可能的结果类型
    friend auto tag_invoke(stdexec::get_completion_signatures_t, const read_some_sender&) {
        return stdexec::completion_signatures<
            stdexec::set_value_t(std::size_t),
            stdexec::set_error_t(std::error_code),
            stdexec::set_stopped_t()
        >{};
    }

    // 提供执行器，确保异步操作在正确的线程上下文执行
    friend auto tag_invoke(stdexec::get_executor_t, const read_some_sender& s) noexcept {
        return s.executor_;
    }
};

// 辅助函数简化Sender创建
template <typename Socket, typename Buffer>
auto make_read_some_sender(Socket& socket, Buffer buffer) {
    return read_some_sender<Socket, Buffer>{socket, buffer, socket.get_executor()};
}

二、回调风格函数与Senders模型的协同工作 #

实际开发中无需完全替换回调代码，两者可无缝协同，主要分两种场景：

1. 在Senders流程中调用回调式函数 #

通过上面的封装，把回调式函数转成Sender后，就能用Senders模型的组合子（如then、when_all、on）串联异步流程，提升代码可读性和组合性：

asio::io_context ctx;
asio::ip::tcp::socket sock(ctx);
// 假设已完成socket连接
std::array<char, 1024> buf;

// 构建Senders流程：读数据 -> 处理结果 -> 提交到ASIO执行器
auto work = make_read_some_sender(sock, buf)
           | stdexec::then([&](std::size_t bytes) {
               std::cout << "读取到" << bytes << "字节数据\n";
           })
           | stdexec::on(ctx.get_executor());

// 启动异步流程并等待完成
stdexec::sync_wait(stdexec::start_detached(work));
ctx.run();

2. 在回调函数中触发Senders流程 #

如果已有大量回调式代码，可在回调内部启动Senders流程，只需确保Senders的执行器与ASIO的io_context绑定，保证线程安全：

asio::io_context ctx;
// 适配ASIO执行器为Senders兼容的执行器
auto ex = stdexec::asio_executor(ctx.get_executor());

std::array<char, 1024> buf;
asio::ip::tcp::socket sock(ctx);

// 传统ASIO回调
sock.async_read_some(buf, [&](std::error_code ec, std::size_t bytes) {
    if (!ec) {
        // 在回调中启动Senders流程处理结果
        auto sender = stdexec::just(bytes)
                    | stdexec::then([](std::size_t n) {
                        std::cout << "处理了" << n << "字节数据\n";
                    })
                    | stdexec::on(ex);
        stdexec::start_detached(sender);
    }
});

ctx.run();

现成适配工具 #

无需手动封装，已有成熟工具可用：

• ASIO 1.20+原生支持std::execution，可直接用asio::async_read_some返回Sender；
• libunifex提供unifex::asio_async_read_some等工具函数，直接将ASIO回调式操作转为Sender。

内容的提问来源于stack exchange，提问作者Bryan Yan