Diff coverage report for

//

//

// Copyright (c) 2025 Vinnie Falco (vinnie.falco@gmail.com)

// Copyright (c) 2025 Vinnie Falco (vinnie.falco@gmail.com)

//

//

// Distributed under the Boost Software License, Version 1.0. (See accompanying

// 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)

// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)

//

//

// Official repository: https://github.com/cppalliance/capy

// Official repository: https://github.com/cppalliance/capy

//

//

#ifndef BOOST_CAPY_TEST_STREAM_HPP

#ifndef BOOST_CAPY_TEST_STREAM_HPP

#define BOOST_CAPY_TEST_STREAM_HPP

#define BOOST_CAPY_TEST_STREAM_HPP

#include <boost/capy/detail/config.hpp>

#include <boost/capy/detail/config.hpp>

#include <boost/capy/buffers.hpp>

#include <boost/capy/buffers.hpp>

#include <boost/capy/buffers/buffer_copy.hpp>

#include <boost/capy/buffers/buffer_copy.hpp>

#include <boost/capy/buffers/make_buffer.hpp>

#include <boost/capy/buffers/make_buffer.hpp>

#include <boost/capy/continuation.hpp>

#include <boost/capy/continuation.hpp>

#include <coroutine>

#include <coroutine>

#include <boost/capy/ex/io_env.hpp>

#include <boost/capy/ex/io_env.hpp>

#include <boost/capy/io_result.hpp>

#include <boost/capy/io_result.hpp>

#include <boost/capy/error.hpp>

#include <boost/capy/error.hpp>

#include <boost/capy/read.hpp>

#include <boost/capy/read.hpp>

#include <boost/capy/task.hpp>

#include <boost/capy/task.hpp>

#include <boost/capy/test/fuse.hpp>

#include <boost/capy/test/fuse.hpp>

#include <boost/capy/test/run_blocking.hpp>

#include <boost/capy/test/run_blocking.hpp>

#include <memory>

#include <memory>

#include <stop_token>

#include <stop_token>

#include <string>

#include <string>

#include <string_view>

#include <string_view>

#include <utility>

#include <utility>

