//
// Copyright (c) 2016-2017 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 BOOST_BEAST_WEBSOCKET_IMPL_WRITE_IPP
#define BOOST_BEAST_WEBSOCKET_IMPL_WRITE_IPP

#include <boost/beast/core/bind_handler.hpp>
#include <boost/beast/core/buffer_cat.hpp>
#include <boost/beast/core/buffer_prefix.hpp>
#include <boost/beast/core/consuming_buffers.hpp>
#include <boost/beast/core/handler_ptr.hpp>
#include <boost/beast/core/flat_static_buffer.hpp>
#include <boost/beast/core/type_traits.hpp>
#include <boost/beast/core/detail/clamp.hpp>
#include <boost/beast/core/detail/config.hpp>
#include <boost/beast/websocket/detail/frame.hpp>
#include <boost/asio/handler_alloc_hook.hpp>
#include <boost/asio/handler_continuation_hook.hpp>
#include <boost/asio/handler_invoke_hook.hpp>
#include <boost/assert.hpp>
#include <boost/config.hpp>
#include <boost/throw_exception.hpp>
#include <algorithm>
#include <memory>

namespace boost {
namespace beast {
namespace websocket {

template<class NextLayer>
template<class Buffers, class Handler>
class stream<NextLayer>::write_some_op
{
    struct data : op
    {
        bool cont;
        stream<NextLayer>& ws;
        consuming_buffers<Buffers> cb;
        bool fin;
        detail::frame_header fh;
        detail::fh_streambuf fh_buf;
        detail::prepared_key key;
        std::uint64_t remain;
        int step = 0;
        int entry_state;
        token tok;

        data(Handler& handler, stream<NextLayer>& ws_,
                bool fin_, Buffers const& bs)
            : ws(ws_)
            , cb(bs)
            , fin(fin_)
            , tok(ws.t_.unique())
        {
            using boost::asio::asio_handler_is_continuation;
            cont = asio_handler_is_continuation(std::addressof(handler));
        }
    };

    handler_ptr<data, Handler> d_;

public:
    write_some_op(write_some_op&&) = default;
    write_some_op(write_some_op const&) = default;

    template<class DeducedHandler, class... Args>
    write_some_op(DeducedHandler&& h,
            stream<NextLayer>& ws, Args&&... args)
        : d_(std::forward<DeducedHandler>(h),
            ws, std::forward<Args>(args)...)
    {
    }

    void operator()()
    {
        (*this)({}, 0, true);
    }

    void operator()(error_code ec,
        std::size_t bytes_transferred,
            bool again = true);

    friend
    void* asio_handler_allocate(
        std::size_t size, write_some_op* op)
    {
        using boost::asio::asio_handler_allocate;
        return asio_handler_allocate(
            size, std::addressof(op->d_.handler()));
    }

    friend
    void asio_handler_deallocate(
        void* p, std::size_t size, write_some_op* op)
    {
        using boost::asio::asio_handler_deallocate;
        asio_handler_deallocate(
            p, size, std::addressof(op->d_.handler()));
    }

    friend
    bool asio_handler_is_continuation(write_some_op* op)
    {
        return op->d_->cont;
    }

