robot/include/asio2/bho/beast/websocket/impl/read.hpp

//
// Copyright (c) 2016-2019 Vinnie Falco (vinnie dot falco at gmail dot 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/boostorg/beast
//

#ifndef BHO_BEAST_WEBSOCKET_IMPL_READ_HPP
#define BHO_BEAST_WEBSOCKET_IMPL_READ_HPP

#include <asio2/bho/beast/core/buffer_traits.hpp>
#include <asio2/bho/beast/websocket/teardown.hpp>
#include <asio2/bho/beast/websocket/detail/mask.hpp>
#include <asio2/bho/beast/websocket/impl/stream_impl.hpp>
#include <asio2/bho/beast/core/async_base.hpp>
#include <asio2/bho/beast/core/bind_handler.hpp>
#include <asio2/bho/beast/core/buffers_prefix.hpp>
#include <asio2/bho/beast/core/buffers_suffix.hpp>
#include <asio2/bho/beast/core/flat_static_buffer.hpp>
#include <asio2/bho/beast/core/read_size.hpp>
#include <asio2/bho/beast/core/stream_traits.hpp>
#include <asio2/bho/beast/core/detail/bind_continuation.hpp>
#include <asio2/bho/beast/core/detail/buffer.hpp>
#include <asio2/bho/beast/core/detail/clamp.hpp>
#include <asio2/bho/beast/core/detail/config.hpp>
#include <asio/coroutine.hpp>
#include <asio2/bho/assert.hpp>
#include <asio2/bho/config.hpp>
#include <optional>
#include <asio2/bho/throw_exception.hpp>
#include <algorithm>
#include <limits>
#include <memory>