namespace boost {

namespace boost {

namespace capy {

namespace capy {

namespace test {

namespace test {

/** A connected stream for testing bidirectional I/O.

/** A connected stream for testing bidirectional I/O.

    Streams are created in pairs via @ref make_stream_pair.

    Streams are created in pairs via @ref make_stream_pair.

    Data written to one end becomes available for reading on

    Data written to one end becomes available for reading on

    the other. If no data is available when @ref read_some

    the other. If no data is available when @ref read_some

    is called, the calling coroutine suspends until the peer

    is called, the calling coroutine suspends until the peer

    calls @ref write_some. The shared @ref fuse enables error

    calls @ref write_some. The shared @ref fuse enables error

    injection at controlled points in both directions.

    injection at controlled points in both directions.

    When the fuse injects an error or throws on one end, the

    When the fuse injects an error or throws on one end, the

    other end is automatically closed: any suspended reader is

    other end is automatically closed: any suspended reader is

    resumed with `error::eof`, and subsequent operations on

    resumed with `error::eof`, and subsequent operations on

    both ends return `error::eof`. Calling @ref close on one

    both ends return `error::eof`. Calling @ref close on one

    end signals eof to the peer's reads after draining any

    end signals eof to the peer's reads after draining any

    buffered data, while the peer may still write.

    buffered data, while the peer may still write.

    @par Thread Safety

    @par Thread Safety

    Single-threaded only. Both ends of the pair must be

    Single-threaded only. Both ends of the pair must be

    accessed from the same thread. Concurrent access is

    accessed from the same thread. Concurrent access is

    undefined behavior.

    undefined behavior.

    @par Example

    @par Example

    @code

    @code

    fuse f;

    fuse f;

    auto [a, b] = make_stream_pair( f );

    auto [a, b] = make_stream_pair( f );

    auto r = f.armed( [&]( fuse& ) -> task<> {

    auto r = f.armed( [&]( fuse& ) -> task<> {

        auto [ec, n] = co_await a.write_some(

        auto [ec, n] = co_await a.write_some(

            const_buffer( "hello", 5 ) );

            const_buffer( "hello", 5 ) );

        if( ec )

        if( ec )

            co_return;

            co_return;

        char buf[32];

        char buf[32];

        auto [ec2, n2] = co_await b.read_some(

        auto [ec2, n2] = co_await b.read_some(

            mutable_buffer( buf, sizeof( buf ) ) );

            mutable_buffer( buf, sizeof( buf ) ) );

        if( ec2 )

        if( ec2 )

            co_return;

            co_return;

        // buf contains "hello"

        // buf contains "hello"

    } );

    } );

    @endcode

    @endcode

    @see make_stream_pair, fuse

    @see make_stream_pair, fuse

*/

*/

class stream

class stream

{

{

    // Single-threaded only. No concurrent access to either

    // Single-threaded only. No concurrent access to either

    // end of the pair. Both streams and all operations must

    // end of the pair. Both streams and all operations must

    // run on the same thread.

    // run on the same thread.

    struct half

    struct half

    {

    {

        std::string buf;

        std::string buf;

        std::size_t max_read_size = std::size_t(-1);

        std::size_t max_read_size = std::size_t(-1);

        continuation pending_cont_;

        continuation pending_cont_;

        executor_ref pending_ex;

        executor_ref pending_ex;

        bool eof = false;

        bool eof = false;

    };

    };

    struct state

    struct state

    {

    {

        fuse f;

        fuse f;

        bool closed = false;

        bool closed = false;

        half sides[2];

        half sides[2];

        {

        {

        // Set closed and resume any suspended readers

        // Set closed and resume any suspended readers

        // with eof on both sides.

        // with eof on both sides.

        {

        {

            {

            {

                {

                {

                }

                }

            }

            }

    };

    };

    // Wraps the maybe_fail() call. If the guard is

    // Wraps the maybe_fail() call. If the guard is

    // not disarmed before destruction (fuse returned

    // not disarmed before destruction (fuse returned

    // an error, or threw an exception), closes both

    // an error, or threw an exception), closes both

    // ends so any suspended peer gets eof.

    // ends so any suspended peer gets eof.

    struct close_guard

    struct close_guard

    {

    {

        state* st;

        state* st;

        bool armed = true;

        bool armed = true;

    };

    };

    std::shared_ptr<state> state_;

    std::shared_ptr<state> state_;

    int index_;

    int index_;

        std::shared_ptr<state> sp,

        std::shared_ptr<state> sp,

        int index) noexcept

        int index) noexcept

    {

    {

    friend std::pair<stream, stream>

    friend std::pair<stream, stream>

    make_stream_pair(fuse);

    make_stream_pair(fuse);

public:

public:

    stream(stream const&) = delete;

    stream(stream const&) = delete;

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

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

    stream& operator=(stream&&) = default;

    stream& operator=(stream&&) = default;

    /** Signal end-of-stream to the peer.

    /** Signal end-of-stream to the peer.

        Marks the peer's read direction as closed.

        Marks the peer's read direction as closed.

        If the peer is suspended in @ref read_some,

        If the peer is suspended in @ref read_some,

        it is resumed. The peer drains any buffered

        it is resumed. The peer drains any buffered

        data before receiving `error::eof`. Writes

        data before receiving `error::eof`. Writes

        from the peer are unaffected.

        from the peer are unaffected.

    */

    */

    void

    void

    {

    {

        {

        {

        }

        }

    /** Set the maximum bytes returned per read.

    /** Set the maximum bytes returned per read.

        Limits how many bytes @ref read_some returns in

        Limits how many bytes @ref read_some returns in

        a single call, simulating chunked network delivery.

        a single call, simulating chunked network delivery.

        The default is unlimited.

        The default is unlimited.

        @param n Maximum bytes per read.

        @param n Maximum bytes per read.

    */

    */

    void

    void

    {

    {

    /** Asynchronously read data from the stream.

    /** Asynchronously read data from the stream.

        Transfers up to `buffer_size(buffers)` bytes from

        Transfers up to `buffer_size(buffers)` bytes from

        data written by the peer. If no data is available,

        data written by the peer. If no data is available,

        the calling coroutine suspends until the peer calls

        the calling coroutine suspends until the peer calls

        @ref write_some. Before every read, the attached

        @ref write_some. Before every read, the attached

        @ref fuse is consulted to possibly inject an error.

        @ref fuse is consulted to possibly inject an error.

        If the fuse fires, the peer is automatically closed.

        If the fuse fires, the peer is automatically closed.

        If the stream is closed, returns `error::eof`.

        If the stream is closed, returns `error::eof`.

        The returned `std::size_t` is the number of bytes

        The returned `std::size_t` is the number of bytes

        transferred.

        transferred.

        @param buffers The mutable buffer sequence to receive data.

        @param buffers The mutable buffer sequence to receive data.

        @return An awaitable that await-returns `(error_code,std::size_t)`.

        @return An awaitable that await-returns `(error_code,std::size_t)`.

        @see fuse, close

        @see fuse, close

    */

    */

    template<MutableBufferSequence MB>

    template<MutableBufferSequence MB>

    auto

    auto

    {

    {

        struct awaitable

        struct awaitable

        {

        {

            stream* self_;

            stream* self_;

            MB buffers_;

            MB buffers_;

            {

            {

            }

            }

                std::coroutine_handle<> h,

                std::coroutine_handle<> h,

                io_env const* env) noexcept

                io_env const* env) noexcept

            {

            {

            }

            }

            io_result<std::size_t>

            io_result<std::size_t>

            {

            {

                {

                {

                    side.max_read_size);

                    side.max_read_size);

            }

            }

        };

        };

    }

    }

    /** Asynchronously write data to the stream.

    /** Asynchronously write data to the stream.

        Transfers up to `buffer_size(buffers)` bytes to the

        Transfers up to `buffer_size(buffers)` bytes to the

        peer's incoming buffer. If the peer is suspended in

        peer's incoming buffer. If the peer is suspended in

        @ref read_some, it is resumed. Before every write,

        @ref read_some, it is resumed. Before every write,

        the attached @ref fuse is consulted to possibly inject

        the attached @ref fuse is consulted to possibly inject

        an error. If the fuse fires, the peer is automatically

        an error. If the fuse fires, the peer is automatically

        closed. If the stream is closed, returns `error::eof`.

        closed. If the stream is closed, returns `error::eof`.

        The returned `std::size_t` is the number of bytes

        The returned `std::size_t` is the number of bytes

        transferred.

        transferred.

        @param buffers The const buffer sequence containing

        @param buffers The const buffer sequence containing

            data to write.

            data to write.

        @return An awaitable that await-returns `(error_code,std::size_t)`.

        @return An awaitable that await-returns `(error_code,std::size_t)`.

        @see fuse, close

        @see fuse, close

    */

    */

    template<ConstBufferSequence CB>

    template<ConstBufferSequence CB>

    auto

    auto

    {

    {

        struct awaitable

        struct awaitable

        {

        {

            stream* self_;

            stream* self_;

            CB buffers_;

            CB buffers_;

                std::coroutine_handle<>,

                std::coroutine_handle<>,

                io_env const*) const noexcept

                io_env const*) const noexcept

            {

            {

            io_result<std::size_t>

            io_result<std::size_t>

            {

            {

                {

                {

                }

                }

        };

        };

    }

    }

    /** Inject data into this stream's peer for reading.

    /** Inject data into this stream's peer for reading.

        Appends data directly to the peer's incoming buffer,

        Appends data directly to the peer's incoming buffer,

        bypassing the fuse. If the peer is suspended in

        bypassing the fuse. If the peer is suspended in

        @ref read_some, it is resumed. This is test setup,

        @ref read_some, it is resumed. This is test setup,

        not an operation under test.

        not an operation under test.

        @param sv The data to inject.

        @param sv The data to inject.

        @see make_stream_pair

        @see make_stream_pair

    */

    */

    void

    void

    {

    {

        {

        {

        }

        }

    /** Read from this stream and verify the content.

    /** Read from this stream and verify the content.

        Reads exactly `expected.size()` bytes from the stream

        Reads exactly `expected.size()` bytes from the stream

        and compares against the expected string. The read goes

        and compares against the expected string. The read goes

        through the normal path including the fuse.

        through the normal path including the fuse.

        @param expected The expected content.

        @param expected The expected content.

        @return A pair of `(error_code, bool)`. The error_code

        @return A pair of `(error_code, bool)`. The error_code

            is set if a read error occurs (e.g. fuse injection).

            is set if a read error occurs (e.g. fuse injection).

            The bool is true if the data matches.

            The bool is true if the data matches.

        @see provide

        @see provide

    */

    */

    std::pair<std::error_code, bool>

    std::pair<std::error_code, bool>

    {

    {

            stream& self,

            stream& self,

            std::string_view expected,

            std::string_view expected,

            std::error_code& result,

            std::error_code& result,

            bool& match) -> task<>

            bool& match) -> task<>

        {

        {

            std::string buf(expected.size(), '\0');

            std::string buf(expected.size(), '\0');

            auto [ec, n] = co_await read(

            auto [ec, n] = co_await read(

                self, mutable_buffer(

                self, mutable_buffer(

                    buf.data(), buf.size()));

                    buf.data(), buf.size()));

            if(ec)

            if(ec)

            {

            {

                result = ec;

                result = ec;

                co_return;

                co_return;

            }

            }

            match = (std::string_view(

            match = (std::string_view(

                buf.data(), n) == expected);

                buf.data(), n) == expected);

    }

    }

    /** Return the stream's pending read data.

    /** Return the stream's pending read data.

        Returns a view of the data waiting to be read

        Returns a view of the data waiting to be read

        from this stream. This is a direct peek at the

        from this stream. This is a direct peek at the

        internal buffer, bypassing the fuse.

        internal buffer, bypassing the fuse.

        @return A view of the pending data.

        @return A view of the pending data.

        @see provide, expect

        @see provide, expect

    */

    */

    std::string_view

    std::string_view

    data() const noexcept

    data() const noexcept

    {

    {

        return state_->sides[index_].buf;

        return state_->sides[index_].buf;

    }

    }

};

};

/** Create a connected pair of test streams.

/** Create a connected pair of test streams.

    Data written to one stream becomes readable on the other.

    Data written to one stream becomes readable on the other.

    If a coroutine calls @ref stream::read_some when no data

    If a coroutine calls @ref stream::read_some when no data

    is available, it suspends until the peer writes. Before

    is available, it suspends until the peer writes. Before

    every read or write, the @ref fuse is consulted to

    every read or write, the @ref fuse is consulted to

    possibly inject an error for testing fault scenarios.

    possibly inject an error for testing fault scenarios.

    When the fuse fires, the peer is automatically closed.

    When the fuse fires, the peer is automatically closed.

    @param f The fuse used to inject errors during operations.

    @param f The fuse used to inject errors during operations.

    @return A pair of connected streams.

    @return A pair of connected streams.

    @see stream, fuse

    @see stream, fuse

*/

*/

inline std::pair<stream, stream>

inline std::pair<stream, stream>

{

{

} // test

} // test

} // capy

} // capy

} // boost

} // boost

#endif

#endif