Using a Capy Stream from Asio

Drive a Capy any_stream with Asio completion tokens, here plain callbacks.

example/CMakeLists.txt adds this example only when a Boost::asio target is available. b2 builds it through example/Jamfile, and CI does not build it. The full source is in example/asio/: asio_callbacks.cpp, and the adapter in api/uni_stream.hpp. Both use api/capy_streams.hpp, which declares asio_context and make_stream_pair.

What This Example Shows

An adapter, uni_stream, that gives a any_stream the Asio universal asynchronous model with net::async_initiate
Each operation starting a Capy task with run_async, on an executor that posts through Asio
net::async_write and net::async_read composed over the adapter
Two callback chains that share one Asio io_context

Source Code

//
// Copyright (c) 2026 Vinnie Falco (vinnie.falco@gmail.com)
//
// Distributed under the Boost Software License, Version 1.0. (See accompanying
// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
//
// Official repository: https://github.com/cppalliance/capy
//

#include "api/uni_stream.hpp"

#include <boost/asio/read.hpp>
#include <boost/asio/write.hpp>

#include <cstdio>
#include <cstring>
#include <memory>

constexpr std::size_t total_bytes = 1024;

struct do_write
{
    struct state
    {
        uni_stream& stream;
        std::size_t total;
        std::size_t written = 0;
        char buf[128];

        state(uni_stream& s, std::size_t t)
            : stream(s)
            , total(t)
        {
            std::memset(buf, 'X', sizeof(buf));
        }
    };

    std::shared_ptr<state> state_;

    do_write(uni_stream& stream, std::size_t total)
        : state_(std::make_shared<state>(stream, total))
    {
    }

    void operator()()
    {
        if(state_->written >= state_->total)
        {
            std::printf("writer: done, wrote %zu bytes\n", state_->written);
            return;
        }

        std::size_t chunk = (std::min)(sizeof(state_->buf), state_->total - state_->written);

        net::async_write(
            state_->stream,
            net::buffer(state_->buf, chunk),
            [self = *this](boost::system::error_code ec, std::size_t n) mutable
            {
                if(ec)
                {
                    std::printf("writer error: %s\n", ec.message().c_str());
                    return;
                }
                self.state_->written += n;
                std::printf("writer: wrote %zu bytes (total %zu)\n", n, self.state_->written);
                self();
            });
    }
};

struct do_read
{
    struct state
    {
        uni_stream& stream;
        std::size_t total;
        std::size_t read_total = 0;
        char buf[128];

        state(uni_stream& s, std::size_t t)
            : stream(s)
            , total(t)
        {
        }
    };

    std::shared_ptr<state> state_;

    do_read(uni_stream& stream, std::size_t total)
        : state_(std::make_shared<state>(stream, total))
    {
    }

    void operator()()
    {
        if(state_->read_total >= state_->total)
        {
            std::printf("reader: done, read %zu bytes\n", state_->read_total);
            std::printf("example complete!\n");
            return;
        }

        std::size_t remaining = state_->total - state_->read_total;
        std::size_t to_read = (std::min)(sizeof(state_->buf), remaining);

        net::async_read(
            state_->stream,
            net::buffer(state_->buf, to_read),
            [self = *this](boost::system::error_code ec, std::size_t n) mutable
            {
                if(ec)
                {
                    std::printf("reader error: %s\n", ec.message().c_str());
                    return;
                }
                self.state_->read_total += n;
                std::printf("reader: read %zu bytes (total %zu)\n", n, self.state_->read_total);
                self();
            });
    }
};

int main()
{
    // Type-erased asio::io_context
    asio_context ctx;

    // Return two connected uni_stream
    auto [client, server] = make_uni_pair(ctx);

    // Start the writer (callback chain)
    do_write(client, total_bytes)();

    // Start the reader (callback chain)
    do_read(server, total_bytes)();

    // Run the I/O context, this returns when all work is done
    ctx.run();

    return 0;
}

api/uni_stream.hpp

//
// Copyright (c) 2026 Vinnie Falco (vinnie.falco@gmail.com)
//
// Distributed under the Boost Software License, Version 1.0. (See accompanying
// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
//
// Official repository: https://github.com/cppalliance/capy
//

#ifndef CAPY_EXAMPLE_UNI_STREAM_HPP
#define CAPY_EXAMPLE_UNI_STREAM_HPP

#include <boost/asio/associated_executor.hpp>
#include <boost/asio/async_result.hpp>
#include <boost/asio/dispatch.hpp>
#include <boost/asio/io_context.hpp>
#include <boost/asio/post.hpp>

#include "abandon_guard.hpp"
#include "capy_streams.hpp"