    template<class Function>
    friend
    void asio_handler_invoke(Function&& f, write_some_op* op)
    {
        using boost::asio::asio_handler_invoke;
        asio_handler_invoke(
            f, std::addressof(op->d_.handler()));
    }
};

template<class NextLayer>
template<class Buffers, class Handler>
void
stream<NextLayer>::
write_some_op<Buffers, Handler>::
operator()(error_code ec,
    std::size_t bytes_transferred, bool again)
{
    using beast::detail::clamp;
    using boost::asio::buffer;
    using boost::asio::buffer_copy;
    using boost::asio::buffer_size;
    enum
    {
        do_init = 0,
        do_nomask_nofrag = 20,
        do_nomask_frag = 30,
        do_mask_nofrag = 40,
        do_mask_frag = 50,
        do_deflate = 60,
        do_maybe_suspend = 80,
        do_upcall = 99
    };
    auto& d = *d_;
    d.cont = d.cont || again;
    if(ec)
    {
        BOOST_ASSERT(d.ws.wr_block_ == d.tok);
        d.ws.failed_ = true;
        goto upcall;
    }
loop:
    switch(d.step)
    {
    case do_init:
        if(! d.ws.wr_.cont)
        {
            d.ws.wr_begin();
            d.fh.rsv1 = d.ws.wr_.compress;
        }
        else
        {
            d.fh.rsv1 = false;
        }
        d.fh.rsv2 = false;
        d.fh.rsv3 = false;
        d.fh.op = d.ws.wr_.cont ?
            detail::opcode::cont : d.ws.wr_opcode_;
        d.fh.mask =
            d.ws.role_ == role_type::client;

        // entry_state determines which algorithm
        // we will use to send. If we suspend, we
        // will transition to entry_state + 1 on
        // the resume.
        if(d.ws.wr_.compress)
        {
            d.entry_state = do_deflate;
        }
        else if(! d.fh.mask)
        {
            if(! d.ws.wr_.autofrag)
            {
                d.entry_state = do_nomask_nofrag;
            }
            else
            {
                BOOST_ASSERT(d.ws.wr_.buf_size != 0);
                d.remain = buffer_size(d.cb);
                if(d.remain > d.ws.wr_.buf_size)
                    d.entry_state = do_nomask_frag;
                else
                    d.entry_state = do_nomask_nofrag;
            }
        }
        else
        {
            if(! d.ws.wr_.autofrag)
            {
                d.entry_state = do_mask_nofrag;
            }
            else
            {
                BOOST_ASSERT(d.ws.wr_.buf_size != 0);
                d.remain = buffer_size(d.cb);
                if(d.remain > d.ws.wr_.buf_size)
                    d.entry_state = do_mask_frag;
                else
                    d.entry_state = do_mask_nofrag;
            }
        }
        d.step = do_maybe_suspend;
        goto loop;

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

    case do_nomask_nofrag:
        BOOST_ASSERT(d.ws.wr_block_ == d.tok);
        d.fh.fin = d.fin;
        d.fh.len = buffer_size(d.cb);
        detail::write<flat_static_buffer_base>(
            d.fh_buf, d.fh);
        d.ws.wr_.cont = ! d.fin;
        // Send frame
        d.step = do_upcall;
        return boost::asio::async_write(d.ws.stream_,
            buffer_cat(d.fh_buf.data(), d.cb),
                std::move(*this));

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

    go_nomask_frag:
    case do_nomask_frag:
    {
        BOOST_ASSERT(d.ws.wr_block_ == d.tok);
        auto const n = clamp(
            d.remain, d.ws.wr_.buf_size);
        d.remain -= n;
        d.fh.len = n;
        d.fh.fin = d.fin ? d.remain == 0 : false;
        detail::write<flat_static_buffer_base>(
            d.fh_buf, d.fh);
        d.ws.wr_.cont = ! d.fin;
        // Send frame
        d.step = d.remain == 0 ?
            do_upcall : do_nomask_frag + 1;
        return boost::asio::async_write(
            d.ws.stream_, buffer_cat(
                d.fh_buf.data(), buffer_prefix(
                    n, d.cb)), std::move(*this));
    }

    case do_nomask_frag + 1:
        BOOST_ASSERT(d.ws.wr_block_ == d.tok);
        d.ws.wr_block_.reset();
        d.cb.consume(
            bytes_transferred - d.fh_buf.size());
        d.fh_buf.consume(d.fh_buf.size());
        d.fh.op = detail::opcode::cont;
        // Allow outgoing control frames to
        // be sent in between message frames
        if( d.ws.close_op_.maybe_invoke() ||
            d.ws.rd_op_.maybe_invoke() ||
            d.ws.ping_op_.maybe_invoke())
        {
            d.step = do_maybe_suspend;
            return d.ws.get_io_service().post(
                std::move(*this));
        }
        d.ws.wr_block_ = d.tok;
        goto go_nomask_frag;

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

    case do_mask_nofrag:
    {
        BOOST_ASSERT(d.ws.wr_block_ == d.tok);
        d.remain = buffer_size(d.cb);
        d.fh.fin = d.fin;
        d.fh.len = d.remain;
        d.fh.key = d.ws.maskgen_();
        detail::prepare_key(d.key, d.fh.key);
        detail::write<flat_static_buffer_base>(
            d.fh_buf, d.fh);
        auto const n =
            clamp(d.remain, d.ws.wr_.buf_size);
        auto const b =
            buffer(d.ws.wr_.buf.get(), n);
        buffer_copy(b, d.cb);
        detail::mask_inplace(b, d.key);
        d.remain -= n;
        d.ws.wr_.cont = ! d.fin;
        // Send frame header and partial payload
        d.step = d.remain == 0 ?
            do_upcall : do_mask_nofrag + 1;
        return boost::asio::async_write(
            d.ws.stream_, buffer_cat(d.fh_buf.data(),
                b), std::move(*this));
    }

    case do_mask_nofrag + 1:
    {
        d.cb.consume(d.ws.wr_.buf_size);
        auto const n =
            clamp(d.remain, d.ws.wr_.buf_size);
        auto const b =
            buffer(d.ws.wr_.buf.get(), n);
        buffer_copy(b, d.cb);
        detail::mask_inplace(b, d.key);
        d.remain -= n;
        // Send partial payload
        if(d.remain == 0)
            d.step = do_upcall;
        return boost::asio::async_write(
            d.ws.stream_, b, std::move(*this));
    }

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

    go_mask_frag:
    case do_mask_frag:
    {
        BOOST_ASSERT(d.ws.wr_block_ == d.tok);
        auto const n = clamp(
            d.remain, d.ws.wr_.buf_size);
        d.remain -= n;
        d.fh.len = n;
        d.fh.key = d.ws.maskgen_();
        d.fh.fin = d.fin ? d.remain == 0 : false;
        detail::prepare_key(d.key, d.fh.key);
        auto const b = buffer(
            d.ws.wr_.buf.get(), n);
        buffer_copy(b, d.cb);
        detail::mask_inplace(b, d.key);
        detail::write<flat_static_buffer_base>(
            d.fh_buf, d.fh);
        d.ws.wr_.cont = ! d.fin;
        // Send frame
        d.step = d.remain == 0 ?
            do_upcall : do_mask_frag + 1;
        return boost::asio::async_write(
            d.ws.stream_, buffer_cat(
                d.fh_buf.data(), b),
                    std::move(*this));
    }

    case do_mask_frag + 1:
        BOOST_ASSERT(d.ws.wr_block_ == d.tok);
        d.ws.wr_block_.reset();
        d.cb.consume(
            bytes_transferred - d.fh_buf.size());
        d.fh_buf.consume(d.fh_buf.size());
        d.fh.op = detail::opcode::cont;
        // Allow outgoing control frames to
        // be sent in between message frames:
        if( d.ws.close_op_.maybe_invoke() ||
            d.ws.rd_op_.maybe_invoke() ||
            d.ws.ping_op_.maybe_invoke())
        {
            d.step = do_maybe_suspend;
            d.ws.get_io_service().post(
                std::move(*this));
            return;
        }
        d.ws.wr_block_ = d.tok;
        goto go_mask_frag;

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

    go_deflate:
    case do_deflate:
    {
        BOOST_ASSERT(d.ws.wr_block_ == d.tok);
        auto b = buffer(d.ws.wr_.buf.get(),
            d.ws.wr_.buf_size);
        auto const more = detail::deflate(
            d.ws.pmd_->zo, b, d.cb, d.fin, ec);
        d.ws.failed_ = !!ec;
        if(d.ws.failed_)
            goto upcall;
        auto const n = buffer_size(b);
        if(n == 0)
        {
            // The input was consumed, but there
            // is no output due to compression
            // latency.
            BOOST_ASSERT(! d.fin);
            BOOST_ASSERT(buffer_size(d.cb) == 0);

            // We can skip the dispatch if the
            // asynchronous initiation function is
            // not on call stack but its hard to
            // figure out so be safe and dispatch.
            d.step = do_upcall;
            d.ws.get_io_service().post(std::move(*this));
            return;
        }
        if(d.fh.mask)
        {
            d.fh.key = d.ws.maskgen_();
            detail::prepared_key key;
            detail::prepare_key(key, d.fh.key);
            detail::mask_inplace(b, key);
        }
        d.fh.fin = ! more;
        d.fh.len = n;
        detail::write<
            flat_static_buffer_base>(d.fh_buf, d.fh);
        d.ws.wr_.cont = ! d.fin;
        // Send frame
        d.step = more ?
            do_deflate + 1 : do_deflate + 2;
        boost::asio::async_write(d.ws.stream_,
            buffer_cat(d.fh_buf.data(), b),
                std::move(*this));
        return;
    }

    case do_deflate + 1:
        BOOST_ASSERT(d.ws.wr_block_ == d.tok);
        d.fh_buf.consume(d.fh_buf.size());
        d.ws.wr_block_.reset();
        d.fh.op = detail::opcode::cont;
        d.fh.rsv1 = false;
        // Allow outgoing control frames to
        // be sent in between message frames:
        if( d.ws.close_op_.maybe_invoke() ||
            d.ws.rd_op_.maybe_invoke() ||
            d.ws.ping_op_.maybe_invoke())
        {
            d.step = do_maybe_suspend;
            d.ws.get_io_service().post(
                std::move(*this));
            return;
        }
        d.ws.wr_block_ = d.tok;
        goto go_deflate;

    case do_deflate + 2:
        BOOST_ASSERT(d.ws.wr_block_ == d.tok);
        if(d.fh.fin && (
            (d.ws.role_ == role_type::client &&
                d.ws.pmd_config_.client_no_context_takeover) ||
            (d.ws.role_ == role_type::server &&
                d.ws.pmd_config_.server_no_context_takeover)))
            d.ws.pmd_->zo.reset();
        goto upcall;

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

    case do_maybe_suspend:
        if(d.ws.wr_block_)
        {
            // suspend
            BOOST_ASSERT(d.ws.wr_block_ != d.tok);
            d.step = do_maybe_suspend + 1;
            d.ws.wr_op_.emplace(std::move(*this));
            return;
        }
        d.ws.wr_block_ = d.tok;
        if(d.ws.failed_ || d.ws.wr_close_)
        {
            // call handler
            return d.ws.get_io_service().post(
                bind_handler(std::move(*this),
                    boost::asio::error::operation_aborted, 0));
        }
        d.step = d.entry_state;
        goto loop;

    case do_maybe_suspend + 1:
        BOOST_ASSERT(! d.ws.wr_block_);
        d.ws.wr_block_ = d.tok;
        d.step = do_maybe_suspend + 2;
        // The current context is safe but might not be
        // the same as the one for this operation (since
        // we are being called from a write operation).
        // Call post to make sure we are invoked the same
        // way as the final handler for this operation.
        d.ws.get_io_service().post(bind_handler(
            std::move(*this), ec, 0));
        return;

    case do_maybe_suspend + 2:
        BOOST_ASSERT(d.ws.wr_block_ == d.tok);
        if(d.ws.failed_ || d.ws.wr_close_)
        {
            // call handler
            ec = boost::asio::error::operation_aborted;
            goto upcall;
        }
        d.step = d.entry_state;
        goto loop;

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

    case do_upcall:
        goto upcall;
    }
upcall:
    if(d.ws.wr_block_ == d.tok)
        d.ws.wr_block_.reset();
    d.ws.close_op_.maybe_invoke() ||
        d.ws.rd_op_.maybe_invoke() ||
        d.ws.ping_op_.maybe_invoke();
    d_.invoke(ec);
}

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

template<class NextLayer>
template<class Buffers, class Handler>
class stream<NextLayer>::write_op
{
    struct data : op
    {
        int step = 0;
        stream<NextLayer>& ws;
        consuming_buffers<Buffers> cb;
        std::size_t remain;
        token tok;