namespace bho {
namespace beast {
namespace websocket {

/*  Read some message data into a buffer sequence.

    Also reads and handles control frames.
*/
template<class NextLayer, bool deflateSupported>
template<class Handler, class MutableBufferSequence>
class stream<NextLayer, deflateSupported>::read_some_op
    : public beast::async_base<
        Handler, beast::executor_type<stream>>
    , public asio::coroutine
{
    std::weak_ptr<impl_type> wp_;
    MutableBufferSequence bs_;
    buffers_suffix<MutableBufferSequence> cb_;
    std::size_t bytes_written_ = 0;
    error_code result_;
    close_code code_;
    bool did_read_ = false;

public:
    static constexpr int id = 1; // for soft_mutex

    template<class Handler_>
    read_some_op(
        Handler_&& h,
        std::shared_ptr<impl_type> const& sp,
        MutableBufferSequence const& bs)
        : async_base<
            Handler, beast::executor_type<stream>>(
                std::forward<Handler_>(h),
                    sp->stream().get_executor())
        , wp_(sp)
        , bs_(bs)
        , cb_(bs)
        , code_(close_code::none)
    {
        (*this)({}, 0, false);
    }

    void operator()(
        error_code ec = {},
        std::size_t bytes_transferred = 0,
        bool cont = true)
    {
        using beast::detail::clamp;
        auto sp = wp_.lock();
        if(! sp)
        {
            BHO_BEAST_ASSIGN_EC(ec, net::error::operation_aborted);
            bytes_written_ = 0;
            return this->complete(cont, ec, bytes_written_);
        }
        auto& impl = *sp;
        ASIO_CORO_REENTER(*this)
        {
            impl.update_timer(this->get_executor());

        acquire_read_lock:
            // Acquire the read lock
            if(! impl.rd_block.try_lock(this))
            {
            do_suspend:
                ASIO_CORO_YIELD
                {
                    ASIO_HANDLER_LOCATION((
                        __FILE__, __LINE__,
                        "websocket::async_read_some"));

                    this->set_allowed_cancellation(net::cancellation_type::all);
                    impl.op_r_rd.emplace(std::move(*this), net::cancellation_type::all);
                }
                if (ec)
                    return this->complete(cont, ec, bytes_written_);

                this->set_allowed_cancellation(net::cancellation_type::terminal);

                impl.rd_block.lock(this);
                ASIO_CORO_YIELD
                {
                    ASIO_HANDLER_LOCATION((
                        __FILE__, __LINE__,
                        "websocket::async_read_some"));

                    const auto ex = this->get_immediate_executor();
                    net::dispatch(ex, std::move(*this));
                }
                BHO_ASSERT(impl.rd_block.is_locked(this));

                BHO_ASSERT(!ec);
                if(impl.check_stop_now(ec))
                {
                    // Issue 2264 - There is no guarantee that the next
                    // error will be operation_aborted.
                    // The error could be a result of the peer resetting the 
                    // connection
                    // BHO_ASSERT(ec == net::error::operation_aborted);
                    goto upcall;
                }
                // VFALCO Should never get here

                // The only way to get read blocked is if
                // a `close_op` wrote a close frame
                BHO_ASSERT(impl.wr_close);
                BHO_ASSERT(impl.status_ != status::open);
                BHO_BEAST_ASSIGN_EC(ec, net::error::operation_aborted);
                goto upcall;
            }
            else
            {
                // Make sure the stream is not closed
                if( impl.status_ == status::closed ||
                    impl.status_ == status::failed)
                {
                    BHO_BEAST_ASSIGN_EC(ec, net::error::operation_aborted);
                    goto upcall;
                }
            }

            // if status_ == status::closing, we want to suspend
            // the read operation until the close completes,
            // then finish the read with operation_aborted.

        loop:
            BHO_ASSERT(impl.rd_block.is_locked(this));
            // See if we need to read a frame header. This
            // condition is structured to give the decompressor
            // a chance to emit the final empty deflate block
            //
            if(impl.rd_remain == 0 &&
                (! impl.rd_fh.fin || impl.rd_done))
            {
                // Read frame header
                while(! impl.parse_fh(
                    impl.rd_fh, impl.rd_buf, result_))
                {
                    if(result_)
                    {
                        // _Fail the WebSocket Connection_
                        if(result_ == error::message_too_big)
                            code_ = close_code::too_big;
                        else
                            code_ = close_code::protocol_error;
                        goto close;
                    }
                    BHO_ASSERT(impl.rd_block.is_locked(this));
                    ASIO_CORO_YIELD
                    {
                        ASIO_HANDLER_LOCATION((
                            __FILE__, __LINE__,
                            "websocket::async_read_some"));

                        impl.stream().async_read_some(
                            impl.rd_buf.prepare(read_size(
                                impl.rd_buf, impl.rd_buf.max_size())),
                                    std::move(*this));
                    }
                    BHO_ASSERT(impl.rd_block.is_locked(this));
                    impl.rd_buf.commit(bytes_transferred);
                    if(impl.check_stop_now(ec))
                        goto upcall;
                    impl.reset_idle();

                    // Allow a close operation
                    // to acquire the read block
                    impl.rd_block.unlock(this);
                    if( impl.op_r_close.maybe_invoke())
                    {
                        // Suspend
                        BHO_ASSERT(impl.rd_block.is_locked());
                        goto do_suspend;
                    }
                    // Acquire read block
                    impl.rd_block.lock(this);
                }
                // Immediately apply the mask to the portion
                // of the buffer holding payload data.
                if(impl.rd_fh.len > 0 && impl.rd_fh.mask)
                    detail::mask_inplace(buffers_prefix(
                        clamp(impl.rd_fh.len),
                            impl.rd_buf.data()),
                                impl.rd_key);
                if(detail::is_control(impl.rd_fh.op))
                {
                    // Clear this otherwise the next
                    // frame will be considered final.
                    impl.rd_fh.fin = false;

                    // Handle ping frame
                    if(impl.rd_fh.op == detail::opcode::ping)
                    {
                        if(impl.ctrl_cb)
                        {
                            if(! cont)
                            {
                                ASIO_CORO_YIELD
                                {
                                    ASIO_HANDLER_LOCATION((
                                        __FILE__, __LINE__,
                                        "websocket::async_read_some"));

                                    const auto ex = this->get_immediate_executor();
                                    net::dispatch(ex, std::move(*this));
                                }
                                BHO_ASSERT(cont);
                                // VFALCO call check_stop_now() here?
                            }
                        }
                        {
                            auto const b = buffers_prefix(
                                clamp(impl.rd_fh.len),
                                    impl.rd_buf.data());
                            auto const len = buffer_bytes(b);
                            BHO_ASSERT(len == impl.rd_fh.len);
                            ping_data payload;
                            detail::read_ping(payload, b);
                            impl.rd_buf.consume(len);
                            // Ignore ping when closing
                            if(impl.status_ == status::closing)
                                goto loop;
                            if(impl.ctrl_cb)
                                impl.ctrl_cb(
                                    frame_type::ping, to_string_view(payload));
                            impl.rd_fb.clear();
                            impl.template write_ping<
                                flat_static_buffer_base>(impl.rd_fb,
                                    detail::opcode::pong, payload);
                        }

                        // Allow a close operation
                        // to acquire the read block
                        impl.rd_block.unlock(this);
                        impl.op_r_close.maybe_invoke();

                        // Acquire the write lock
                        if(! impl.wr_block.try_lock(this))
                        {
                            ASIO_CORO_YIELD
                            {
                                ASIO_HANDLER_LOCATION((
                                    __FILE__, __LINE__,
                                    "websocket::async_read_some"));

                                impl.op_rd.emplace(std::move(*this));
                            }
                            if (ec)
                                return this->complete(cont, ec, bytes_written_);

                            impl.wr_block.lock(this);
                            ASIO_CORO_YIELD
                            {
                                ASIO_HANDLER_LOCATION((
                                    __FILE__, __LINE__,
                                    "websocket::async_read_some"));

                                const auto ex = this->get_immediate_executor();
                                net::dispatch(ex, std::move(*this));
                            }
                            BHO_ASSERT(impl.wr_block.is_locked(this));
                            if(impl.check_stop_now(ec))
                                goto upcall;
                        }

                        // Send pong
                        BHO_ASSERT(impl.wr_block.is_locked(this));
                        ASIO_CORO_YIELD
                        {
                            ASIO_HANDLER_LOCATION((
                                __FILE__, __LINE__,
                                "websocket::async_read_some"));

                            net::async_write(
                                impl.stream(), net::const_buffer(impl.rd_fb.data()),
                                beast::detail::bind_continuation(std::move(*this)));
                        }
                        BHO_ASSERT(impl.wr_block.is_locked(this));
                        if(impl.check_stop_now(ec))
                            goto upcall;
                        impl.wr_block.unlock(this);
                        impl.op_close.maybe_invoke()
                            || impl.op_idle_ping.maybe_invoke()
                            || impl.op_ping.maybe_invoke()
                            || impl.op_wr.maybe_invoke();
                        goto acquire_read_lock;
                    }

                    // Handle pong frame
                    if(impl.rd_fh.op == detail::opcode::pong)
                    {
                        // Ignore pong when closing
                        if(! impl.wr_close && impl.ctrl_cb)
                        {
                            if(! cont)
                            {
                                ASIO_CORO_YIELD
                                {
                                    ASIO_HANDLER_LOCATION((
                                        __FILE__, __LINE__,
                                        "websocket::async_read_some"));

                                    const auto ex = this->get_immediate_executor();
                                    net::dispatch(ex, std::move(*this));
                                }
                                BHO_ASSERT(cont);
                            }
                        }
                        auto const cb = buffers_prefix(clamp(
                            impl.rd_fh.len), impl.rd_buf.data());
                        auto const len = buffer_bytes(cb);
                        BHO_ASSERT(len == impl.rd_fh.len);
                        ping_data payload;
                        detail::read_ping(payload, cb);
                        impl.rd_buf.consume(len);
                        // Ignore pong when closing
                        if(! impl.wr_close && impl.ctrl_cb)
                            impl.ctrl_cb(frame_type::pong, to_string_view(payload));
                        goto loop;
                    }

                    // Handle close frame
                    BHO_ASSERT(impl.rd_fh.op == detail::opcode::close);
                    {
                        if(impl.ctrl_cb)
                        {
                            if(! cont)
                            {
                                ASIO_CORO_YIELD
                                {
                                    ASIO_HANDLER_LOCATION((
                                        __FILE__, __LINE__,
                                        "websocket::async_read_some"));

                                    const auto ex = this->get_immediate_executor();
                                    net::dispatch(ex, std::move(*this));
                                }
                                BHO_ASSERT(cont);
                            }
                        }
                        auto const cb = buffers_prefix(clamp(
                            impl.rd_fh.len), impl.rd_buf.data());
                        auto const len = buffer_bytes(cb);
                        BHO_ASSERT(len == impl.rd_fh.len);
                        BHO_ASSERT(! impl.rd_close);
                        impl.rd_close = true;
                        close_reason cr;
                        detail::read_close(cr, cb, result_);
                        if(result_)
                        {
                            // _Fail the WebSocket Connection_
                            code_ = close_code::protocol_error;
                            goto close;
                        }
                        impl.cr = cr;
                        impl.rd_buf.consume(len);
                        if(impl.ctrl_cb)
                            impl.ctrl_cb(frame_type::close,
                                to_string_view(impl.cr.reason));
                        // See if we are already closing
                        if(impl.status_ == status::closing)
                        {
                            // _Close the WebSocket Connection_
                            BHO_ASSERT(impl.wr_close);
                            code_ = close_code::none;
                            result_ = error::closed;
                            goto close;
                        }
                        // _Start the WebSocket Closing Handshake_
                        code_ = cr.code == close_code::none ?
                            close_code::normal :
                            static_cast<close_code>(cr.code);
                        result_ = error::closed;
                        goto close;
                    }
                }
                if(impl.rd_fh.len == 0 && ! impl.rd_fh.fin)
                {
                    // Empty non-final frame
                    goto loop;
                }
                impl.rd_done = false;
            }
            if(! impl.rd_deflated())
            {
                if(impl.rd_remain > 0)
                {
                    if(impl.rd_buf.size() == 0 && impl.rd_buf.max_size() >
                        (std::min)(clamp(impl.rd_remain),
                            buffer_bytes(cb_)))
                    {
                        // Fill the read buffer first, otherwise we
                        // get fewer bytes at the cost of one I/O.
                        ASIO_CORO_YIELD
                        {
                            ASIO_HANDLER_LOCATION((
                                __FILE__, __LINE__,
                                "websocket::async_read_some"));

                            impl.stream().async_read_some(
                                impl.rd_buf.prepare(read_size(
                                    impl.rd_buf, impl.rd_buf.max_size())),
                                        std::move(*this));
                        }
                        impl.rd_buf.commit(bytes_transferred);
                        if(impl.check_stop_now(ec))
                            goto upcall;
                        impl.reset_idle();
                        if(impl.rd_fh.mask)
                            detail::mask_inplace(buffers_prefix(clamp(
                                impl.rd_remain), impl.rd_buf.data()),
                                    impl.rd_key);
                    }
                    if(impl.rd_buf.size() > 0)
                    {
                        // Copy from the read buffer.
                        // The mask was already applied.
                        bytes_transferred = net::buffer_copy(cb_,
                            impl.rd_buf.data(), clamp(impl.rd_remain));
                        auto const mb = buffers_prefix(
                            bytes_transferred, cb_);
                        impl.rd_remain -= bytes_transferred;
                        if(impl.rd_op == detail::opcode::text)
                        {
                            if(! impl.rd_utf8.write(mb) ||
                                (impl.rd_remain == 0 && impl.rd_fh.fin &&
                                    ! impl.rd_utf8.finish()))
                            {
                                // _Fail the WebSocket Connection_
                                code_ = close_code::bad_payload;
                                result_ = error::bad_frame_payload;
                                goto close;
                            }
                        }
                        bytes_written_ += bytes_transferred;
                        impl.rd_size += bytes_transferred;
                        impl.rd_buf.consume(bytes_transferred);
                    }
                    else
                    {
                        // Read into caller's buffer
                        BHO_ASSERT(impl.rd_remain > 0);
                        BHO_ASSERT(buffer_bytes(cb_) > 0);
                        BHO_ASSERT(buffer_bytes(buffers_prefix(
                            clamp(impl.rd_remain), cb_)) > 0);
                        ASIO_CORO_YIELD
                        {
                            ASIO_HANDLER_LOCATION((
                                __FILE__, __LINE__,
                                "websocket::async_read_some"));

                            impl.stream().async_read_some(buffers_prefix(
                                clamp(impl.rd_remain), cb_), std::move(*this));
                        }
                        if(impl.check_stop_now(ec))
                            goto upcall;
                        impl.reset_idle();
                        BHO_ASSERT(bytes_transferred > 0);
                        auto const mb = buffers_prefix(
                            bytes_transferred, cb_);
                        impl.rd_remain -= bytes_transferred;
                        if(impl.rd_fh.mask)
                            detail::mask_inplace(mb, impl.rd_key);
                        if(impl.rd_op == detail::opcode::text)
                        {
                            if(! impl.rd_utf8.write(mb) ||
                                (impl.rd_remain == 0 && impl.rd_fh.fin &&
                                    ! impl.rd_utf8.finish()))
                            {
                                // _Fail the WebSocket Connection_
                                code_ = close_code::bad_payload;
                                result_ = error::bad_frame_payload;
                                goto close;
                            }
                        }
                        bytes_written_ += bytes_transferred;
                        impl.rd_size += bytes_transferred;
                    }
                }
                BHO_ASSERT( ! impl.rd_done );
                if( impl.rd_remain == 0 && impl.rd_fh.fin )
                    impl.rd_done = true;
            }
            else
            {
                // Read compressed message frame payload:
                // inflate even if rd_fh_.len == 0, otherwise we
                // never emit the end-of-stream deflate block.
                while(buffer_bytes(cb_) > 0)
                {
                    if( impl.rd_remain > 0 &&
                        impl.rd_buf.size() == 0 &&
                        ! did_read_)
                    {
                        // read new
                        ASIO_CORO_YIELD
                        {
                            ASIO_HANDLER_LOCATION((
                                __FILE__, __LINE__,
                                "websocket::async_read_some"));

                            impl.stream().async_read_some(
                                impl.rd_buf.prepare(read_size(
                                    impl.rd_buf, impl.rd_buf.max_size())),
                                        std::move(*this));
                        }
                        if(impl.check_stop_now(ec))
                            goto upcall;
                        impl.reset_idle();
                        BHO_ASSERT(bytes_transferred > 0);
                        impl.rd_buf.commit(bytes_transferred);
                        if(impl.rd_fh.mask)
                            detail::mask_inplace(
                                buffers_prefix(clamp(impl.rd_remain),
                                    impl.rd_buf.data()), impl.rd_key);
                        did_read_ = true;
                    }
                    zlib::z_params zs;
                    {
                        auto const out = buffers_front(cb_);
                        zs.next_out = out.data();
                        zs.avail_out = out.size();
                        BHO_ASSERT(zs.avail_out > 0);
                    }
                    // boolean to track the end of the message.
                    bool fin = false;
                    if(impl.rd_remain > 0)
                    {
                        if(impl.rd_buf.size() > 0)
                        {
                            // use what's there
                            auto const in = buffers_prefix(
                                clamp(impl.rd_remain), buffers_front(
                                    impl.rd_buf.data()));
                            zs.avail_in = in.size();
                            zs.next_in = in.data();
                        }
                        else
                        {
                            break;
                        }
                    }
                    else if(impl.rd_fh.fin)
                    {
                        // append the empty block codes
                        static std::uint8_t constexpr
                            empty_block[4] = { 0x00, 0x00, 0xff, 0xff };
                        zs.next_in = empty_block;
                        zs.avail_in = sizeof(empty_block);
                        fin = true;
                    }
                    else
                    {
                        break;
                    }
                    impl.inflate(zs, zlib::Flush::sync, ec);
                    if(impl.check_stop_now(ec))
                        goto upcall;
                    if(fin && zs.total_out == 0) {
                        impl.do_context_takeover_read(impl.role);
                        impl.rd_done = true;
                        break;
                    }
                    if(impl.rd_msg_max && beast::detail::sum_exceeds(
                        impl.rd_size, zs.total_out, impl.rd_msg_max))
                    {
                        // _Fail the WebSocket Connection_
                        code_ = close_code::too_big;
                        result_ = error::message_too_big;
                        goto close;
                    }
                    cb_.consume(zs.total_out);
                    impl.rd_size += zs.total_out;
                    if (! fin) {
                        impl.rd_remain -= zs.total_in;
                        impl.rd_buf.consume(zs.total_in);
                    }
                    bytes_written_ += zs.total_out;
                }
                if(impl.rd_op == detail::opcode::text)
                {
                    // check utf8
                    if(! impl.rd_utf8.write(
                        buffers_prefix(bytes_written_, bs_)) || (
                            impl.rd_done && ! impl.rd_utf8.finish()))
                    {
                        // _Fail the WebSocket Connection_
                        code_ = close_code::bad_payload;
                        result_ = error::bad_frame_payload;
                        goto close;
                    }
                }
            }
            goto upcall;

        close:
            // Acquire the write lock
            if(! impl.wr_block.try_lock(this))
            {
                ASIO_CORO_YIELD
                {
                    ASIO_HANDLER_LOCATION((
                        __FILE__, __LINE__,
                        "websocket::async_read_some"));

                    impl.op_rd.emplace(std::move(*this));
                }
                if (ec)
                    return this->complete(cont, ec, bytes_written_);

                impl.wr_block.lock(this);
                ASIO_CORO_YIELD
                {
                    ASIO_HANDLER_LOCATION((
                        __FILE__, __LINE__,
                        "websocket::async_read_some"));

                    const auto ex = this->get_immediate_executor();
                    net::dispatch(ex, std::move(*this));
                }
                BHO_ASSERT(impl.wr_block.is_locked(this));
                if(impl.check_stop_now(ec))
                    goto upcall;
            }

            impl.change_status(status::closing);

            if(! impl.wr_close)
            {
                impl.wr_close = true;

                // Serialize close frame
                impl.rd_fb.clear();
                impl.template write_close<
                    flat_static_buffer_base>(
                        impl.rd_fb, code_);

                // Send close frame
                BHO_ASSERT(impl.wr_block.is_locked(this));
                ASIO_CORO_YIELD
                {
                    ASIO_HANDLER_LOCATION((
                        __FILE__, __LINE__,
                        "websocket::async_read_some"));

                    net::async_write(impl.stream(), net::const_buffer(impl.rd_fb.data()),
                        beast::detail::bind_continuation(std::move(*this)));
                }
                BHO_ASSERT(impl.wr_block.is_locked(this));
                if(impl.check_stop_now(ec))
                    goto upcall;
            }

            // Teardown
            using beast::websocket::async_teardown;
            BHO_ASSERT(impl.wr_block.is_locked(this));
            ASIO_CORO_YIELD
            {
                ASIO_HANDLER_LOCATION((
                    __FILE__, __LINE__,
                    "websocket::async_read_some"));

                async_teardown(impl.role, impl.stream(),
                    beast::detail::bind_continuation(std::move(*this)));
            }
            BHO_ASSERT(impl.wr_block.is_locked(this));
            if(ec == net::error::eof)
            {
                // Rationale:
                // http://stackoverflow.com/questions/25587403/boost-asio-ssl-async-shutdown-always-finishes-with-an-error
                ec = {};
            }
            if(! ec)
            {
                BHO_BEAST_ASSIGN_EC(ec, result_);
            }
            if(ec && ec != error::closed)
                impl.change_status(status::failed);
            else
                impl.change_status(status::closed);
            impl.close();

        upcall:
            impl.rd_block.try_unlock(this);
            impl.op_r_close.maybe_invoke();
            if(impl.wr_block.try_unlock(this))
                impl.op_close.maybe_invoke()
                    || impl.op_idle_ping.maybe_invoke()
                    || impl.op_ping.maybe_invoke()
                    || impl.op_wr.maybe_invoke();
            this->complete(cont, ec, bytes_written_);
        }
    }
};

//------------------------------------------------------------------------------

template<class NextLayer, bool deflateSupported>
template<class Handler,  class DynamicBuffer>
class stream<NextLayer, deflateSupported>::read_op
    : public beast::async_base<
        Handler, beast::executor_type<stream>>
    , public asio::coroutine
{
    std::weak_ptr<impl_type> wp_;
    DynamicBuffer& b_;
    std::size_t limit_;
    std::size_t bytes_written_ = 0;
    bool some_;

public:
    template<class Handler_>
    read_op(
        Handler_&& h,
        std::shared_ptr<impl_type> const& sp,
        DynamicBuffer& b,
        std::size_t limit,
        bool some)
        : async_base<Handler,
            beast::executor_type<stream>>(
                std::forward<Handler_>(h),
                    sp->stream().get_executor())
        , wp_(sp)
        , b_(b)
        , limit_(limit ? limit : (
            std::numeric_limits<std::size_t>::max)())
        , some_(some)
    {
        (*this)({}, 0, false);
    }

    void operator()(
        error_code ec = {},
        std::size_t bytes_transferred = 0,
        bool cont = true)
    {
        using beast::detail::clamp;
        auto sp = wp_.lock();
        if(! sp)
        {
            BHO_BEAST_ASSIGN_EC(ec, net::error::operation_aborted);
            bytes_written_ = 0;
            return this->complete(cont, ec, bytes_written_);
        }
        auto& impl = *sp;
        using mutable_buffers_type = typename
            DynamicBuffer::mutable_buffers_type;
        ASIO_CORO_REENTER(*this)
        {
            do
            {
                // VFALCO TODO use bho::beast::bind_continuation
                ASIO_CORO_YIELD
                {
                    auto mb = beast::detail::dynamic_buffer_prepare(b_,
                        clamp(impl.read_size_hint_db(b_), limit_),
                            ec, error::buffer_overflow);
                    if(impl.check_stop_now(ec))
                        goto upcall;

                    ASIO_HANDLER_LOCATION((
                        __FILE__, __LINE__,
                        "websocket::async_read"));

                    read_some_op<read_op, mutable_buffers_type>(
                        std::move(*this), sp, *mb);
                }

                b_.commit(bytes_transferred);
                bytes_written_ += bytes_transferred;
                if(ec)
                    goto upcall;
            }
            while(! some_ && ! impl.rd_done);

        upcall:
            this->complete(cont, ec, bytes_written_);
        }
    }
};

template<class NextLayer, bool deflateSupported>
struct stream<NextLayer, deflateSupported>::
    run_read_some_op
{
    template<
        class ReadHandler,
        class MutableBufferSequence>
    void
    operator()(
        ReadHandler&& h,
        std::shared_ptr<impl_type> const& sp,
        MutableBufferSequence const& b)
    {
        // If you get an error on the following line it means
        // that your handler does not meet the documented type
        // requirements for the handler.

        static_assert(
            beast::detail::is_invocable<ReadHandler,
                void(error_code, std::size_t)>::value,
            "ReadHandler type requirements not met");

        read_some_op<
            typename std::decay<ReadHandler>::type,
            MutableBufferSequence>(
                std::forward<ReadHandler>(h),
                sp,
                b);
    }
};

template<class NextLayer, bool deflateSupported>
struct stream<NextLayer, deflateSupported>::
    run_read_op
{
    template<
        class ReadHandler,
        class DynamicBuffer>
    void
    operator()(
        ReadHandler&& h,
        std::shared_ptr<impl_type> const& sp,
        DynamicBuffer* b,
        std::size_t limit,
        bool some)
    {
        // If you get an error on the following line it means
        // that your handler does not meet the documented type
        // requirements for the handler.

        static_assert(
            beast::detail::is_invocable<ReadHandler,
                void(error_code, std::size_t)>::value,
            "ReadHandler type requirements not met");

        read_op<
            typename std::decay<ReadHandler>::type,
            DynamicBuffer>(
                std::forward<ReadHandler>(h),
                sp,
                *b,
                limit,
                some);
    }
};

//------------------------------------------------------------------------------

template<class NextLayer, bool deflateSupported>
template<class DynamicBuffer>
std::size_t
stream<NextLayer, deflateSupported>::
read(DynamicBuffer& buffer)
{
    static_assert(is_sync_stream<next_layer_type>::value,
        "SyncStream type requirements not met");
    static_assert(
        net::is_dynamic_buffer<DynamicBuffer>::value,
        "DynamicBuffer type requirements not met");
    error_code ec;
    auto const bytes_written = read(buffer, ec);
    if(ec)
        BHO_THROW_EXCEPTION(system_error{ec});
    return bytes_written;
}

template<class NextLayer, bool deflateSupported>
template<class DynamicBuffer>
std::size_t
stream<NextLayer, deflateSupported>::
read(DynamicBuffer& buffer, error_code& ec)
{
    static_assert(is_sync_stream<next_layer_type>::value,
        "SyncStream type requirements not met");
    static_assert(
        net::is_dynamic_buffer<DynamicBuffer>::value,
        "DynamicBuffer type requirements not met");
    std::size_t bytes_written = 0;
    do
    {
        bytes_written += read_some(buffer, 0, ec);
        if(ec)
            return bytes_written;
    }
    while(! is_message_done());
    return bytes_written;
}

template<class NextLayer, bool deflateSupported>
template<class DynamicBuffer, BHO_BEAST_ASYNC_TPARAM2 ReadHandler>
BHO_BEAST_ASYNC_RESULT2(ReadHandler)
stream<NextLayer, deflateSupported>::
async_read(DynamicBuffer& buffer, ReadHandler&& handler)
{
    static_assert(is_async_stream<next_layer_type>::value,
        "AsyncStream type requirements not met");
    static_assert(
        net::is_dynamic_buffer<DynamicBuffer>::value,
        "DynamicBuffer type requirements not met");
    return net::async_initiate<
        ReadHandler,
        void(error_code, std::size_t)>(
            run_read_op{},
            handler,
            impl_,
            &buffer,
            0,
            false);
}

//------------------------------------------------------------------------------

template<class NextLayer, bool deflateSupported>
template<class DynamicBuffer>
std::size_t
stream<NextLayer, deflateSupported>::
read_some(
    DynamicBuffer& buffer,
    std::size_t limit)
{
    static_assert(is_sync_stream<next_layer_type>::value,
        "SyncStream type requirements not met");
    static_assert(
        net::is_dynamic_buffer<DynamicBuffer>::value,
        "DynamicBuffer type requirements not met");
    error_code ec;
    auto const bytes_written =
        read_some(buffer, limit, ec);
    if(ec)
        BHO_THROW_EXCEPTION(system_error{ec});
    return bytes_written;
}

template<class NextLayer, bool deflateSupported>
template<class DynamicBuffer>
std::size_t
stream<NextLayer, deflateSupported>::
read_some(
    DynamicBuffer& buffer,
    std::size_t limit,
    error_code& ec)
{
    static_assert(is_sync_stream<next_layer_type>::value,
        "SyncStream type requirements not met");
    static_assert(
        net::is_dynamic_buffer<DynamicBuffer>::value,
        "DynamicBuffer type requirements not met");
    using beast::detail::clamp;
    if(! limit)
        limit = (std::numeric_limits<std::size_t>::max)();
    auto const size =
        clamp(read_size_hint(buffer), limit);
    BHO_ASSERT(size > 0);
    auto mb = beast::detail::dynamic_buffer_prepare(
        buffer, size, ec, error::buffer_overflow);
    if(impl_->check_stop_now(ec))
        return 0;
    auto const bytes_written = read_some(*mb, ec);
    buffer.commit(bytes_written);
    return bytes_written;
}

template<class NextLayer, bool deflateSupported>
template<class DynamicBuffer, BHO_BEAST_ASYNC_TPARAM2 ReadHandler>
BHO_BEAST_ASYNC_RESULT2(ReadHandler)
stream<NextLayer, deflateSupported>::
async_read_some(
    DynamicBuffer& buffer,
    std::size_t limit,
    ReadHandler&& handler)
{
    static_assert(is_async_stream<next_layer_type>::value,
        "AsyncStream type requirements not met");
    static_assert(
        net::is_dynamic_buffer<DynamicBuffer>::value,
        "DynamicBuffer type requirements not met");
    return net::async_initiate<
        ReadHandler,
        void(error_code, std::size_t)>(
            run_read_op{},
            handler,
            impl_,
            &buffer,
            limit,
            true);
}

//------------------------------------------------------------------------------

template<class NextLayer, bool deflateSupported>
template<class MutableBufferSequence>
std::size_t
stream<NextLayer, deflateSupported>::
read_some(
    MutableBufferSequence const& buffers)
{
    static_assert(is_sync_stream<next_layer_type>::value,
        "SyncStream type requirements not met");
    static_assert(net::is_mutable_buffer_sequence<
            MutableBufferSequence>::value,
        "MutableBufferSequence type requirements not met");
    error_code ec;
    auto const bytes_written = read_some(buffers, ec);
    if(ec)
        BHO_THROW_EXCEPTION(system_error{ec});
    return bytes_written;
}

template<class NextLayer, bool deflateSupported>
template<class MutableBufferSequence>
std::size_t
stream<NextLayer, deflateSupported>::
read_some(
    MutableBufferSequence const& buffers,
    error_code& ec)
{
    static_assert(is_sync_stream<next_layer_type>::value,
        "SyncStream type requirements not met");
    static_assert(net::is_mutable_buffer_sequence<
            MutableBufferSequence>::value,
        "MutableBufferSequence type requirements not met");
    using beast::detail::clamp;
    auto& impl = *impl_;
    close_code code{};
    std::size_t bytes_written = 0;
    ec = {};
    // Make sure the stream is open
    if(impl.check_stop_now(ec))
        return bytes_written;
loop:
    // See if we need to read a frame header. This
    // condition is structured to give the decompressor
    // a chance to emit the final empty deflate block
    //
    if(impl.rd_remain == 0 && (
        ! impl.rd_fh.fin || impl.rd_done))
    {
        // Read frame header
        error_code result;
        while(! impl.parse_fh(impl.rd_fh, impl.rd_buf, result))
        {
            if(result)
            {
                // _Fail the WebSocket Connection_
                if(result == error::message_too_big)
                    code = close_code::too_big;
                else
                    code = close_code::protocol_error;
                do_fail(code, result, ec);
                return bytes_written;
            }
            auto const bytes_transferred =
                impl.stream().read_some(
                    impl.rd_buf.prepare(read_size(
                        impl.rd_buf, impl.rd_buf.max_size())),
                    ec);
            impl.rd_buf.commit(bytes_transferred);
            if(impl.check_stop_now(ec))
                return bytes_written;
        }
        // Immediately apply the mask to the portion
        // of the buffer holding payload data.
        if(impl.rd_fh.len > 0 && impl.rd_fh.mask)
            detail::mask_inplace(buffers_prefix(
                clamp(impl.rd_fh.len), impl.rd_buf.data()),
                    impl.rd_key);
        if(detail::is_control(impl.rd_fh.op))
        {
            // Get control frame payload
            auto const b = buffers_prefix(
                clamp(impl.rd_fh.len), impl.rd_buf.data());
            auto const len = buffer_bytes(b);
            BHO_ASSERT(len == impl.rd_fh.len);

            // Clear this otherwise the next
            // frame will be considered final.
            impl.rd_fh.fin = false;

            // Handle ping frame
            if(impl.rd_fh.op == detail::opcode::ping)
            {
                ping_data payload;
                detail::read_ping(payload, b);
                impl.rd_buf.consume(len);
                if(impl.wr_close)
                {
                    // Ignore ping when closing
                    goto loop;
                }
                if(impl.ctrl_cb)
                    impl.ctrl_cb(frame_type::ping, to_string_view(payload));
                detail::frame_buffer fb;
                impl.template write_ping<flat_static_buffer_base>(fb,
                    detail::opcode::pong, payload);
                net::write(impl.stream(), fb.data(), ec);
                if(impl.check_stop_now(ec))
                    return bytes_written;
                goto loop;
            }
            // Handle pong frame
            if(impl.rd_fh.op == detail::opcode::pong)
            {
                ping_data payload;
                detail::read_ping(payload, b);
                impl.rd_buf.consume(len);
                if(impl.ctrl_cb)
                    impl.ctrl_cb(frame_type::pong, to_string_view(payload));
                goto loop;
            }
            // Handle close frame
            BHO_ASSERT(impl.rd_fh.op == detail::opcode::close);
            {
                BHO_ASSERT(! impl.rd_close);
                impl.rd_close = true;
                close_reason cr;
                detail::read_close(cr, b, result);
                if(result)
                {
                    // _Fail the WebSocket Connection_
                    do_fail(close_code::protocol_error,
                        result, ec);
                    return bytes_written;
                }
                impl.cr = cr;
                impl.rd_buf.consume(len);
                if(impl.ctrl_cb)
                    impl.ctrl_cb(frame_type::close, to_string_view(impl.cr.reason));
                BHO_ASSERT(! impl.wr_close);
                // _Start the WebSocket Closing Handshake_
                do_fail(
                    cr.code == close_code::none ?
                        close_code::normal :
                        static_cast<close_code>(cr.code),
                    error::closed, ec);
                return bytes_written;
            }
        }
        if(impl.rd_fh.len == 0 && ! impl.rd_fh.fin)
        {
            // Empty non-final frame
            goto loop;
        }
        impl.rd_done = false;
    }
    else
    {
        ec = {};
    }
    if(! impl.rd_deflated())
    {
        if(impl.rd_remain > 0)
        {
            if(impl.rd_buf.size() == 0 && impl.rd_buf.max_size() >
                (std::min)(clamp(impl.rd_remain),
                    buffer_bytes(buffers)))
            {
                // Fill the read buffer first, otherwise we
                // get fewer bytes at the cost of one I/O.
                impl.rd_buf.commit(impl.stream().read_some(
                    impl.rd_buf.prepare(read_size(impl.rd_buf,
                        impl.rd_buf.max_size())), ec));
                if(impl.check_stop_now(ec))
                    return bytes_written;
                if(impl.rd_fh.mask)
                    detail::mask_inplace(
                        buffers_prefix(clamp(impl.rd_remain),
                            impl.rd_buf.data()), impl.rd_key);
            }
            if(impl.rd_buf.size() > 0)
            {
                // Copy from the read buffer.
                // The mask was already applied.
                auto const bytes_transferred = net::buffer_copy(
                    buffers, impl.rd_buf.data(),
                        clamp(impl.rd_remain));
                auto const mb = buffers_prefix(
                    bytes_transferred, buffers);
                impl.rd_remain -= bytes_transferred;
                if(impl.rd_op == detail::opcode::text)
                {
                    if(! impl.rd_utf8.write(mb) ||
                        (impl.rd_remain == 0 && impl.rd_fh.fin &&
                            ! impl.rd_utf8.finish()))
                    {
                        // _Fail the WebSocket Connection_
                        do_fail(close_code::bad_payload,
                            error::bad_frame_payload, ec);
                        return bytes_written;
                    }
                }
                bytes_written += bytes_transferred;
                impl.rd_size += bytes_transferred;
                impl.rd_buf.consume(bytes_transferred);
            }
            else
            {
                // Read into caller's buffer
                BHO_ASSERT(impl.rd_remain > 0);
                BHO_ASSERT(buffer_bytes(buffers) > 0);
                BHO_ASSERT(buffer_bytes(buffers_prefix(
                    clamp(impl.rd_remain), buffers)) > 0);
                auto const bytes_transferred =
                    impl.stream().read_some(buffers_prefix(
                        clamp(impl.rd_remain), buffers), ec);
                // VFALCO What if some bytes were written?
                if(impl.check_stop_now(ec))
                    return bytes_written;
                BHO_ASSERT(bytes_transferred > 0);
                auto const mb = buffers_prefix(
                    bytes_transferred, buffers);
                impl.rd_remain -= bytes_transferred;
                if(impl.rd_fh.mask)
                    detail::mask_inplace(mb, impl.rd_key);
                if(impl.rd_op == detail::opcode::text)
                {
                    if(! impl.rd_utf8.write(mb) ||
                        (impl.rd_remain == 0 && impl.rd_fh.fin &&
                            ! impl.rd_utf8.finish()))
                    {
                        // _Fail the WebSocket Connection_
                        do_fail(close_code::bad_payload,
                            error::bad_frame_payload, ec);
                        return bytes_written;
                    }
                }
                bytes_written += bytes_transferred;
                impl.rd_size += bytes_transferred;
            }
        }
        BHO_ASSERT( ! impl.rd_done );
        if( impl.rd_remain == 0 && impl.rd_fh.fin )
            impl.rd_done = true;
    }
    else
    {
        // Read compressed message frame payload:
        // inflate even if rd_fh_.len == 0, otherwise we
        // never emit the end-of-stream deflate block.
        //
        bool did_read = false;
        buffers_suffix<MutableBufferSequence> cb(buffers);
        while(buffer_bytes(cb) > 0)
        {
            zlib::z_params zs;
            {
                auto const out = beast::buffers_front(cb);
                zs.next_out = out.data();
                zs.avail_out = out.size();
                BHO_ASSERT(zs.avail_out > 0);
            }
            // boolean to track the end of the message.
            bool fin = false;
            if(impl.rd_remain > 0)
            {
                if(impl.rd_buf.size() > 0)
                {
                    // use what's there
                    auto const in = buffers_prefix(
                        clamp(impl.rd_remain), beast::buffers_front(
                            impl.rd_buf.data()));
                    zs.avail_in = in.size();
                    zs.next_in = in.data();
                }
                else if(! did_read)
                {
                    // read new
                    auto const bytes_transferred =
                        impl.stream().read_some(
                            impl.rd_buf.prepare(read_size(
                                impl.rd_buf, impl.rd_buf.max_size())),
                            ec);
                    if(impl.check_stop_now(ec))
                        return bytes_written;
                    BHO_ASSERT(bytes_transferred > 0);
                    impl.rd_buf.commit(bytes_transferred);
                    if(impl.rd_fh.mask)
                        detail::mask_inplace(
                            buffers_prefix(clamp(impl.rd_remain),
                                impl.rd_buf.data()), impl.rd_key);
                    auto const in = buffers_prefix(
                        clamp(impl.rd_remain), buffers_front(
                            impl.rd_buf.data()));
                    zs.avail_in = in.size();
                    zs.next_in = in.data();
                    did_read = true;
                }
                else
                {
                    break;
                }
            }
            else if(impl.rd_fh.fin)
            {
                // append the empty block codes
                static std::uint8_t constexpr
                    empty_block[4] = { 0x00, 0x00, 0xff, 0xff };
                zs.next_in = empty_block;
                zs.avail_in = sizeof(empty_block);
                fin = true;
            }
            else
            {
                break;
            }
            impl.inflate(zs, zlib::Flush::sync, ec);
            if(impl.check_stop_now(ec))
                return bytes_written;
            if (fin && zs.total_out == 0) {
                impl.do_context_takeover_read(impl.role);
                impl.rd_done = true;
                break;
            }
            if(impl.rd_msg_max && beast::detail::sum_exceeds(
                impl.rd_size, zs.total_out, impl.rd_msg_max))
            {
                do_fail(close_code::too_big,
                    error::message_too_big, ec);
                return bytes_written;
            }
            cb.consume(zs.total_out);
            impl.rd_size += zs.total_out;
            if (! fin) {
                impl.rd_remain -= zs.total_in;
                impl.rd_buf.consume(zs.total_in);
            }
            bytes_written += zs.total_out;
        }
        if(impl.rd_op == detail::opcode::text)
        {
            // check utf8
            if(! impl.rd_utf8.write(beast::buffers_prefix(
                bytes_written, buffers)) || (
                    impl.rd_done && ! impl.rd_utf8.finish()))
            {
                // _Fail the WebSocket Connection_
                do_fail(close_code::bad_payload,
                    error::bad_frame_payload, ec);
                return bytes_written;
            }
        }
    }
    return bytes_written;
}

template<class NextLayer, bool deflateSupported>
template<class MutableBufferSequence, BHO_BEAST_ASYNC_TPARAM2 ReadHandler>
BHO_BEAST_ASYNC_RESULT2(ReadHandler)
stream<NextLayer, deflateSupported>::
async_read_some(
    MutableBufferSequence const& buffers,
    ReadHandler&& handler)
{
    static_assert(is_async_stream<next_layer_type>::value,
        "AsyncStream type requirements not met");
    static_assert(net::is_mutable_buffer_sequence<
            MutableBufferSequence>::value,
        "MutableBufferSequence type requirements not met");
    return net::async_initiate<
        ReadHandler,
        void(error_code, std::size_t)>(
            run_read_some_op{},
            handler,
            impl_,
            buffers);
}

} // websocket
} // beast
} // bho

#endif