#include <boost/capy/buffers/asio.hpp>
#include <boost/capy/io/any_stream.hpp>
#include <boost/capy/ex/frame_allocator.hpp>
#include <boost/capy/ex/run_async.hpp>
#include <boost/capy/task.hpp>

#include <utility>

namespace detail {

template<class AsioExecutor>
class asio_executor_wrapper
{
    AsioExecutor ex_;
    asio_context* ctx_;

public:
    asio_executor_wrapper(
        AsioExecutor ex,
        asio_context* ctx) noexcept
        : ex_(std::move(ex))
        , ctx_(ctx)
    {
    }

    bool operator==(asio_executor_wrapper const& other) const noexcept
    {
        return ex_ == other.ex_;
    }

    asio_context& context() const noexcept
    {
        return *ctx_;
    }

    void on_work_started() const noexcept {}
    void on_work_finished() const noexcept {}

    std::coroutine_handle<> dispatch(capy::continuation& c) const
    {
        post(c);
        return std::noop_coroutine();
    }

    void post(capy::continuation& c) const
    {
        resume_or_abandon r(c);
        try
        {
            net::post(ex_, std::move(r));
        }
        catch(...)
        {
            // A handler Asio took and then destroyed has given `c` up,
            // so the post must not also report it as still held.
            if(r.release())
                throw;
        }
    }
};

} // namespace detail

/** Asio-compatible stream wrapper for capy::any_stream.

    This class wraps a capy::any_stream and provides async_read_some
    and async_write_some member functions that conform to Asio's
    Universal Async Model, accepting completion tokens.

    @par Example
    @code
    asio_context ctx;
    auto [s1, s2] = make_stream_pair(ctx);
    uni_stream stream(std::move(s1), ctx);

    // Use with completion handler
    stream.async_read_some(net::buffer(data, size),
        [](boost::system::error_code ec, std::size_t n) {
            // handle completion
        });

    // Use with use_awaitable
    auto [ec, n] = co_await stream.async_read_some(
        net::buffer(data, size), net::as_tuple(net::use_awaitable));
    @endcode
*/
class uni_stream
{
    capy::any_stream stream_;
    asio_context* ctx_;

    class initiate_async_read_some
    {
        uni_stream* self_;

    public:
        explicit initiate_async_read_some(uni_stream* self) noexcept
            : self_(self)
        {
        }

        template<class ReadHandler, class MutableBufferSequence>
        void operator()(
            ReadHandler&& handler,
            MutableBufferSequence const& buffers) const
        {
            auto ex = net::get_associated_executor(
                handler, self_->ctx_->context().get_executor());

            detail::asio_executor_wrapper wrapper(std::move(ex), self_->ctx_);

            capy::run_async(wrapper)(
                [](capy::any_stream* stream,
                   MutableBufferSequence bufs,
                   ReadHandler h) -> capy::task<>
                {
                    auto [ec, n] = co_await stream->read_some(
                        capy::from_asio(bufs));
                    std::move(h)(ec, n);
                }(&self_->stream_, buffers, std::forward<ReadHandler>(handler)));
        }
    };

    class initiate_async_write_some
    {
        uni_stream* self_;

    public:
        explicit initiate_async_write_some(uni_stream* self) noexcept
            : self_(self)
        {
        }

        template<class WriteHandler, class ConstBufferSequence>
        void operator()(
            WriteHandler&& handler,
            ConstBufferSequence const& buffers) const
        {
            auto ex = net::get_associated_executor(
                handler, self_->ctx_->context().get_executor());

            detail::asio_executor_wrapper wrapper(std::move(ex), self_->ctx_);

            capy::run_async(wrapper)(
                [](capy::any_stream* stream,
                   ConstBufferSequence bufs,
                   WriteHandler h) -> capy::task<>
                {
                    auto [ec, n] = co_await stream->write_some(
                        capy::from_asio(bufs));
                    std::move(h)(ec, n);
                }(&self_->stream_, buffers, std::forward<WriteHandler>(handler)));
        }
    };

public:
    /// The executor type associated with the stream.
    using executor_type = net::io_context::executor_type;

    /** Constructor.

        @param stream The any_stream to wrap (moved).
        @param ctx The asio_context for executor operations.
    */
    uni_stream(
        capy::any_stream stream,
        asio_context& ctx) noexcept
        : stream_(std::move(stream))
        , ctx_(&ctx)
    {
    }

    uni_stream(uni_stream const&) = delete;
    uni_stream& operator=(uni_stream const&) = delete;

    uni_stream(uni_stream&&) = default;
    uni_stream& operator=(uni_stream&&) = default;

    /** Returns the executor associated with the stream.
    */
    executor_type get_executor() const noexcept
    {
        return ctx_->context().get_executor();
    }