        data(Handler&, stream<NextLayer>& ws_,
                Buffers const& bs)
            : ws(ws_)
            , cb(bs)
            , remain(boost::asio::buffer_size(cb))
            , tok(ws.t_.unique())
        {
        }
    };

    handler_ptr<data, Handler> d_;

public:
    write_op(write_op&&) = default;
    write_op(write_op const&) = default;

    template<class DeducedHandler, class... Args>
    explicit
    write_op(DeducedHandler&& h,
            stream<NextLayer>& ws, Args&&... args)
        : d_(std::forward<DeducedHandler>(h),
            ws, std::forward<Args>(args)...)
    {
    }

    void operator()(error_code ec);

    friend
    void* asio_handler_allocate(
        std::size_t size, write_op* op)
    {
        using boost::asio::asio_handler_allocate;
        return asio_handler_allocate(
            size, std::addressof(op->d_.handler()));
    }

    friend
    void asio_handler_deallocate(
        void* p, std::size_t size, write_op* op)
    {
        using boost::asio::asio_handler_deallocate;
        asio_handler_deallocate(
            p, size, std::addressof(op->d_.handler()));
    }

    friend
    bool asio_handler_is_continuation(write_op* op)
    {
        using boost::asio::asio_handler_is_continuation;
        return op->d_->step > 2 ||
            asio_handler_is_continuation(
                std::addressof(op->d_.handler()));
    }

    template<class Function>
    friend
    void asio_handler_invoke(Function&& f, write_op* op)
    {
        using boost::asio::asio_handler_invoke;
        asio_handler_invoke(
            f, std::addressof(op->d_.handler()));
    }
};

template<class NextLayer>
template<class Buffers, class Handler>
void
stream<NextLayer>::
write_op<Buffers, Handler>::
operator()(error_code ec)
{
    auto& d = *d_;
    switch(d.step)
    {
    case 2:
        d.step = 3;
        BOOST_BEAST_FALLTHROUGH;
    case 3:
    case 0:
    {
        auto const n = d.remain;
        d.remain -= n;
        auto const fin = d.remain <= 0;
        if(fin)
            d.step = d.step ? 4 : 1;
        else
            d.step = d.step ? 3 : 2;
        auto const pb = buffer_prefix(n, d.cb);
        d.cb.consume(n);
        return d.ws.async_write_some(
            fin, pb, std::move(*this));
    }

    case 1:
    case 4:
        break;
    }
    d_.invoke(ec);
}

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

template<class NextLayer>
template<class ConstBufferSequence>
void
stream<NextLayer>::
write_some(bool fin, ConstBufferSequence const& buffers)
{
    static_assert(is_sync_stream<next_layer_type>::value,
        "SyncStream requirements not met");
    static_assert(beast::is_const_buffer_sequence<
        ConstBufferSequence>::value,
            "ConstBufferSequence requirements not met");
    error_code ec;
    write_some(fin, buffers, ec);
    if(ec)
        BOOST_THROW_EXCEPTION(system_error{ec});
}

template<class NextLayer>
template<class ConstBufferSequence>
void
stream<NextLayer>::
write_some(bool fin,
    ConstBufferSequence const& buffers, error_code& ec)
{
    static_assert(is_sync_stream<next_layer_type>::value,
        "SyncStream requirements not met");
    static_assert(beast::is_const_buffer_sequence<
        ConstBufferSequence>::value,
            "ConstBufferSequence requirements not met");
    using beast::detail::clamp;
    using boost::asio::buffer;
    using boost::asio::buffer_copy;
    using boost::asio::buffer_size;
    detail::frame_header fh;
    if(! wr_.cont)
    {
        wr_begin();
        fh.rsv1 = wr_.compress;
    }
    else
    {
        fh.rsv1 = false;
    }
    fh.rsv2 = false;
    fh.rsv3 = false;
    fh.op = wr_.cont ?
        detail::opcode::cont : wr_opcode_;
    fh.mask = role_ == role_type::client;
    auto remain = buffer_size(buffers);
    if(wr_.compress)
    {
        consuming_buffers<
            ConstBufferSequence> cb{buffers};
        for(;;)
        {
            auto b = buffer(
                wr_.buf.get(), wr_.buf_size);
            auto const more = detail::deflate(
                pmd_->zo, b, cb, fin, ec);
            failed_ = !!ec;
            if(failed_)
                return;
            auto const n = buffer_size(b);
            if(n == 0)
            {
                // The input was consumed, but there
                // is no output due to compression
                // latency.
                BOOST_ASSERT(! fin);
                BOOST_ASSERT(buffer_size(cb) == 0);
                fh.fin = false;
                break;
            }
            if(fh.mask)
            {
                fh.key = maskgen_();
                detail::prepared_key key;
                detail::prepare_key(key, fh.key);
                detail::mask_inplace(b, key);
            }
            fh.fin = ! more;
            fh.len = n;
            detail::fh_streambuf fh_buf;
            detail::write<flat_static_buffer_base>(fh_buf, fh);
            wr_.cont = ! fin;
            boost::asio::write(stream_,
                buffer_cat(fh_buf.data(), b), ec);
            failed_ = !!ec;
            if(failed_)
                return;
            if(! more)
                break;
            fh.op = detail::opcode::cont;
            fh.rsv1 = false;
        }
        if(fh.fin && (
            (role_ == role_type::client &&
                pmd_config_.client_no_context_takeover) ||
            (role_ == role_type::server &&
                pmd_config_.server_no_context_takeover)))
            pmd_->zo.reset();
        return;
    }
    if(! fh.mask)
    {
        if(! wr_.autofrag)
        {
            // no mask, no autofrag
            fh.fin = fin;
            fh.len = remain;
            detail::fh_streambuf fh_buf;
            detail::write<flat_static_buffer_base>(fh_buf, fh);
            wr_.cont = ! fin;
            boost::asio::write(stream_,
                buffer_cat(fh_buf.data(), buffers), ec);
            failed_ = !!ec;
            if(failed_)
                return;
        }
        else
        {
            // no mask, autofrag
            BOOST_ASSERT(wr_.buf_size != 0);
            consuming_buffers<
                ConstBufferSequence> cb{buffers};
            for(;;)
            {
                auto const n = clamp(remain, wr_.buf_size);
                remain -= n;
                fh.len = n;
                fh.fin = fin ? remain == 0 : false;
                detail::fh_streambuf fh_buf;
                detail::write<flat_static_buffer_base>(fh_buf, fh);
                wr_.cont = ! fin;
                boost::asio::write(stream_,
                    buffer_cat(fh_buf.data(),
                        buffer_prefix(n, cb)), ec);
                failed_ = !!ec;
                if(failed_)
                    return;
                if(remain == 0)
                    break;
                fh.op = detail::opcode::cont;
                cb.consume(n);
            }
        }
        return;
    }
    if(! wr_.autofrag)
    {
        // mask, no autofrag
        fh.fin = fin;
        fh.len = remain;
        fh.key = maskgen_();
        detail::prepared_key key;
        detail::prepare_key(key, fh.key);
        detail::fh_streambuf fh_buf;
        detail::write<flat_static_buffer_base>(fh_buf, fh);
        consuming_buffers<
            ConstBufferSequence> cb{buffers};
        {
            auto const n = clamp(remain, wr_.buf_size);
            auto const b = buffer(wr_.buf.get(), n);
            buffer_copy(b, cb);
            cb.consume(n);
            remain -= n;
            detail::mask_inplace(b, key);
            wr_.cont = ! fin;
            boost::asio::write(stream_,
                buffer_cat(fh_buf.data(), b), ec);
            failed_ = !!ec;
            if(failed_)
                return;
        }
        while(remain > 0)
        {
            auto const n = clamp(remain, wr_.buf_size);
            auto const b = buffer(wr_.buf.get(), n);
            buffer_copy(b, cb);
            cb.consume(n);
            remain -= n;
            detail::mask_inplace(b, key);
            boost::asio::write(stream_, b, ec);
            failed_ = !!ec;
            if(failed_)
                return;
        }
        return;
    }
    {
        // mask, autofrag
        BOOST_ASSERT(wr_.buf_size != 0);
        consuming_buffers<
            ConstBufferSequence> cb{buffers};
        for(;;)
        {
            fh.key = maskgen_();
            detail::prepared_key key;
            detail::prepare_key(key, fh.key);
            auto const n = clamp(remain, wr_.buf_size);
            auto const b = buffer(wr_.buf.get(), n);
            buffer_copy(b, cb);
            detail::mask_inplace(b, key);
            fh.len = n;
            remain -= n;
            fh.fin = fin ? remain == 0 : false;
            wr_.cont = ! fh.fin;
            detail::fh_streambuf fh_buf;
            detail::write<flat_static_buffer_base>(fh_buf, fh);
            boost::asio::write(stream_,
                buffer_cat(fh_buf.data(), b), ec);
            failed_ = !!ec;
            if(failed_)
                return;
            if(remain == 0)
                break;
            fh.op = detail::opcode::cont;
            cb.consume(n);
        }
        return;
    }
}

template<class NextLayer>
template<class ConstBufferSequence, class WriteHandler>
async_return_type<
    WriteHandler, void(error_code)>
stream<NextLayer>::
async_write_some(bool fin,
    ConstBufferSequence const& bs, WriteHandler&& handler)
{
    static_assert(is_async_stream<next_layer_type>::value,
        "AsyncStream requirements not met");
    static_assert(beast::is_const_buffer_sequence<
        ConstBufferSequence>::value,
            "ConstBufferSequence requirements not met");
    async_completion<WriteHandler,
        void(error_code)> init{handler};
    write_some_op<ConstBufferSequence, handler_type<
        WriteHandler, void(error_code)>>{init.completion_handler,
            *this, fin, bs}({}, 0, false);
    return init.result.get();
}

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

template<class NextLayer>
template<class ConstBufferSequence>
void
stream<NextLayer>::
write(ConstBufferSequence const& buffers)
{
    static_assert(is_sync_stream<next_layer_type>::value,
        "SyncStream requirements not met");
    static_assert(beast::is_const_buffer_sequence<
        ConstBufferSequence>::value,
            "ConstBufferSequence requirements not met");
    error_code ec;
    write(buffers, ec);
    if(ec)
        BOOST_THROW_EXCEPTION(system_error{ec});
}

template<class NextLayer>
template<class ConstBufferSequence>
void
stream<NextLayer>::
write(ConstBufferSequence const& buffers, error_code& ec)
{
    static_assert(is_sync_stream<next_layer_type>::value,
        "SyncStream requirements not met");
    static_assert(beast::is_const_buffer_sequence<
        ConstBufferSequence>::value,
            "ConstBufferSequence requirements not met");
    write_some(true, buffers, ec);
}

template<class NextLayer>
template<class ConstBufferSequence, class WriteHandler>
async_return_type<
    WriteHandler, void(error_code)>
stream<NextLayer>::
async_write(
    ConstBufferSequence const& bs, WriteHandler&& handler)
{
    static_assert(is_async_stream<next_layer_type>::value,
        "AsyncStream requirements not met");
    static_assert(beast::is_const_buffer_sequence<
        ConstBufferSequence>::value,
            "ConstBufferSequence requirements not met");
    async_completion<WriteHandler,
        void(error_code)> init{handler};
    write_op<ConstBufferSequence, handler_type<
        WriteHandler, void(error_code)>>{
            init.completion_handler, *this, bs}(
                error_code{});
    return init.result.get();
}

} // websocket
} // beast
} // boost

#endif