    /** Start an asynchronous read operation.

        This function reads data into the provided buffer sequence.
        The operation completes when at least one byte has been read,
        or an error occurs.

        @param buffers The buffer sequence to read into.
        @param token The completion token. The completion handler
            signature is `void(boost::system::error_code, std::size_t)`.

        @return The return value depends on the completion token type.
    */
    template<class MutableBufferSequence,
        BOOST_ASIO_COMPLETION_TOKEN_FOR(
            void(boost::system::error_code, std::size_t)) ReadToken>
    auto
    async_read_some(
        MutableBufferSequence const& buffers,
        ReadToken&& token)
    {
        return net::async_initiate<ReadToken,
            void(boost::system::error_code, std::size_t)>(
                initiate_async_read_some(this),
                token,
                buffers);
    }

    /** Start an asynchronous write operation.

        This function writes data from the provided buffer sequence.
        The operation completes when at least one byte has been written,
        or an error occurs.

        @param buffers The buffer sequence to write from.
        @param token The completion token. The completion handler
            signature is `void(boost::system::error_code, std::size_t)`.

        @return The return value depends on the completion token type.
    */
    template<class ConstBufferSequence,
        BOOST_ASIO_COMPLETION_TOKEN_FOR(
            void(boost::system::error_code, std::size_t)) WriteToken>
    auto
    async_write_some(
        ConstBufferSequence const& buffers,
        WriteToken&& token)
    {
        return net::async_initiate<WriteToken,
            void(boost::system::error_code, std::size_t)>(
                initiate_async_write_some(this),
                token,
                buffers);
    }
};

/** Create a connected pair of uni_stream.

    @param ctx The asio_context for both streams.
    @return A pair of connected uni_stream objects.
*/
inline
std::pair<uni_stream, uni_stream>
make_uni_pair(asio_context& ctx)
{
    auto [s1, s2] = make_stream_pair(ctx);
    return {
        uni_stream(std::move(s1), ctx),
        uni_stream(std::move(s2), ctx)
    };
}

#endif

Build

add_executable(capy_example_asio_callbacks
    asio_callbacks.cpp
    api/capy_streams.cpp)
target_link_libraries(capy_example_asio_callbacks
    Boost::capy
    Boost::asio)

b2 builds the same program as exe asio_callbacks from example/asio/Jamfile.

Walkthrough

The Initiation

uni_stream holds a any_stream and a pointer to the asio_context. Its async_read_some passes three things to net::async_initiate: an initiate_async_read_some object, the completion token, and the buffers. The completion token decides what async_read_some returns. With a plain callback, it returns nothing.

Asio then calls initiate_async_read_some with the completion handler and the buffers. The initiation looks up the executor associated with the completion handler. The default is the executor of the io_context.

The initiation wraps that executor in asio_executor_wrapper, a Capy executor that posts every continuation through Asio. Then it starts a task with run_async on the wrapper. The task converts the buffers with capy::from_asio, awaits read_some on the any_stream, and calls the completion handler with the error code and the byte count.

The wrapper posts even when Capy asks it to dispatch. So the completion handler never runs inside the call to async_read_some, and it runs on the associated executor, as Asio requires. async_write_some and initiate_async_write_some follow the same pattern for writes.

A Callback Chain

do_write keeps its state in a std::shared_ptr, so every copy of the function object shares it. Each call writes one chunk of up to 128 bytes with net::async_write. That composed operation calls async_write_some on the adapter until the whole chunk is written.

The completion handler of each write adds the bytes written, prints the total, and calls a copy of do_write again. The chain ends when it has written 1024 bytes. do_read is the same chain for net::async_read.

main makes a connected pair of adapters with make_uni_pair, starts both chains, and calls ctx.run(). The call returns when both chains end and no work remains.

Output

The two chains run over a local TCP connection, so their lines can interleave differently from run to run.

writer: wrote 128 bytes (total 128)
reader: read 128 bytes (total 128)
writer: wrote 128 bytes (total 256)
reader: read 128 bytes (total 256)
writer: wrote 128 bytes (total 384)
reader: read 128 bytes (total 384)
writer: wrote 128 bytes (total 512)
reader: read 128 bytes (total 512)
writer: wrote 128 bytes (total 640)
reader: read 128 bytes (total 640)
writer: wrote 128 bytes (total 768)
reader: read 128 bytes (total 768)
writer: wrote 128 bytes (total 896)
reader: read 128 bytes (total 896)
writer: wrote 128 bytes (total 1024)
writer: done, wrote 1024 bytes
reader: read 128 bytes (total 1024)
reader: done, read 1024 bytes
example complete!

Exercises

Rewrite the reader as an Asio coroutine with net::co_spawn, using the net::use_awaitable form that the comment on uni_stream shows.
Bind the write handler to an Asio strand with net::bind_executor, and confirm that the completion handler runs on that strand.