Add websocket::stream timeouts

include/boost/beast/_experimental/test/fail_count.hpp

@@ -10,6 +10,7 @@
 #ifndef BOOST_BEAST_TEST_FAIL_COUNT_HPP
 #define BOOST_BEAST_TEST_FAIL_COUNT_HPP
 
+#include <boost/beast/core/detail/config.hpp>
 #include <boost/beast/core/error.hpp>
 #include <boost/beast/_experimental/test/error.hpp>
 #include <cstdlib>

include/boost/beast/_experimental/test/handler.hpp

@@ -62,7 +62,7 @@ public:
 
     template<class... Args>
     void
-    operator()(error_code ec, Args&&... args)
+    operator()(error_code ec, Args&&...)
     {
         BEAST_EXPECT(! pass_); // can't call twice
         BEAST_EXPECTS(! ec_ || ec == *ec_,

include/boost/beast/_experimental/test/impl/stream.hpp

@@ -629,6 +629,12 @@ connect(stream& to)
     return from;
 }
 
+void
+connect(stream& s1, stream& s2)
+{
+    s1.connect(s2);
+}
+
 template<class Arg1, class... ArgN>
 stream
 connect(stream& to, Arg1&& arg1, ArgN&&... argn)

include/boost/beast/_experimental/test/stream.hpp

@@ -10,6 +10,7 @@
 #ifndef BOOST_BEAST_TEST_STREAM_HPP
 #define BOOST_BEAST_TEST_STREAM_HPP
 
+#include <boost/beast/core/detail/config.hpp>
 #include <boost/beast/core/bind_handler.hpp>
 #include <boost/beast/core/flat_buffer.hpp>
 #include <boost/beast/core/string.hpp>
@@ -512,6 +513,10 @@ BOOST_BEAST_DECL
 stream
 connect(stream& to);
 
+BOOST_BEAST_DECL
+void
+connect(stream& s1, stream& s2);
+
 template<class Arg1, class... ArgN>
 stream
 connect(stream& to, Arg1&& arg1, ArgN&&... argn);

include/boost/beast/_experimental/test/tcp.hpp

@@ -0,0 +1,110 @@
+//
+// Copyright (c) 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 BOOST_BEAST_TEST_TCP_HPP
+#define BOOST_BEAST_TEST_TCP_HPP
+
+#include <boost/beast/core/detail/config.hpp>
+#include <boost/beast/_experimental/unit_test/suite.hpp>
+#include <boost/beast/_experimental/test/handler.hpp>
+#include <boost/asio/ip/tcp.hpp>
+#include <chrono>
+
+namespace boost {
+namespace beast {
+namespace test {
+
+/** Run an I/O context.
+    
+    This function runs and dispatches handlers on the specified
+    I/O context, until one of the following conditions is true:
+        
+    @li The I/O context runs out of work.
+*/
+inline
+void
+run(net::io_context& ioc)
+{
+    ioc.run();
+    ioc.restart();
+}
+
+/** Run an I/O context for a certain amount of time.
+    
+    This function runs and dispatches handlers on the specified
+    I/O context, until one of the following conditions is true:
+        
+    @li The I/O context runs out of work.
+
+    @li No completions occur and the specified amount of time has elapsed.
+
+    @param elapsed The maximum amount of time to run for.
+*/
+template<class Rep, class Period>
+void
+run_for(
+    net::io_context& ioc,
+    std::chrono::duration<Rep, Period> elapsed)
+{
+    ioc.run_for(elapsed);
+    ioc.restart();
+}
+
+/** Connect two TCP/IP sockets together.
+*/
+inline
+bool
+connect(
+    net::ip::tcp::socket& s1,
+    net::ip::tcp::socket& s2)
+
+{
+    // Sockets must use the same I/O context
+    BOOST_ASSERT(
+        std::addressof(s1.get_executor().context()) ==
+        std::addressof(s2.get_executor().context()));
+    auto& ioc = s1.get_executor().context();
+    s1 = net::ip::tcp::socket(ioc);
+    s2 = net::ip::tcp::socket(ioc);
+    try
+    {
+        net::ip::tcp::acceptor a(
+            s1.get_executor().context());
+        auto ep = net::ip::tcp::endpoint(
+            net::ip::make_address_v4("127.0.0.1"), 0);
+        a.open(ep.protocol());
+        a.set_option(
+            net::socket_base::reuse_address(true));
+        a.bind(ep);
+        a.listen(0);
+        ep = a.local_endpoint();
+        a.async_accept(s2, test::success_handler());
+        s1.async_connect(ep, test::success_handler());
+        run(ioc);
+        if(! BEAST_EXPECT(
+            s1.remote_endpoint() == s2.local_endpoint()))
+            return false;
+        if(! BEAST_EXPECT(
+            s2.remote_endpoint() == s1.local_endpoint()))
+            return false;
+    }
+    catch(std::exception const& e)
+    {
+        beast::unit_test::suite::this_suite()->fail(
+            e.what(), __FILE__, __LINE__);
+        return false;
+    }
+    return true;
+}
+
+} // test
+} // beast
+} // boost
+
+#endif

include/boost/beast/core/async_op_base.hpp

@@ -347,8 +347,8 @@ public:
         }
         else
         {
-            wg1_.reset();
             h_(std::forward<Args>(args)...);
+            wg1_.reset();
         }
     }
 

include/boost/beast/core/basic_stream.hpp

@@ -18,6 +18,7 @@
 #include <boost/asio/basic_stream_socket.hpp>
 #include <boost/asio/connect.hpp>
 #include <boost/asio/executor.hpp>
+#include <boost/asio/is_executor.hpp>
 #include <boost/core/empty_value.hpp>
 #include <boost/config/workaround.hpp>
 #include <boost/optional.hpp>
@@ -195,6 +196,9 @@ class basic_stream
     : private detail::stream_base
 #endif
 {
+    static_assert(net::is_executor<Executor>::value,
+        "Executor requirements not met");
+
 // friend class template declaration in a class template is ignored
 // https://gcc.gnu.org/bugzilla/show_bug.cgi?id=88672
 #if BOOST_WORKAROUND(BOOST_GCC, > 0)

include/boost/beast/websocket/detail/hybi13.hpp

@@ -14,7 +14,7 @@
 #include <boost/beast/core/string.hpp>
 #include <boost/beast/core/detail/base64.hpp>
 #include <boost/beast/core/detail/sha1.hpp>
-#include <boost/beast/websocket/detail/stream_base.hpp>
+#include <boost/beast/websocket/detail/prng.hpp>
 #include <boost/assert.hpp>
 #include <array>
 #include <cstdint>
@@ -53,7 +53,8 @@ make_sec_ws_key(sec_ws_key_type& key)
 
 template<class = void>
 void
-make_sec_ws_accept(sec_ws_accept_type& accept,
+make_sec_ws_accept(
+    sec_ws_accept_type& accept,
     string_view key)
 {
     BOOST_ASSERT(key.size() <= sec_ws_key_type::max_size_n);

include/boost/beast/websocket/detail/impl_base.hpp

@@ -7,8 +7,8 @@
 // Official repository: https://github.com/boostorg/beast
 //
 
-#ifndef BOOST_BEAST_WEBSOCKET_DETAIL_STREAM_BASE_HPP
-#define BOOST_BEAST_WEBSOCKET_DETAIL_STREAM_BASE_HPP
+#ifndef BOOST_BEAST_WEBSOCKET_DETAIL_IMPL_BASE_HPP
+#define BOOST_BEAST_WEBSOCKET_DETAIL_IMPL_BASE_HPP
 
 #include <boost/beast/core/buffer_size.hpp>
 #include <boost/beast/http/empty_body.hpp>
@@ -17,9 +17,7 @@
 #include <boost/beast/websocket/option.hpp>
 #include <boost/beast/websocket/role.hpp>
 #include <boost/beast/websocket/detail/frame.hpp>
-#include <boost/beast/websocket/detail/prng.hpp>
 #include <boost/beast/websocket/detail/pmd_extension.hpp>
-#include <boost/beast/websocket/detail/prng.hpp>
 #include <boost/beast/zlib/deflate_stream.hpp>
 #include <boost/beast/zlib/inflate_stream.hpp>
 #include <boost/beast/core/buffers_suffix.hpp>
@@ -489,31 +487,6 @@ struct impl_base<false>
     }
 };
 
-//------------------------------------------------------------------------------
-
-struct stream_base
-{
-protected:
-    enum class status
-    {
-        open,
-        closing,
-        closed,
-        failed
-    };
-
-    std::uint32_t
-    create_mask()
-    {
-        auto g = make_prng(secure_prng_);
-        for(;;)
-            if(auto key = g())
-                return key;
-    }
-
-    bool secure_prng_ = true;
-};
-
 } // detail
 } // websocket
 } // beast

include/boost/beast/websocket/impl/accept.hpp

@@ -11,6 +11,7 @@
 #define BOOST_BEAST_WEBSOCKET_IMPL_ACCEPT_IPP
 
 #include <boost/beast/core/buffer_size.hpp>
+#include <boost/beast/websocket/impl/stream_impl.hpp>
 #include <boost/beast/websocket/detail/type_traits.hpp>
 #include <boost/beast/http/empty_body.hpp>
 #include <boost/beast/http/parser.hpp>
@@ -42,7 +43,7 @@ class stream<NextLayer, deflateSupported>::response_op
         Handler, beast::executor_type<stream>>
     , public net::coroutine
 {
-    stream<NextLayer, deflateSupported>& ws_;
+    boost::weak_ptr<impl_type> wp_;
     error_code result_; // must come before res_
     response_type& res_;
 
@@ -53,37 +54,53 @@ public:
         class Decorator>
     response_op(
         Handler_&& h,
-        stream<NextLayer, deflateSupported>& ws,
-        http::request<Body, http::basic_fields<Allocator>> const& req,
-        Decorator const& decorator)
-        : stable_async_op_base<
-            Handler, beast::executor_type<stream>>(
-                std::forward<Handler_>(h), ws.get_executor())
-        , ws_(ws)
+        boost::shared_ptr<impl_type> const& sp,
+        http::request<Body,
+            http::basic_fields<Allocator>> const& req,
+        Decorator const& decorator,
+        bool cont = false)
+        : stable_async_op_base<Handler,
+            beast::executor_type<stream>>(
+                std::forward<Handler_>(h),
+                    sp->stream.get_executor())
+        , wp_(sp)
         , res_(beast::allocate_stable<response_type>(*this, 
-            ws.build_response(req, decorator, result_)))
+            sp->build_response(req, decorator, result_)))
     {
+        (*this)({}, 0, cont);
     }
 
     void operator()(
         error_code ec = {},
-        std::size_t bytes_transferred = 0)
+        std::size_t bytes_transferred = 0,
+        bool cont = true)
     {
         boost::ignore_unused(bytes_transferred);
+        auto sp = wp_.lock();
+        if(! sp)
+            return this->invoke(cont,
+                net::error::operation_aborted);
+        auto& impl = *sp;
         BOOST_ASIO_CORO_REENTER(*this)
         {
+            impl.change_status(status::handshake);
+            impl.update_timer(this->get_executor());
+
             // Send response
             BOOST_ASIO_CORO_YIELD
             http::async_write(
-                ws_.next_layer(), res_, std::move(*this));
+                impl.stream, res_, std::move(*this));
+            if(impl.check_stop_now(ec))
+                goto upcall;
             if(! ec)
                 ec = result_;
             if(! ec)
             {
-                ws_.impl_->do_pmd_config(res_);
-                ws_.impl_->open(role_type::server);
+                impl.do_pmd_config(res_);
+                impl.open(role_type::server);
             }
-            this->invoke_now(ec);
+        upcall:
+            this->invoke(cont, ec);
         }
     }
 };
@@ -99,90 +116,144 @@ class stream<NextLayer, deflateSupported>::accept_op
         Handler, beast::executor_type<stream>>
     , public net::coroutine
 {
-    stream<NextLayer, deflateSupported>& ws_;
+    boost::weak_ptr<impl_type> wp_;
     http::request_parser<http::empty_body>& p_;
     Decorator d_;
 
 public:
-    template<class Handler_>
+    template<class Handler_, class Buffers>
     accept_op(
         Handler_&& h,
-        stream<NextLayer, deflateSupported>& ws,
+        boost::shared_ptr<impl_type> const& sp,
+        Buffers const& buffers,
         Decorator const& decorator)
-        : stable_async_op_base<
-            Handler, beast::executor_type<stream>>(
-                std::forward<Handler_>(h), ws.get_executor())
-        , ws_(ws)
+        : stable_async_op_base<Handler,
+            beast::executor_type<stream>>(
+                std::forward<Handler_>(h),
+                    sp->stream.get_executor())
+        , wp_(sp)
         , p_(beast::allocate_stable<
             http::request_parser<http::empty_body>>(*this))
         , d_(decorator)
     {
-    }
-
-    template<class Buffers>
-    void run(Buffers const& buffers)
-    {
+        auto& impl = *sp;
         error_code ec;
-        auto const mb = beast::detail::dynamic_buffer_prepare(
-            ws_.impl_->rd_buf, buffer_size(buffers), ec,
-                error::buffer_overflow);
-        if(ec)
-            return (*this)(ec);
-        ws_.impl_->rd_buf.commit(net::buffer_copy(*mb, buffers));
+        auto const mb =
+            beast::detail::dynamic_buffer_prepare(
+            impl.rd_buf, buffer_size(buffers),
+                ec, error::buffer_overflow);
+        if(! ec)
+            impl.rd_buf.commit(
+                net::buffer_copy(*mb, buffers));
         (*this)(ec);
     }
 
     void operator()(
         error_code ec = {},
-        std::size_t bytes_transferred = 0)
+        std::size_t bytes_transferred = 0,
+        bool cont = true)
     {
         boost::ignore_unused(bytes_transferred);
-
+        auto sp = wp_.lock();
+        if(! sp)
+            return this->invoke(cont,
+                net::error::operation_aborted);
+        auto& impl = *sp;
         BOOST_ASIO_CORO_REENTER(*this)
         {
+            impl.change_status(status::handshake);
+            impl.update_timer(this->get_executor());
+
+            // The constructor could have set ec
             if(ec)
+                goto upcall;
+
+            BOOST_ASIO_CORO_YIELD
+            http::async_read(impl.stream,
+                impl.rd_buf, p_, std::move(*this));
+            if(ec == http::error::end_of_stream)
+                ec = error::closed;
+            if(impl.check_stop_now(ec))
+                goto upcall;
+            
             {
-                BOOST_ASIO_CORO_YIELD
-                net::post(
-                    ws_.get_executor(),
-                    beast::bind_front_handler(std::move(*this), ec));
+                // Arguments from our state must be
+                // moved to the stack before releasing
+                // the handler.
+                auto const req = p_.release();
+                auto const decorator = d_;
+                response_op<Handler>(
+                    this->release_handler(),
+                        sp, req, decorator, true);
+                return;
             }
-            else
-            {
-                BOOST_ASIO_CORO_YIELD
-                http::async_read(
-                    ws_.next_layer(), ws_.impl_->rd_buf,
-                        p_, std::move(*this));
-                if(ec == http::error::end_of_stream)
-                    ec = error::closed;
-                if(! ec)
-                {
-                    // Arguments from our state must be
-                    // moved to the stack before releasing
-                    // the handler.
-                    auto& ws = ws_;
-                    auto const req = p_.release();
-                    auto const decorator = d_;
-                #if 1
-                    return response_op<Handler>{
-                        this->release_handler(),
-                            ws, req, decorator}(ec);
-                #else
-                    // VFALCO This *should* work but breaks
-                    //        coroutine invariants in the unit test.
-                    //        Also it calls reset() when it shouldn't.
-                    return ws.async_accept_ex(
-                        req, decorator, this->release_handler());
-                #endif
-                }
-            }
-            this->invoke_now(ec);
+
+        upcall:
+            this->invoke(cont, ec);
         }
     }
 };
 
 //------------------------------------------------------------------------------
 
+template<class NextLayer, bool deflateSupported>
+template<class Body, class Allocator,
+    class Decorator>
+void
+stream<NextLayer, deflateSupported>::
+do_accept(
+    http::request<Body,
+        http::basic_fields<Allocator>> const& req,
+    Decorator const& decorator,
+    error_code& ec)
+{
+    impl_->change_status(status::handshake);
+
+    error_code result;
+    auto const res = impl_->build_response(req, decorator, result);
+    http::write(impl_->stream, res, ec);
+    if(ec)
+        return;
+    ec = result;
+    if(ec)
+    {
+        // VFALCO TODO Respect keep alive setting, perform
+        //             teardown if Connection: close.
+        return;
+    }
+    impl_->do_pmd_config(res);
+    impl_->open(role_type::server);
+}
+
+template<class NextLayer, bool deflateSupported>
+template<class Buffers, class Decorator>
+void
+stream<NextLayer, deflateSupported>::
+do_accept(
+    Buffers const& buffers,
+    Decorator const& decorator,
+    error_code& ec)
+{
+    impl_->reset();
+    auto const mb =
+        beast::detail::dynamic_buffer_prepare(
+        impl_->rd_buf, buffer_size(buffers), ec,
+            error::buffer_overflow);
+    if(ec)
+        return;
+    impl_->rd_buf.commit(net::buffer_copy(*mb, buffers));
+
+    http::request_parser<http::empty_body> p;
+    http::read(next_layer(), impl_->rd_buf, p, ec);
+    if(ec == http::error::end_of_stream)
+        ec = error::closed;
+    if(ec)
+        return;
+    do_accept(p.get(), decorator, ec);
+}
+
+//------------------------------------------------------------------------------
+
 template<class NextLayer, bool deflateSupported>
 void
 stream<NextLayer, deflateSupported>::
@@ -220,8 +291,9 @@ accept(error_code& ec)
 {
     static_assert(is_sync_stream<next_layer_type>::value,
         "SyncStream requirements not met");
-    impl_->reset();
-    do_accept(&default_decorate_res, ec);
+    do_accept(
+        net::const_buffer{},
+        &default_decorate_res, ec);
 }
 
 template<class NextLayer, bool deflateSupported>
@@ -235,8 +307,9 @@ accept_ex(ResponseDecorator const& decorator, error_code& ec)
     static_assert(detail::is_response_decorator<
         ResponseDecorator>::value,
             "ResponseDecorator requirements not met");
-    impl_->reset();
-    do_accept(decorator, ec);
+    do_accept(
+        net::const_buffer{},
+        decorator, ec);
 }
 
 template<class NextLayer, bool deflateSupported>
@@ -295,14 +368,7 @@ accept(
     static_assert(net::is_const_buffer_sequence<
         ConstBufferSequence>::value,
             "ConstBufferSequence requirements not met");
-    impl_->reset();
-    auto const mb = beast::detail::dynamic_buffer_prepare(
-        impl_->rd_buf, buffer_size(buffers), ec,
-            error::buffer_overflow);
-    if(ec)
-        return;
-    impl_->rd_buf.commit(net::buffer_copy(*mb, buffers));
-    do_accept(&default_decorate_res, ec);
+    do_accept(buffers, &default_decorate_res, ec);
 }
 
 template<class NextLayer, bool deflateSupported>
@@ -325,14 +391,7 @@ accept_ex(
     static_assert(net::is_const_buffer_sequence<
         ConstBufferSequence>::value,
             "ConstBufferSequence requirements not met");
-    impl_->reset();
-    auto const mb = beast::detail::dynamic_buffer_prepare(
-        impl_->rd_buf, buffer_size(buffers), ec,
-            error::buffer_overflow);
-    if(ec)
-        return;
-    impl_->rd_buf.commit(net::buffer_copy(*mb, buffers));
-    do_accept(decorator, ec);
+    do_accept(buffers, decorator, ec);
 }
 
 template<class NextLayer, bool deflateSupported>
@@ -428,10 +487,10 @@ async_accept(
     accept_op<
         decltype(&default_decorate_res),
         BOOST_ASIO_HANDLER_TYPE(
-            AcceptHandler, void(error_code))>{
+            AcceptHandler, void(error_code))>(
                 std::move(init.completion_handler),
-                *this,
-                &default_decorate_res}({});
+                impl_, net::const_buffer{},
+                &default_decorate_res);;
     return init.result.get();
 }
 
@@ -457,10 +516,10 @@ async_accept_ex(
     accept_op<
         ResponseDecorator,
         BOOST_ASIO_HANDLER_TYPE(
-            AcceptHandler, void(error_code))>{
+            AcceptHandler, void(error_code))>(
                 std::move(init.completion_handler),
-                *this,
-                decorator}({});
+                impl_, net::const_buffer{},
+                decorator);
     return init.result.get();
 }
 
@@ -488,10 +547,9 @@ async_accept(
     accept_op<
         decltype(&default_decorate_res),
         BOOST_ASIO_HANDLER_TYPE(
-            AcceptHandler, void(error_code))>{
+            AcceptHandler, void(error_code))>(
                 std::move(init.completion_handler),
-                *this,
-                &default_decorate_res}.run(buffers);
+                    impl_, buffers, &default_decorate_res);
     return init.result.get();
 }
 
@@ -524,10 +582,9 @@ async_accept_ex(
     accept_op<
         ResponseDecorator,
         BOOST_ASIO_HANDLER_TYPE(
-            AcceptHandler, void(error_code))>{
+            AcceptHandler, void(error_code))>(
                 std::move(init.completion_handler),
-                *this,
-                decorator}.run(buffers);
+                    impl_, buffers, decorator);
     return init.result.get();
 }
 
@@ -549,11 +606,9 @@ async_accept(
     impl_->reset();
     response_op<
         BOOST_ASIO_HANDLER_TYPE(
-            AcceptHandler, void(error_code))>{
+            AcceptHandler, void(error_code))>(
                 std::move(init.completion_handler),
-                *this,
-                req,
-                &default_decorate_res}();
+                    impl_, req, &default_decorate_res);
     return init.result.get();
 }
 
@@ -580,60 +635,12 @@ async_accept_ex(
     impl_->reset();
     response_op<
         BOOST_ASIO_HANDLER_TYPE(
-            AcceptHandler, void(error_code))>{
+            AcceptHandler, void(error_code))>(
                 std::move(init.completion_handler),
-                *this,
-                req,
-                decorator}();
+                    impl_, req, decorator);
     return init.result.get();
 }
 
-//------------------------------------------------------------------------------
-
-template<class NextLayer, bool deflateSupported>
-template<class Decorator>
-void
-stream<NextLayer, deflateSupported>::
-do_accept(
-    Decorator const& decorator,
-    error_code& ec)
-{
-    http::request_parser<http::empty_body> p;
-    http::read(next_layer(), impl_->rd_buf, p, ec);
-    if(ec == http::error::end_of_stream)
-        ec = error::closed;
-    if(ec)
-        return;
-    do_accept(p.get(), decorator, ec);
-}
-
-template<class NextLayer, bool deflateSupported>
-template<class Body, class Allocator,
-    class Decorator>
-void
-stream<NextLayer, deflateSupported>::
-do_accept(
-    http::request<Body,
-        http::basic_fields<Allocator>> const& req,
-    Decorator const& decorator,
-    error_code& ec)
-{
-    error_code result;
-    auto const res = build_response(req, decorator, result);
-    http::write(impl_->stream, res, ec);
-    if(ec)
-        return;
-    ec = result;
-    if(ec)
-    {
-        // VFALCO TODO Respect keep alive setting, perform
-        //             teardown if Connection: close.
-        return;
-    }
-    impl_->do_pmd_config(res);
-    impl_->open(role_type::server);
-}
-
 } // websocket
 } // beast
 } // boost

include/boost/beast/websocket/impl/close.hpp

@@ -12,6 +12,7 @@
 
 #include <boost/beast/websocket/teardown.hpp>
 #include <boost/beast/websocket/detail/mask.hpp>
+#include <boost/beast/websocket/impl/stream_impl.hpp>
 #include <boost/beast/core/async_op_base.hpp>
 #include <boost/beast/core/flat_static_buffer.hpp>
 #include <boost/beast/core/stream_traits.hpp>
@@ -39,26 +40,28 @@ class stream<NextLayer, deflateSupported>::close_op
         Handler, beast::executor_type<stream>>
     , public net::coroutine
 {
-    stream<NextLayer, deflateSupported>& ws_;
+    boost::weak_ptr<impl_type> wp_;
     error_code ev_;
     detail::frame_buffer& fb_;
 
 public:
-    static constexpr int id = 4; // for soft_mutex
+    static constexpr int id = 5; // for soft_mutex
 
     template<class Handler_>
     close_op(
         Handler_&& h,
-        stream<NextLayer, deflateSupported>& ws,
+        boost::shared_ptr<impl_type> const& sp,
         close_reason const& cr)
-        : stable_async_op_base<
-            Handler, beast::executor_type<stream>>(
-                std::forward<Handler_>(h), ws.get_executor())
-        , ws_(ws)
-        , fb_(beast::allocate_stable<detail::frame_buffer>(*this))
+        : stable_async_op_base<Handler,
+            beast::executor_type<stream>>(
+                std::forward<Handler_>(h),
+                    sp->stream.get_executor())
+        , wp_(sp)
+        , fb_(beast::allocate_stable<
+            detail::frame_buffer>(*this))
     {
         // Serialize the close frame
-        ws.template write_close<
+        sp->template write_close<
             flat_static_buffer_base>(fb_, cr);
         (*this)({}, 0, false);
     }
@@ -70,14 +73,18 @@ public:
         bool cont = true)
     {
         using beast::detail::clamp;
-        auto& impl = *ws_.impl_;
+        auto sp = wp_.lock();
+        if(! sp)
+            return this->invoke(cont,
+                net::error::operation_aborted);
+        auto& impl = *sp;
         BOOST_ASIO_CORO_REENTER(*this)
         {
             // Acquire the write lock
             if(! impl.wr_block.try_lock(this))
             {
                 BOOST_ASIO_CORO_YIELD
-                impl.paused_close.emplace(std::move(*this));
+                impl.op_close.emplace(std::move(*this));
                 impl.wr_block.lock(this);
                 BOOST_ASIO_CORO_YIELD
                 net::post(std::move(*this));
@@ -93,6 +100,7 @@ public:
             // Send close frame
             impl.wr_close = true;
             impl.change_status(status::closing);
+            impl.update_timer(this->get_executor());
             BOOST_ASIO_CORO_YIELD
             net::async_write(impl.stream, fb_.data(),
                 beast::detail::bind_continuation(std::move(*this)));
@@ -111,7 +119,7 @@ public:
             if(! impl.rd_block.try_lock(this))
             {
                 BOOST_ASIO_CORO_YIELD
-                impl.paused_r_close.emplace(std::move(*this));
+                impl.op_r_close.emplace(std::move(*this));
                 impl.rd_block.lock(this);
                 BOOST_ASIO_CORO_YIELD
                 net::post(std::move(*this));
@@ -128,7 +136,7 @@ public:
             for(;;)
             {
                 // Read frame header
-                while(! ws_.parse_fh(
+                while(! impl.parse_fh(
                     impl.rd_fh, impl.rd_buf, ev_))
                 {
                     if(ev_)
@@ -212,10 +220,11 @@ public:
         upcall:
             impl.wr_block.unlock(this);
             impl.rd_block.try_unlock(this)
-                && impl.paused_r_rd.maybe_invoke();
-            impl.paused_rd.maybe_invoke()
-                || impl.paused_ping.maybe_invoke()
-                || impl.paused_wr.maybe_invoke();
+                && impl.op_r_rd.maybe_invoke();
+            impl.op_rd.maybe_invoke()
+                || impl.op_idle_ping.maybe_invoke()
+                || impl.op_ping.maybe_invoke()
+                || impl.op_wr.maybe_invoke();
             this->invoke(cont, ec);
         }
     }
@@ -259,7 +268,7 @@ close(close_reason const& cr, error_code& ec)
         impl.wr_close = true;
         impl.change_status(status::closing);
         detail::frame_buffer fb;
-        write_close<flat_static_buffer_base>(fb, cr);
+        impl.template write_close<flat_static_buffer_base>(fb, cr);
         net::write(impl.stream, fb.data(), ec);
         if(impl.check_stop_now(ec))
             return;
@@ -272,7 +281,7 @@ close(close_reason const& cr, error_code& ec)
     for(;;)
     {
         // Read frame header
-        while(! parse_fh(
+        while(! impl.parse_fh(
             impl.rd_fh, impl.rd_buf, ev))
         {
             if(ev)
@@ -355,7 +364,7 @@ async_close(close_reason const& cr, CloseHandler&& handler)
         CloseHandler, void(error_code));
     close_op<BOOST_ASIO_HANDLER_TYPE(
         CloseHandler, void(error_code))>(
-            std::move(init.completion_handler), *this, cr);
+            std::move(init.completion_handler), impl_, cr);
     return init.result.get();
 }
 

include/boost/beast/websocket/impl/handshake.hpp

@@ -10,6 +10,7 @@
 #ifndef BOOST_BEAST_WEBSOCKET_IMPL_HANDSHAKE_HPP
 #define BOOST_BEAST_WEBSOCKET_IMPL_HANDSHAKE_HPP
 
+#include <boost/beast/websocket/impl/stream_impl.hpp>
 #include <boost/beast/websocket/detail/type_traits.hpp>
 #include <boost/beast/http/empty_body.hpp>
 #include <boost/beast/http/message.hpp>
@@ -45,7 +46,7 @@ class stream<NextLayer, deflateSupported>::handshake_op
         response_type res;
     };
 
-    stream<NextLayer, deflateSupported>& ws_;
+    boost::weak_ptr<impl_type> wp_;
     detail::sec_ws_key_type key_;
     response_type* res_p_;
     data& d_;
@@ -54,54 +55,103 @@ public:
     template<class Handler_, class Decorator>
     handshake_op(
         Handler_&& h,
-        stream& ws,
+        boost::shared_ptr<impl_type> const& sp,
         response_type* res_p,
         string_view host, string_view target,
         Decorator const& decorator)
         : stable_async_op_base<Handler,
             beast::executor_type<stream>>(
-                std::forward<Handler_>(h), ws.get_executor())
-        , ws_(ws)
+                std::forward<Handler_>(h),
+                    sp->stream.get_executor())
+        , wp_(sp)
         , res_p_(res_p)
         , d_(beast::allocate_stable<data>(*this))
     {
-        d_.req = ws_.build_request(
+        d_.req = sp->build_request(
             key_, host, target, decorator);
-        ws_.impl_->reset(); // VFALCO I don't like this
+        sp->reset(); // VFALCO I don't like this
     }
 
     void
     operator()(
         error_code ec = {},
-        std::size_t bytes_used = 0)
+        std::size_t bytes_used = 0,
+        bool cont = true)
     {
         boost::ignore_unused(bytes_used);
+        auto sp = wp_.lock();
+        if(! sp)
+            return this->invoke(cont,
+                net::error::operation_aborted);
+        auto& impl = *sp;
         BOOST_ASIO_CORO_REENTER(*this)
         {
-            // Send HTTP Upgrade
-            ws_.impl_->do_pmd_config(d_.req);
+            impl.change_status(status::handshake);
+            impl.update_timer(this->get_executor());
+
+            // write HTTP request
+            impl.do_pmd_config(d_.req);
             BOOST_ASIO_CORO_YIELD
-            http::async_write(ws_.impl_->stream,
+            http::async_write(impl.stream,
                 d_.req, std::move(*this));
-            if(ec)
+            if(impl.check_stop_now(ec))
                 goto upcall;
 
-            // Read HTTP response
+            // read HTTP response
             BOOST_ASIO_CORO_YIELD
-            http::async_read(ws_.next_layer(),
-                ws_.impl_->rd_buf, d_.res,
+            http::async_read(impl.stream,
+                impl.rd_buf, d_.res,
                     std::move(*this));
-            if(ec)
+            if(impl.check_stop_now(ec))
                 goto upcall;
-            ws_.on_response(d_.res, key_, ec);
+
+            // success
+            impl.reset_idle();
+            impl.on_response(d_.res, key_, ec);
             if(res_p_)
                 swap(d_.res, *res_p_);
+
         upcall:
-            this->invoke_now(ec);
+            this->invoke(cont ,ec);
         }
     }
 };
 
+//------------------------------------------------------------------------------
+
+template<class NextLayer, bool deflateSupported>
+template<class RequestDecorator>
+void
+stream<NextLayer, deflateSupported>::
+do_handshake(
+    response_type* res_p,
+    string_view host,
+    string_view target,
+    RequestDecorator const& decorator,
+    error_code& ec)
+{
+    response_type res;
+    impl_->change_status(status::handshake);
+    impl_->reset();
+    detail::sec_ws_key_type key;
+    {
+        auto const req = impl_->build_request(
+            key, host, target, decorator);
+        this->impl_->do_pmd_config(req);
+        http::write(impl_->stream, req, ec);
+    }
+    if(ec)
+        return;
+    http::read(next_layer(), impl_->rd_buf, res, ec);
+    if(ec)
+        return;
+    impl_->on_response(res, key, ec);
+    if(res_p)
+        *res_p = std::move(res);
+}
+
+//------------------------------------------------------------------------------
+
 template<class NextLayer, bool deflateSupported>
 template<class HandshakeHandler>
 BOOST_ASIO_INITFN_RESULT_TYPE(
@@ -116,9 +166,10 @@ async_handshake(string_view host,
     BOOST_BEAST_HANDLER_INIT(
         HandshakeHandler, void(error_code));
     handshake_op<BOOST_ASIO_HANDLER_TYPE(
-        HandshakeHandler, void(error_code))>{
-            std::move(init.completion_handler), *this, nullptr, host,
-                target, &default_decorate_req}();
+        HandshakeHandler, void(error_code))>(
+            std::move(init.completion_handler),
+                impl_, nullptr, host, target,
+                    &default_decorate_req)();
     return init.result.get();
 }
 
@@ -137,9 +188,10 @@ async_handshake(response_type& res,
     BOOST_BEAST_HANDLER_INIT(
         HandshakeHandler, void(error_code));
     handshake_op<BOOST_ASIO_HANDLER_TYPE(
-        HandshakeHandler, void(error_code))>{
-            std::move(init.completion_handler), *this, &res, host,
-                target, &default_decorate_req}();
+        HandshakeHandler, void(error_code))>(
+            std::move(init.completion_handler),
+                impl_, &res, host, target,
+                    &default_decorate_req)();
     return init.result.get();
 }
 
@@ -161,9 +213,10 @@ async_handshake_ex(string_view host,
     BOOST_BEAST_HANDLER_INIT(
         HandshakeHandler, void(error_code));
     handshake_op<BOOST_ASIO_HANDLER_TYPE(
-        HandshakeHandler, void(error_code))>{
-            std::move(init.completion_handler), *this, nullptr, host,
-                target, decorator}();
+        HandshakeHandler, void(error_code))>(
+            std::move(init.completion_handler),
+                impl_, nullptr, host, target,
+                    decorator)();
     return init.result.get();
 }
 
@@ -186,9 +239,10 @@ async_handshake_ex(response_type& res,
     BOOST_BEAST_HANDLER_INIT(
         HandshakeHandler, void(error_code));
     handshake_op<BOOST_ASIO_HANDLER_TYPE(
-        HandshakeHandler, void(error_code))>{
-            std::move(init.completion_handler), *this, &res, host,
-                target, decorator}();
+        HandshakeHandler, void(error_code))>(
+            std::move(init.completion_handler),
+                impl_, &res, host, target,
+                    decorator)();
     return init.result.get();
 }
 
@@ -324,38 +378,6 @@ handshake_ex(response_type& res,
         host, target, decorator, ec);
 }
 
-//------------------------------------------------------------------------------
-
-template<class NextLayer, bool deflateSupported>
-template<class RequestDecorator>
-void
-stream<NextLayer, deflateSupported>::
-do_handshake(
-    response_type* res_p,
-    string_view host,
-    string_view target,
-    RequestDecorator const& decorator,
-    error_code& ec)
-{
-    response_type res;
-    impl_->reset();
-    detail::sec_ws_key_type key;
-    {
-        auto const req = build_request(
-            key, host, target, decorator);
-        this->impl_->do_pmd_config(req);
-        http::write(impl_->stream, req, ec);
-    }
-    if(ec)
-        return;
-    http::read(next_layer(), impl_->rd_buf, res, ec);
-    if(ec)
-        return;
-    on_response(res, key, ec);
-    if(res_p)
-        *res_p = std::move(res);
-}
-
 } // websocket
 } // beast
 } // boost

include/boost/beast/websocket/impl/ping.hpp

@@ -15,6 +15,7 @@
 #include <boost/beast/core/stream_traits.hpp>
 #include <boost/beast/core/detail/bind_continuation.hpp>
 #include <boost/beast/websocket/detail/frame.hpp>
+#include <boost/beast/websocket/impl/stream_impl.hpp>
 #include <boost/asio/coroutine.hpp>
 #include <boost/asio/post.hpp>
 #include <boost/throw_exception.hpp>
@@ -36,7 +37,7 @@ class stream<NextLayer, deflateSupported>::ping_op
         Handler, beast::executor_type<stream>>
     , public net::coroutine
 {
-    stream<NextLayer, deflateSupported>& ws_;
+    boost::weak_ptr<impl_type> wp_;
     detail::frame_buffer& fb_;
 
 public:
@@ -45,18 +46,19 @@ public:
     template<class Handler_>
     ping_op(
         Handler_&& h,
-        stream<NextLayer, deflateSupported>& ws,
+        boost::shared_ptr<impl_type> const& sp,
         detail::opcode op,
         ping_data const& payload)
-        : stable_async_op_base<
-            Handler, beast::executor_type<stream>>(
-                std::forward<Handler_>(h), ws.get_executor())
-        , ws_(ws)
+        : stable_async_op_base<Handler,
+            beast::executor_type<stream>>(
+                std::forward<Handler_>(h),
+                    sp->stream.get_executor())
+        , wp_(sp)
         , fb_(beast::allocate_stable<
             detail::frame_buffer>(*this))
     {
         // Serialize the ping or pong frame
-        ws.template write_ping<
+        sp->template write_ping<
             flat_static_buffer_base>(fb_, op, payload);
         (*this)({}, 0, false);
     }
@@ -67,14 +69,18 @@ public:
         bool cont = true)
     {
         boost::ignore_unused(bytes_transferred);
-        auto& impl = *ws_.impl_;
+        auto sp = wp_.lock();
+        if(! sp)
+            return this->invoke(cont,
+                net::error::operation_aborted);
+        auto& impl = *sp;
         BOOST_ASIO_CORO_REENTER(*this)
         {
             // Acquire the write lock
             if(! impl.wr_block.try_lock(this))
             {
                 BOOST_ASIO_CORO_YIELD
-                impl.paused_ping.emplace(std::move(*this));
+                impl.op_ping.emplace(std::move(*this));
                 impl.wr_block.lock(this);
                 BOOST_ASIO_CORO_YIELD
                 net::post(std::move(*this));
@@ -92,9 +98,10 @@ public:
 
         upcall:
             impl.wr_block.unlock(this);
-            impl.paused_close.maybe_invoke()
-                || impl.paused_rd.maybe_invoke()
-                || impl.paused_wr.maybe_invoke();
+            impl.op_close.maybe_invoke()
+                || impl.op_idle_ping.maybe_invoke()
+                || impl.op_rd.maybe_invoke()
+                || impl.op_wr.maybe_invoke();
             this->invoke(cont, ec);
         }
     }
@@ -102,6 +109,103 @@ public:
 
 //------------------------------------------------------------------------------
 
+#if 0
+template<
+    class NextLayer,
+    bool deflateSupported,
+    class Handler>
+void
+async_auto_ping(
+    stream<NextLayer, deflateSupported>& ws,
+    Handler&& handler)
+{
+    using handler_type =
+        typename std::decay<Handler>::type;
+
+    using base_type =
+        beast::stable_async_op_base<
+            handler_type, beast::executor_type<
+                stream<NextLayer, deflateSupported>>>;
+
+    struct async_op : base_type, net::coroutine
+    {
+        boost::weak_ptr<impl_type> impl_;
+        detail::frame_buffer& fb_;
+
+    public:
+        static constexpr int id = 4; // for soft_mutex
+
+        async_op(
+            Handler&& h,
+            stream<NextLayer, deflateSupported>& ws)
+            : base_type(std::move(h), ws.get_executor())
+            , impl_(ws.impl_)
+            , fb_(beast::allocate_stable<
+                detail::frame_buffer>(*this))
+        {
+            // Serialize the ping or pong frame
+            ping_data payload;
+            ws.template write_ping<
+                flat_static_buffer_base>(fb_, op, payload);
+            (*this)({}, 0, false);
+        }
+
+        void operator()(
+            error_code ec = {},
+            std::size_t bytes_transferred = 0,
+            bool cont = true)
+        {
+            boost::ignore_unused(bytes_transferred);
+            auto sp = impl_.lock();
+            if(! sp)
+                return;
+
+            auto& impl = *ws_.impl_;
+            BOOST_ASIO_CORO_REENTER(*this)
+            {
+                // Acquire the write lock
+                if(! impl.wr_block.try_lock(this))
+                {
+                    BOOST_ASIO_CORO_YIELD
+                    impl.op_idle_ping.emplace(std::move(*this));
+                    impl.wr_block.lock(this);
+                    BOOST_ASIO_CORO_YIELD
+                    net::post(std::move(*this));
+                    BOOST_ASSERT(impl.wr_block.is_locked(this));
+                }
+                if(impl.check_stop_now(ec))
+                    goto upcall;
+
+                // Send ping frame
+                BOOST_ASIO_CORO_YIELD
+                net::async_write(impl.stream, fb_.data(),
+                    beast::detail::bind_continuation(std::move(*this)));
+                if(impl.check_stop_now(ec))
+                    goto upcall;
+
+            upcall:
+                impl.wr_block.unlock(this);
+                impl.op_close.maybe_invoke()
+                    || impl.op_ping.maybe_invoke()
+                    || impl.op_rd.maybe_invoke()
+                    || impl.op_wr.maybe_invoke();
+                if(! cont)
+                {
+                    BOOST_ASIO_CORO_YIELD
+                    net::post(bind_front_handler(
+                        std::move(*this), ec));
+                }
+                this->invoke(ec);
+            }
+        }
+    };
+
+    async_op op(ws, std::forward<Handler>(handler));
+}
+#endif
+
+//------------------------------------------------------------------------------
+
 template<class NextLayer, bool deflateSupported>
 void
 stream<NextLayer, deflateSupported>::
@@ -121,7 +225,7 @@ ping(ping_data const& payload, error_code& ec)
     if(impl_->check_stop_now(ec))
         return;
     detail::frame_buffer fb;
-    write_ping<flat_static_buffer_base>(
+    impl_->template write_ping<flat_static_buffer_base>(
         fb, detail::opcode::ping, payload);
     net::write(impl_->stream, fb.data(), ec);
     if(impl_->check_stop_now(ec))
@@ -147,7 +251,7 @@ pong(ping_data const& payload, error_code& ec)
     if(impl_->check_stop_now(ec))
         return;
     detail::frame_buffer fb;
-    write_ping<flat_static_buffer_base>(
+    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))
@@ -167,7 +271,7 @@ async_ping(ping_data const& payload, WriteHandler&& handler)
         WriteHandler, void(error_code));
     ping_op<BOOST_ASIO_HANDLER_TYPE(
         WriteHandler, void(error_code))>(
-            std::move(init.completion_handler), *this,
+            std::move(init.completion_handler), impl_,
                 detail::opcode::ping, payload);
     return init.result.get();
 }
@@ -185,7 +289,7 @@ async_pong(ping_data const& payload, WriteHandler&& handler)
         WriteHandler, void(error_code));
     ping_op<BOOST_ASIO_HANDLER_TYPE(
         WriteHandler, void(error_code))>(
-            std::move(init.completion_handler), *this,
+            std::move(init.completion_handler), impl_,
                 detail::opcode::pong, payload);
     return init.result.get();
 }

include/boost/beast/websocket/impl/read.hpp

@@ -13,6 +13,7 @@
 #include <boost/beast/core/buffer_size.hpp>
 #include <boost/beast/websocket/teardown.hpp>
 #include <boost/beast/websocket/detail/mask.hpp>
+#include <boost/beast/websocket/impl/stream_impl.hpp>
 #include <boost/beast/core/async_op_base.hpp>
 #include <boost/beast/core/bind_handler.hpp>
 #include <boost/beast/core/buffers_prefix.hpp>
@@ -50,7 +51,7 @@ class stream<NextLayer, deflateSupported>::read_some_op
         Handler, beast::executor_type<stream>>
     , public net::coroutine
 {
-    stream& ws_;
+    boost::weak_ptr<impl_type> wp_;
     MutableBufferSequence bs_;
     buffers_suffix<MutableBufferSequence> cb_;
     std::size_t bytes_written_ = 0;
@@ -64,12 +65,13 @@ public:
     template<class Handler_>
     read_some_op(
         Handler_&& h,
-        stream<NextLayer, deflateSupported>& ws,
+        boost::shared_ptr<impl_type> const& sp,
         MutableBufferSequence const& bs)
         : async_op_base<
             Handler, beast::executor_type<stream>>(
-                std::forward<Handler_>(h), ws.get_executor())
-        , ws_(ws)
+                std::forward<Handler_>(h),
+                    sp->stream.get_executor())
+        , wp_(sp)
         , bs_(bs)
         , cb_(bs)
         , code_(close_code::none)
@@ -83,16 +85,22 @@ public:
         bool cont = true)
     {
         using beast::detail::clamp;
-        auto& impl = *ws_.impl_;
+        auto sp = wp_.lock();
+        if(! sp)
+            return this->invoke(cont,
+                net::error::operation_aborted, 0);
+        auto& impl = *sp;
         BOOST_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:
                 BOOST_ASIO_CORO_YIELD
-                impl.paused_r_rd.emplace(std::move(*this));
+                impl.op_r_rd.emplace(std::move(*this));
                 impl.rd_block.lock(this);
                 BOOST_ASIO_CORO_YIELD
                 net::post(std::move(*this));
@@ -139,7 +147,7 @@ public:
                 (! impl.rd_fh.fin || impl.rd_done))
             {
                 // Read frame header
-                while(! ws_.parse_fh(
+                while(! impl.parse_fh(
                     impl.rd_fh, impl.rd_buf, result_))
                 {
                     if(result_)
@@ -158,14 +166,15 @@ public:
                             impl.rd_buf, impl.rd_buf.max_size())),
                                 std::move(*this));
                     BOOST_ASSERT(impl.rd_block.is_locked(this));
+                    impl.rd_buf.commit(bytes_transferred);
                     if(impl.check_stop_now(ec))
                         goto upcall;
-                    impl.rd_buf.commit(bytes_transferred);
+                    impl.reset_idle();
 
                     // Allow a close operation
                     // to acquire the read block
                     impl.rd_block.unlock(this);
-                    if( impl.paused_r_close.maybe_invoke())
+                    if( impl.op_r_close.maybe_invoke())
                     {
                         // Suspend
                         BOOST_ASSERT(impl.rd_block.is_locked());
@@ -216,7 +225,7 @@ public:
                                 impl.ctrl_cb(
                                     frame_type::ping, payload);
                             impl.rd_fb.clear();
-                            ws_.template write_ping<
+                            impl.template write_ping<
                                 flat_static_buffer_base>(impl.rd_fb,
                                     detail::opcode::pong, payload);
                         }
@@ -224,13 +233,13 @@ public:
                         // Allow a close operation
                         // to acquire the read block
                         impl.rd_block.unlock(this);
-                        impl.paused_r_close.maybe_invoke();
+                        impl.op_r_close.maybe_invoke();
 
                         // Acquire the write lock
                         if(! impl.wr_block.try_lock(this))
                         {
                             BOOST_ASIO_CORO_YIELD
-                            impl.paused_rd.emplace(std::move(*this));
+                            impl.op_rd.emplace(std::move(*this));
                             impl.wr_block.lock(this);
                             BOOST_ASIO_CORO_YIELD
                             net::post(std::move(*this));
@@ -249,9 +258,10 @@ public:
                         if(impl.check_stop_now(ec))
                             goto upcall;
                         impl.wr_block.unlock(this);
-                        impl.paused_close.maybe_invoke() ||
-                            impl.paused_ping.maybe_invoke() ||
-                            impl.paused_wr.maybe_invoke();
+                        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;
                     }
 
@@ -264,9 +274,7 @@ public:
                             if(! cont)
                             {
                                 BOOST_ASIO_CORO_YIELD
-                                net::post(
-                                    ws_.get_executor(),
-                                        std::move(*this));
+                                net::post(std::move(*this));
                                 BOOST_ASSERT(cont);
                             }
                         }
@@ -291,9 +299,7 @@ public:
                             if(! cont)
                             {
                                 BOOST_ASIO_CORO_YIELD
-                                net::post(
-                                    ws_.get_executor(),
-                                        std::move(*this));
+                                net::post(std::move(*this));
                                 BOOST_ASSERT(cont);
                             }
                         }
@@ -355,9 +361,10 @@ public:
                             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.rd_buf.commit(bytes_transferred);
+                        impl.reset_idle();
                         if(impl.rd_fh.mask)
                             detail::mask_inplace(buffers_prefix(clamp(
                                 impl.rd_remain), impl.rd_buf.data()),
@@ -400,6 +407,7 @@ public:
                             clamp(impl.rd_remain), cb_), std::move(*this));
                         if(impl.check_stop_now(ec))
                             goto upcall;
+                        impl.reset_idle();
                         BOOST_ASSERT(bytes_transferred > 0);
                         auto const mb = buffers_prefix(
                             bytes_transferred, cb_);
@@ -443,6 +451,7 @@ public:
                                     std::move(*this));
                         if(impl.check_stop_now(ec))
                             goto upcall;
+                        impl.reset_idle();
                         BOOST_ASSERT(bytes_transferred > 0);
                         impl.rd_buf.commit(bytes_transferred);
                         if(impl.rd_fh.mask)
@@ -477,9 +486,8 @@ public:
                     else if(impl.rd_fh.fin)
                     {
                         // append the empty block codes
-                        static std::uint8_t constexpr
-                            empty_block[4] = {
-                                0x00, 0x00, 0xff, 0xff };
+                        std::uint8_t constexpr
+                            empty_block[4] = { 0x00, 0x00, 0xff, 0xff };
                         zs.next_in = empty_block;
                         zs.avail_in = sizeof(empty_block);
                         impl.inflate(zs, zlib::Flush::sync, ec);
@@ -537,7 +545,7 @@ public:
             if(! impl.wr_block.try_lock(this))
             {
                 BOOST_ASIO_CORO_YIELD
-                impl.paused_rd.emplace(std::move(*this));
+                impl.op_rd.emplace(std::move(*this));
                 impl.wr_block.lock(this);
                 BOOST_ASIO_CORO_YIELD
                 net::post(std::move(*this));
@@ -546,8 +554,7 @@ public:
                     goto upcall;
             }
 
-            // Set the status
-            impl.status_ = status::closing;
+            impl.change_status(status::closing);
 
             if(! impl.wr_close)
             {
@@ -555,7 +562,7 @@ public:
 
                 // Serialize close frame
                 impl.rd_fb.clear();
-                ws_.template write_close<
+                impl.template write_close<
                     flat_static_buffer_base>(
                         impl.rd_fb, code_);
 
@@ -585,18 +592,19 @@ public:
             if(! ec)
                 ec = result_;
             if(ec && ec != error::closed)
-                impl.status_ = status::failed;
+                impl.change_status(status::failed);
             else
-                impl.status_ = status::closed;
+                impl.change_status(status::closed);
             impl.close();
 
         upcall:
             impl.rd_block.try_unlock(this);
-            impl.paused_r_close.maybe_invoke();
+            impl.op_r_close.maybe_invoke();
             if(impl.wr_block.try_unlock(this))
-                impl.paused_close.maybe_invoke()
-                    || impl.paused_ping.maybe_invoke()
-                    || impl.paused_wr.maybe_invoke();
+                impl.op_close.maybe_invoke()
+                    || impl.op_idle_ping.maybe_invoke()
+                    || impl.op_ping.maybe_invoke()
+                    || impl.op_wr.maybe_invoke();
             this->invoke(cont, ec, bytes_written_);
         }
     }
@@ -613,7 +621,7 @@ class stream<NextLayer, deflateSupported>::read_op
         Handler, beast::executor_type<stream>>
     , public net::coroutine
 {
-    stream<NextLayer, deflateSupported>& ws_;
+    boost::weak_ptr<impl_type> wp_;
     DynamicBuffer& b_;
     std::size_t limit_;
     std::size_t bytes_written_ = 0;
@@ -623,14 +631,15 @@ public:
     template<class Handler_>
     read_op(
         Handler_&& h,
-        stream<NextLayer, deflateSupported>& ws,
+        boost::shared_ptr<impl_type> const& sp,
         DynamicBuffer& b,
         std::size_t limit,
         bool some)
-        : async_op_base<
-            Handler, beast::executor_type<stream>>(
-                std::forward<Handler_>(h), ws.get_executor())
-        , ws_(ws)
+        : async_op_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)())
@@ -645,26 +654,35 @@ public:
         bool cont = true)
     {
         using beast::detail::clamp;
-        boost::optional<typename
-            DynamicBuffer::mutable_buffers_type> mb;
+        auto sp = wp_.lock();
+        if(! sp)
+            return this->invoke(cont,
+                net::error::operation_aborted, 0);
+        auto& impl = *sp;
+        using mutable_buffers_type = typename
+            DynamicBuffer::mutable_buffers_type;
+        boost::optional<mutable_buffers_type> mb;
         BOOST_ASIO_CORO_REENTER(*this)
         {
             do
             {
                 mb = beast::detail::dynamic_buffer_prepare(b_,
-                    clamp(ws_.read_size_hint(b_), limit_),
+                    clamp(impl.read_size_hint_db(b_), limit_),
                         ec, error::buffer_overflow);
-                if(ec)
+                if(impl.check_stop_now(ec))
                     goto upcall;
+
+                // VFALCO TODO use boost::beast::bind_continuation
                 BOOST_ASIO_CORO_YIELD
-                ws_.async_read_some(*mb,
-                    beast::detail::bind_continuation(std::move(*this)));
+                read_some_op<mutable_buffers_type, read_op>(
+                    std::move(*this), sp, *mb);
                 b_.commit(bytes_transferred);
                 bytes_written_ += bytes_transferred;
                 if(ec)
                     goto upcall;
             }
-            while(! some_ && ! ws_.is_message_done());
+            while(! some_ && ! impl.rd_done);
+
         upcall:
             this->invoke(cont, ec, bytes_written_);
         }
@@ -730,7 +748,7 @@ async_read(DynamicBuffer& buffer, ReadHandler&& handler)
     read_op<DynamicBuffer, BOOST_ASIO_HANDLER_TYPE(
         ReadHandler, void(error_code, std::size_t))>(
             std::move(init.completion_handler),
-                *this, buffer, 0, false);
+                impl_, buffer, 0, false);
     return init.result.get();
 }
 
@@ -779,7 +797,7 @@ read_some(
     BOOST_ASSERT(size > 0);
     auto mb = beast::detail::dynamic_buffer_prepare(
         buffer, size, ec, error::buffer_overflow);
-    if(ec)
+    if(impl_->check_stop_now(ec))
         return 0;
     auto const bytes_written = read_some(*mb, ec);
     buffer.commit(bytes_written);
@@ -806,7 +824,7 @@ async_read_some(
     read_op<DynamicBuffer, BOOST_ASIO_HANDLER_TYPE(
         ReadHandler, void(error_code, std::size_t))>(
             std::move(init.completion_handler),
-                *this, buffer, limit, true);
+                impl_, buffer, limit, true);
     return init.result.get();
 }
 
@@ -862,7 +880,7 @@ loop:
     {
         // Read frame header
         error_code result;
-        while(! parse_fh(impl.rd_fh, impl.rd_buf, result))
+        while(! impl.parse_fh(impl.rd_fh, impl.rd_buf, result))
         {
             if(result)
             {
@@ -915,7 +933,7 @@ loop:
                 if(impl.ctrl_cb)
                     impl.ctrl_cb(frame_type::ping, payload);
                 detail::frame_buffer fb;
-                write_ping<flat_static_buffer_base>(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))
@@ -1186,7 +1204,7 @@ async_read_some(
         ReadHandler, void(error_code, std::size_t));
     read_some_op<MutableBufferSequence, BOOST_ASIO_HANDLER_TYPE(
         ReadHandler, void(error_code, std::size_t))>(
-            std::move(init.completion_handler), *this, buffers);
+            std::move(init.completion_handler), impl_, buffers);
     return init.result.get();
 }
 

include/boost/beast/websocket/impl/stream.hpp

@@ -15,6 +15,7 @@
 #include <boost/beast/websocket/teardown.hpp>
 #include <boost/beast/websocket/detail/hybi13.hpp>
 #include <boost/beast/websocket/detail/mask.hpp>
+#include <boost/beast/websocket/impl/stream_impl.hpp>
 #include <boost/beast/version.hpp>
 #include <boost/beast/http/read.hpp>
 #include <boost/beast/http/write.hpp>
@@ -29,6 +30,7 @@
 #include <boost/asio/steady_timer.hpp>
 #include <boost/assert.hpp>
 #include <boost/endian/buffers.hpp>
+#include <boost/make_shared.hpp>
 #include <boost/make_unique.hpp>
 #include <boost/throw_exception.hpp>
 #include <algorithm>
@@ -45,7 +47,7 @@ template<class NextLayer, bool deflateSupported>
 template<class... Args>
 stream<NextLayer, deflateSupported>::
 stream(Args&&... args)
-    : impl_(std::make_shared<impl_type>(
+    : impl_(boost::make_shared<impl_type>(
         std::forward<Args>(args)...))
 {
     BOOST_ASSERT(impl_->rd_buf.max_size() >=
@@ -140,6 +142,30 @@ read_size_hint(DynamicBuffer& buffer) const
 }
 
 //------------------------------------------------------------------------------
+//
+// Settings
+//
+//------------------------------------------------------------------------------
+
+// timeout
+
+template<class NextLayer, bool deflateSupported>
+void
+stream<NextLayer, deflateSupported>::
+get_option(timeout& opt)
+{
+    opt = impl_->timeout_opt;
+}
+
+template<class NextLayer, bool deflateSupported>
+void
+stream<NextLayer, deflateSupported>::
+set_option(timeout const& opt)
+{
+    impl_->set_option(opt);
+}
+
+//
 
 template<class NextLayer, bool deflateSupported>
 void
@@ -229,7 +255,7 @@ void
 stream<NextLayer, deflateSupported>::
 secure_prng(bool value)
 {
-    this->secure_prng_ = value;
+    this->impl_->secure_prng_ = value;
 }
 
 template<class NextLayer, bool deflateSupported>
@@ -271,471 +297,6 @@ text() const
 
 //------------------------------------------------------------------------------
 
-// Attempt to read a complete frame header.
-// Returns `false` if more bytes are needed
-template<class NextLayer, bool deflateSupported>
-template<class DynamicBuffer>
-bool
-stream<NextLayer, deflateSupported>::
-parse_fh(
-    detail::frame_header& fh,
-    DynamicBuffer& b,
-    error_code& ec)
-{
-    if(buffer_size(b.data()) < 2)
-    {
-        // need more bytes
-        ec = {};
-        return false;
-    }
-    buffers_suffix<typename
-        DynamicBuffer::const_buffers_type> cb{
-            b.data()};
-    std::size_t need;
-    {
-        std::uint8_t tmp[2];
-        cb.consume(net::buffer_copy(
-            net::buffer(tmp), cb));
-        fh.len = tmp[1] & 0x7f;
-        switch(fh.len)
-        {
-            case 126: need = 2; break;
-            case 127: need = 8; break;
-            default:
-                need = 0;
-        }
-        fh.mask = (tmp[1] & 0x80) != 0;
-        if(fh.mask)
-            need += 4;
-        if(buffer_size(cb) < need)
-        {
-            // need more bytes
-            ec = {};
-            return false;
-        }
-        fh.op   = static_cast<
-            detail::opcode>(tmp[0] & 0x0f);
-        fh.fin  = (tmp[0] & 0x80) != 0;
-        fh.rsv1 = (tmp[0] & 0x40) != 0;
-        fh.rsv2 = (tmp[0] & 0x20) != 0;
-        fh.rsv3 = (tmp[0] & 0x10) != 0;
-    }
-    switch(fh.op)
-    {
-    case detail::opcode::binary:
-    case detail::opcode::text:
-        if(impl_->rd_cont)
-        {
-            // new data frame when continuation expected
-            ec = error::bad_data_frame;
-            return false;
-        }
-        if(fh.rsv2 || fh.rsv3 ||
-            ! impl_->rd_deflated(fh.rsv1))
-        {
-            // reserved bits not cleared
-            ec = error::bad_reserved_bits;
-            return false;
-        }
-        break;
-
-    case detail::opcode::cont:
-        if(! impl_->rd_cont)
-        {
-            // continuation without an active message
-            ec = error::bad_continuation;
-            return false;
-        }
-        if(fh.rsv1 || fh.rsv2 || fh.rsv3)
-        {
-            // reserved bits not cleared
-            ec = error::bad_reserved_bits;
-            return false;
-        }
-        break;
-
-    default:
-        if(detail::is_reserved(fh.op))
-        {
-            // reserved opcode
-            ec = error::bad_opcode;
-            return false;
-        }
-        if(! fh.fin)
-        {
-            // fragmented control message
-            ec = error::bad_control_fragment;
-            return false;
-        }
-        if(fh.len > 125)
-        {
-            // invalid length for control message
-            ec = error::bad_control_size;
-            return false;
-        }
-        if(fh.rsv1 || fh.rsv2 || fh.rsv3)
-        {
-            // reserved bits not cleared
-            ec = error::bad_reserved_bits;
-            return false;
-        }
-        break;
-    }
-    if(impl_->role == role_type::server && ! fh.mask)
-    {
-        // unmasked frame from client
-        ec = error::bad_unmasked_frame;
-        return false;
-    }
-    if(impl_->role == role_type::client && fh.mask)
-    {
-        // masked frame from server
-        ec = error::bad_masked_frame;
-        return false;
-    }
-    if(detail::is_control(fh.op) &&
-        buffer_size(cb) < need + fh.len)
-    {
-        // Make the entire control frame payload
-        // get read in before we return `true`
-        return false;
-    }
-    switch(fh.len)
-    {
-    case 126:
-    {
-        std::uint8_t tmp[2];
-        BOOST_ASSERT(buffer_size(cb) >= sizeof(tmp));
-        cb.consume(net::buffer_copy(net::buffer(tmp), cb));
-        fh.len = detail::big_uint16_to_native(&tmp[0]);
-        if(fh.len < 126)
-        {
-            // length not canonical
-            ec = error::bad_size;
-            return false;
-        }
-        break;
-    }
-    case 127:
-    {
-        std::uint8_t tmp[8];
-        BOOST_ASSERT(buffer_size(cb) >= sizeof(tmp));
-        cb.consume(net::buffer_copy(net::buffer(tmp), cb));
-        fh.len = detail::big_uint64_to_native(&tmp[0]);
-        if(fh.len < 65536)
-        {
-            // length not canonical
-            ec = error::bad_size;
-            return false;
-        }
-        break;
-    }
-    }
-    if(fh.mask)
-    {
-        std::uint8_t tmp[4];
-        BOOST_ASSERT(buffer_size(cb) >= sizeof(tmp));
-        cb.consume(net::buffer_copy(net::buffer(tmp), cb));
-        fh.key = detail::little_uint32_to_native(&tmp[0]);
-        detail::prepare_key(impl_->rd_key, fh.key);
-    }
-    else
-    {
-        // initialize this otherwise operator== breaks
-        fh.key = 0;
-    }
-    if(! detail::is_control(fh.op))
-    {
-        if(fh.op != detail::opcode::cont)
-        {
-            impl_->rd_size = 0;
-            impl_->rd_op = fh.op;
-        }
-        else
-        {
-            if(impl_->rd_size > (std::numeric_limits<
-                std::uint64_t>::max)() - fh.len)
-            {
-                // message size exceeds configured limit
-                ec = error::message_too_big;
-                return false;
-            }
-        }
-        if(! impl_->rd_deflated())
-        {
-            if(impl_->rd_msg_max && beast::detail::sum_exceeds(
-                impl_->rd_size, fh.len, impl_->rd_msg_max))
-            {
-                // message size exceeds configured limit
-                ec = error::message_too_big;
-                return false;
-            }
-        }
-        impl_->rd_cont = ! fh.fin;
-        impl_->rd_remain = fh.len;
-    }
-    b.consume(b.size() - buffer_size(cb));
-    ec = {};
-    return true;
-}
-
-template<class NextLayer, bool deflateSupported>
-template<class DynamicBuffer>
-void
-stream<NextLayer, deflateSupported>::
-write_close(DynamicBuffer& db, close_reason const& cr)
-{
-    using namespace boost::endian;
-    detail::frame_header fh;
-    fh.op = detail::opcode::close;
-    fh.fin = true;
-    fh.rsv1 = false;
-    fh.rsv2 = false;
-    fh.rsv3 = false;
-    fh.len = cr.code == close_code::none ?
-        0 : 2 + cr.reason.size();
-    if(impl_->role == role_type::client)
-    {
-        fh.mask = true;
-        fh.key = this->create_mask();
-    }
-    else
-    {
-        fh.mask = false;
-    }
-    detail::write(db, fh);
-    if(cr.code != close_code::none)
-    {
-        detail::prepared_key key;
-        if(fh.mask)
-            detail::prepare_key(key, fh.key);
-        {
-            std::uint8_t tmp[2];
-            ::new(&tmp[0]) big_uint16_buf_t{
-                (std::uint16_t)cr.code};
-            auto mb = db.prepare(2);
-            net::buffer_copy(mb,
-                net::buffer(tmp));
-            if(fh.mask)
-                detail::mask_inplace(mb, key);
-            db.commit(2);
-        }
-        if(! cr.reason.empty())
-        {
-            auto mb = db.prepare(cr.reason.size());
-            net::buffer_copy(mb,
-                net::const_buffer(
-                    cr.reason.data(), cr.reason.size()));
-            if(fh.mask)
-                detail::mask_inplace(mb, key);
-            db.commit(cr.reason.size());
-        }
-    }
-}
-
-template<class NextLayer, bool deflateSupported>
-template<class DynamicBuffer>
-void
-stream<NextLayer, deflateSupported>::
-write_ping(DynamicBuffer& db,
-    detail::opcode code, ping_data const& data)
-{
-    detail::frame_header fh;
-    fh.op = code;
-    fh.fin = true;
-    fh.rsv1 = false;
-    fh.rsv2 = false;
-    fh.rsv3 = false;
-    fh.len = data.size();
-    fh.mask = impl_->role == role_type::client;
-    if(fh.mask)
-        fh.key = this->create_mask();
-    detail::write(db, fh);
-    if(data.empty())
-        return;
-    detail::prepared_key key;
-    if(fh.mask)
-        detail::prepare_key(key, fh.key);
-    auto mb = db.prepare(data.size());
-    net::buffer_copy(mb,
-        net::const_buffer(
-            data.data(), data.size()));
-    if(fh.mask)
-        detail::mask_inplace(mb, key);
-    db.commit(data.size());
-}
-
-//------------------------------------------------------------------------------
-
-template<class NextLayer, bool deflateSupported>
-template<class Decorator>
-request_type
-stream<NextLayer, deflateSupported>::
-build_request(detail::sec_ws_key_type& key,
-    string_view host, string_view target,
-        Decorator const& decorator)
-{
-    request_type req;
-    req.target(target);
-    req.version(11);
-    req.method(http::verb::get);
-    req.set(http::field::host, host);
-    req.set(http::field::upgrade, "websocket");
-    req.set(http::field::connection, "upgrade");
-    detail::make_sec_ws_key(key);
-    req.set(http::field::sec_websocket_key, key);
-    req.set(http::field::sec_websocket_version, "13");
-    impl_->build_request_pmd(req);
-    decorator(req);
-    if(! req.count(http::field::user_agent))
-        req.set(http::field::user_agent,
-            BOOST_BEAST_VERSION_STRING);
-    return req;
-}
-
-template<class NextLayer, bool deflateSupported>
-template<class Body, class Allocator, class Decorator>
-response_type
-stream<NextLayer, deflateSupported>::
-build_response(
-    http::request<Body,
-        http::basic_fields<Allocator>> const& req,
-    Decorator const& decorator,
-    error_code& result)
-{
-    auto const decorate =
-        [&decorator](response_type& res)
-        {
-            decorator(res);
-            if(! res.count(http::field::server))
-            {
-                BOOST_STATIC_ASSERT(sizeof(BOOST_BEAST_VERSION_STRING) < 20);
-                static_string<20> s(BOOST_BEAST_VERSION_STRING);
-                res.set(http::field::server, s);
-            }
-        };
-    auto err =
-        [&](error e)
-        {
-            result = e;
-            response_type res;
-            res.version(req.version());
-            res.result(http::status::bad_request);
-            res.body() = result.message();
-            res.prepare_payload();
-            decorate(res);
-            return res;
-        };
-    if(req.version() != 11)
-        return err(error::bad_http_version);
-    if(req.method() != http::verb::get)
-        return err(error::bad_method);
-    if(! req.count(http::field::host))
-        return err(error::no_host);
-    {
-        auto const it = req.find(http::field::connection);
-        if(it == req.end())
-            return err(error::no_connection);
-        if(! http::token_list{it->value()}.exists("upgrade"))
-            return err(error::no_connection_upgrade);
-    }
-    {
-        auto const it = req.find(http::field::upgrade);
-        if(it == req.end())
-            return err(error::no_upgrade);
-        if(! http::token_list{it->value()}.exists("websocket"))
-            return err(error::no_upgrade_websocket);
-    }
-    string_view key;
-    {
-        auto const it = req.find(http::field::sec_websocket_key);
-        if(it == req.end())
-            return err(error::no_sec_key);
-        key = it->value();
-        if(key.size() > detail::sec_ws_key_type::max_size_n)
-            return err(error::bad_sec_key);
-    }
-    {
-        auto const it = req.find(http::field::sec_websocket_version);
-        if(it == req.end())
-            return err(error::no_sec_version);
-        if(it->value() != "13")
-        {
-            response_type res;
-            res.result(http::status::upgrade_required);
-            res.version(req.version());
-            res.set(http::field::sec_websocket_version, "13");
-            result = error::bad_sec_version;
-            res.body() = result.message();
-            res.prepare_payload();
-            decorate(res);
-            return res;
-        }
-    }
-
-    response_type res;
-    res.result(http::status::switching_protocols);
-    res.version(req.version());
-    res.set(http::field::upgrade, "websocket");
-    res.set(http::field::connection, "upgrade");
-    {
-        detail::sec_ws_accept_type acc;
-        detail::make_sec_ws_accept(acc, key);
-        res.set(http::field::sec_websocket_accept, acc);
-    }
-    impl_->build_response_pmd(res, req);
-    decorate(res);
-    result = {};
-    return res;
-}
-
-// Called when the WebSocket Upgrade response is received
-template<class NextLayer, bool deflateSupported>
-void
-stream<NextLayer, deflateSupported>::
-on_response(
-    response_type const& res,
-    detail::sec_ws_key_type const& key,
-    error_code& ec)
-{
-    auto const err =
-        [&](error e)
-        {
-            ec = e;
-        };
-    if(res.result() != http::status::switching_protocols)
-        return err(error::upgrade_declined);
-    if(res.version() != 11)
-        return err(error::bad_http_version);
-    {
-        auto const it = res.find(http::field::connection);
-        if(it == res.end())
-            return err(error::no_connection);
-        if(! http::token_list{it->value()}.exists("upgrade"))
-            return err(error::no_connection_upgrade);
-    }
-    {
-        auto const it = res.find(http::field::upgrade);
-        if(it == res.end())
-            return err(error::no_upgrade);
-        if(! http::token_list{it->value()}.exists("websocket"))
-            return err(error::no_upgrade_websocket);
-    }
-    {
-        auto const it = res.find(http::field::sec_websocket_accept);
-        if(it == res.end())
-            return err(error::no_sec_accept);
-        detail::sec_ws_accept_type acc;
-        detail::make_sec_ws_accept(acc, key);
-        if(acc.compare(it->value()) != 0)
-            return err(error::bad_sec_accept);
-    }
-
-    ec = {};
-    impl_->on_response_pmd(res);
-    impl_->open(role_type::client);
-}
-
 // _Fail the WebSocket Connection_
 template<class NextLayer, bool deflateSupported>
 void
@@ -746,12 +307,12 @@ do_fail(
     error_code& ec)             // set to the error, else set to ev
 {
     BOOST_ASSERT(ev);
-    impl_->status_ = status::closing;
+    impl_->change_status(status::closing);
     if(code != close_code::none && ! impl_->wr_close)
     {
         impl_->wr_close = true;
         detail::frame_buffer fb;
-        write_close<
+        impl_->template write_close<
             flat_static_buffer_base>(fb, code);
         net::write(impl_->stream, fb.data(), ec);
         if(impl_->check_stop_now(ec))
@@ -768,9 +329,9 @@ do_fail(
     if(! ec)
         ec = ev;
     if(ec && ec != error::closed)
-        impl_->status_ = status::failed;
+        impl_->change_status(status::failed);
     else
-        impl_->status_ = status::closed;
+        impl_->change_status(status::closed);
     impl_->close();
 }
 

include/boost/beast/websocket/impl/stream_impl.hpp

@@ -10,16 +10,32 @@
 #ifndef BOOST_BEAST_WEBSOCKET_IMPL_STREAM_IMPL_HPP
 #define BOOST_BEAST_WEBSOCKET_IMPL_STREAM_IMPL_HPP
 
+#include <boost/beast/version.hpp>
+#include <boost/beast/http/read.hpp>
+#include <boost/beast/http/write.hpp>
+#include <boost/beast/http/rfc7230.hpp>
+#include <boost/beast/core/buffers_cat.hpp>
+#include <boost/beast/core/buffers_prefix.hpp>
+#include <boost/beast/core/buffers_suffix.hpp>
+#include <boost/beast/core/flat_static_buffer.hpp>
 #include <boost/beast/core/saved_handler.hpp>
 #include <boost/beast/core/static_buffer.hpp>
 #include <boost/beast/core/stream_traits.hpp>
+#include <boost/beast/core/detail/clamp.hpp>
+#include <boost/beast/core/detail/type_traits.hpp>
+#include <boost/beast/websocket/rfc6455.hpp>
 #include <boost/beast/websocket/detail/frame.hpp>
+#include <boost/beast/websocket/detail/hybi13.hpp>
+#include <boost/beast/websocket/detail/mask.hpp>
 #include <boost/beast/websocket/detail/pmd_extension.hpp>
+#include <boost/beast/websocket/detail/prng.hpp>
 #include <boost/beast/websocket/detail/soft_mutex.hpp>
 #include <boost/beast/websocket/detail/utf8_checker.hpp>
+#include <boost/beast/version.hpp>
 #include <boost/asio/bind_executor.hpp>
 #include <boost/asio/steady_timer.hpp>
 #include <boost/core/empty_value.hpp>
+#include <boost/enable_shared_from_this.hpp>
 #include <boost/optional.hpp>
 
 namespace boost {
@@ -29,7 +45,7 @@ namespace websocket {
 template<
     class NextLayer, bool deflateSupported>
 struct stream<NextLayer, deflateSupported>::impl_type
-    : std::enable_shared_from_this<impl_type>
+    : boost::enable_shared_from_this<impl_type>
     , detail::impl_base<deflateSupported>
 {
     NextLayer               stream;         // The underlying stream
@@ -68,17 +84,25 @@ struct stream<NextLayer, deflateSupported>::impl_type
     std::size_t             wr_buf_opt      /* write buffer size option setting */ = 4096;
     detail::fh_buffer       wr_fb;          // header buffer used for writes
 
-    saved_handler           paused_rd;      // paused read op
-    saved_handler           paused_wr;      // paused write op
-    saved_handler           paused_ping;    // paused ping op
-    saved_handler           paused_close;   // paused close op
-    saved_handler           paused_r_rd;    // paused read op (async read)
-    saved_handler           paused_r_close; // paused close op (async read)
+    saved_handler           op_rd;          // paused read op
+    saved_handler           op_wr;          // paused write op
+    saved_handler           op_ping;        // paused ping op
+    saved_handler           op_idle_ping;   // paused idle ping op
+    saved_handler           op_close;       // paused close op
+    saved_handler           op_r_rd;        // paused read op (async read)
+    saved_handler           op_r_close;     // paused close op (async read)
 
-    boost::optional<bool>   tm_auto_ping;
-    bool                    tm_opt          /* true if auto-timeout option is set */ = false;
-    bool                    tm_idle;        // set to false on incoming frames
-    time_point::duration    tm_dur          /* duration of timer */ = std::chrono::seconds(1);
+    bool secure_prng_ = true;
+
+    bool ec_delivered = false;
+    bool timed_out = false;
+    int idle_counter = 0;
+
+    // settings
+
+    timeout timeout_opt;
+
+    //
 
     template<class... Args>
     impl_type(Args&&... args)
@@ -91,6 +115,9 @@ struct stream<NextLayer, deflateSupported>::impl_type
     open(role_type role_)
     {
         // VFALCO TODO analyze and remove dupe code in reset()
+        timer.expires_at(never());
+        timed_out = false;
+        cr.code = close_code::none;
         role = role_;
         status_ = status::open;
         rd_remain = 0;
@@ -106,13 +133,10 @@ struct stream<NextLayer, deflateSupported>::impl_type
         // stream exhibits undefined behavior.
         wr_block.reset();
         rd_block.reset();
-        cr.code = close_code::none;
 
         wr_cont = false;
         wr_buf_size = 0;
 
-        tm_idle = false;
-
         this->open_pmd(role);
     }
 
@@ -128,6 +152,8 @@ struct stream<NextLayer, deflateSupported>::impl_type
     reset()
     {
         BOOST_ASSERT(status_ != status::open);
+        timer.expires_at(never());
+        cr.code = close_code::none;
         rd_remain = 0;
         rd_cont = false;
         rd_done = true;
@@ -141,8 +167,6 @@ struct stream<NextLayer, deflateSupported>::impl_type
         // stream exhibits undefined behavior.
         wr_block.reset();
         rd_block.reset();
-        cr.code = close_code::none;
-        tm_idle = false;
 
         // VFALCO Is this needed?
         timer.cancel();
@@ -173,126 +197,88 @@ struct stream<NextLayer, deflateSupported>::impl_type
         }
     }
 
-private:
-    bool
-    is_timer_set() const
-    {
-        return timer.expiry() == never();
-    }
+    //--------------------------------------------------------------------------
 
-    // returns `true` if we try sending a ping and
-    // getting a pong before closing an idle stream.
-    bool
-    is_auto_ping_enabled() const
-    {
-        if(tm_auto_ping.has_value())
-            return *tm_auto_ping;
-        if(role == role_type::server)
-            return true;
-        return false;
-    }
+    template<class Decorator>
+    request_type
+    build_request(
+        detail::sec_ws_key_type& key,
+        string_view host, string_view target,
+        Decorator const& decorator);
 
-    template<class Executor>
-    class timeout_handler
-        : boost::empty_value<Executor>
-    {
-        std::weak_ptr<impl_type> wp_;
-
-    public:
-        timeout_handler(
-            Executor const& ex,
-            std::shared_ptr<impl_type> const& sp)
-            : boost::empty_value<Executor>(
-                boost::empty_init_t{}, ex)
-            , wp_(sp)
-        {
-        }
-
-        using executor_type = Executor;
-
-        executor_type
-        get_executor() const
-        {
-            return this->get();
-        }
-
-        void
-        operator()(error_code ec)
-        {
-            // timer canceled?
-            if(ec == net::error::operation_aborted)
-                return;
-            BOOST_ASSERT(! ec);
-
-            // stream destroyed?
-            auto sp = wp_.lock();
-            if(! sp)
-                return;
-            auto& impl = *sp;
-
-            close_socket(get_lowest_layer(impl.stream));
-#if 0
-            if(! impl.tm_idle)
-            {
-                impl.tm_idle = true;
-                BOOST_VERIFY(
-                    impl.timer.expires_after(impl.tm_dur) == 0);
-                return impl.timer.async_wait(std::move(*this));
-            }
-            if(impl.is_auto_ping_enabled())
-            {
-                // send ping
-            }
-            else
-            {
-                // timeout
-            }
-#endif
-        }
-    };
-public:
-
-    // called when there is qualified activity
     void
-    activity()
+    on_response(
+        response_type const& res,
+        detail::sec_ws_key_type const& key,
+        error_code& ec);
+
+    template<class Body, class Allocator, class Decorator>
+    response_type
+    build_response(
+        http::request<Body,
+            http::basic_fields<Allocator>> const& req,
+        Decorator const& decorator,
+        error_code& result);
+
+    // Attempt to read a complete frame header.
+    // Returns `false` if more bytes are needed
+    template<class DynamicBuffer>
+    bool
+    parse_fh(detail::frame_header& fh,
+        DynamicBuffer& b, error_code& ec);
+
+    std::uint32_t
+    create_mask()
     {
-        tm_idle = false;
+        auto g = detail::make_prng(secure_prng_);
+        for(;;)
+            if(auto key = g())
+                return key;
     }
 
-    // Maintain the expiration timer
-    template<class Executor>
-    void
-    update_timer(Executor const& ex)
+    std::size_t
+    read_size_hint(std::size_t initial_size) const
     {
-        if(role == role_type::server)
-        {
-            if(! is_timer_set())
-            {
-                // turn timer on
-                timer.expires_after(tm_dur);
-                timer.async_wait(
-                    timeout_handler<Executor>(
-                        ex, this->shared_from_this()));
-            }
-        }
-        else if(tm_opt && ! is_timer_set())
-        {
-            // turn timer on
-            timer.expires_after(tm_dur);
-            timer.async_wait(
-                timeout_handler<Executor>(
-                    ex, this->shared_from_this()));
-        }
-        else if(! tm_opt && is_timer_set())
-        {
-            // turn timer off
-            timer.cancel();
-        }
+        return this->read_size_hint_pmd(
+            initial_size, rd_done, rd_remain, rd_fh);
     }
 
+    template<class DynamicBuffer>
+    std::size_t
+    read_size_hint_db(DynamicBuffer& buffer) const
+    {
+        auto const initial_size = (std::min)(
+            +tcp_frame_size,
+            buffer.max_size() - buffer.size());
+        if(initial_size == 0)
+            return 1; // buffer is full
+        return this->read_size_hint(initial_size);
+    }
+
+    template<class DynamicBuffer>
+    void
+    write_ping(DynamicBuffer& db,
+        detail::opcode code, ping_data const& data);
+
+    template<class DynamicBuffer>
+    void
+    write_close(DynamicBuffer& db, close_reason const& cr);
+
     //--------------------------------------------------------------------------
 
-    bool ec_delivered = false;
+    void
+    set_option(timeout const& opt)
+    {
+        if( opt.handshake_timeout == none() &&
+            opt.idle_timeout == none())
+        {
+            // turn timer off
+            timer.cancel();
+            timer.expires_at(never());
+        }
+
+        timeout_opt = opt;
+    }
 
     // Determine if an operation should stop and
     // deliver an error code to the completion handler.
@@ -305,6 +291,14 @@ public:
     bool
     check_stop_now(error_code& ec)
     {
+        // Deliver the timeout to the first caller
+        if(timed_out)
+        {
+            timed_out = false;
+            ec = beast::error::timeout;
+            return true;
+        }
+
         // If the stream is closed then abort
         if( status_ == status::closed ||
             status_ == status::failed)
@@ -339,8 +333,11 @@ public:
     {
         switch(new_status)
         {
+        case status::handshake:
+            break;
+
         case status::closing:
-            BOOST_ASSERT(status_ == status::open);
+            //BOOST_ASSERT(status_ == status::open);
             break;
 
         case status::failed:
@@ -353,8 +350,639 @@ public:
         }
         status_ = new_status;
     }
+
+    // Called to disarm the idle timeout counter
+    void
+    reset_idle()
+    {
+        idle_counter = 0;
+    }
+
+    // Maintain the expiration timer
+    template<class Executor>
+    void
+    update_timer(Executor const& ex)
+    {
+        switch(status_)
+        {
+        case status::handshake:
+            BOOST_ASSERT(idle_counter == 0);
+            if(! is_timer_set() &&
+                timeout_opt.handshake_timeout != none())
+            {
+                timer.expires_after(
+                    timeout_opt.handshake_timeout);
+                timer.async_wait(
+                    timeout_handler<Executor>(
+                        ex, this->weak_from_this()));
+            }
+            break;
+
+        case status::open:
+            if(timeout_opt.idle_timeout != none())
+            {
+                idle_counter = 0;
+                timer.expires_after(
+                    timeout_opt.idle_timeout);
+                timer.async_wait(
+                    timeout_handler<Executor>(
+                        ex, this->weak_from_this()));
+            }
+            else
+            {
+                timer.cancel();
+                timer.expires_at(never());
+            }
+            break;
+
+        case status::closing:
+            if(timeout_opt.handshake_timeout != none())
+            {
+                idle_counter = 0;
+                timer.expires_after(
+                    timeout_opt.handshake_timeout);
+                timer.async_wait(
+                    timeout_handler<Executor>(
+                        ex, this->weak_from_this()));
+            }
+            else
+            {
+                BOOST_ASSERT(! is_timer_set());
+            }
+            break;
+
+        case status::failed:
+        case status::closed:
+            // this->close(); // Is this right?
+            timer.cancel();
+            timer.expires_at(never());
+            break;
+        }
+    }
+
+private:
+    bool
+    is_timer_set() const
+    {
+        return timer.expiry() != never();
+    }
+
+    template<class Executor>
+    class timeout_handler
+        : boost::empty_value<Executor>
+    {
+        boost::weak_ptr<impl_type> wp_;
+
+    public:
+        timeout_handler(
+            Executor const& ex,
+            boost::weak_ptr<impl_type>&& wp)
+            : boost::empty_value<Executor>(
+                boost::empty_init_t{}, ex)
+            , wp_(std::move(wp))
+        {
+        }
+
+        using executor_type = Executor;
+
+        executor_type
+        get_executor() const
+        {
+            return this->get();
+        }
+
+        void
+        operator()(error_code ec)
+        {
+            // timer canceled?
+            if(ec == net::error::operation_aborted)
+                return;
+            BOOST_ASSERT(! ec);
+
+            // stream destroyed?
+            auto sp = wp_.lock();
+            if(! sp)
+                return;
+            auto& impl = *sp;
+
+            switch(impl.status_)
+            {
+            case status::handshake:
+                impl.timed_out = true;
+                close_socket(get_lowest_layer(impl.stream));
+                return;
+
+            case status::open:
+                // timeout was disabled
+                if(impl.timeout_opt.idle_timeout == none())
+                    return;
+
+                if( impl.timeout_opt.keep_alive_pings &&
+                    impl.idle_counter < 1)
+                {
+                    // <- send ping
+
+                    ++impl.idle_counter;
+                    impl.timer.expires_after(
+                        impl.timeout_opt.idle_timeout / 2);
+                    impl.timer.async_wait(std::move(*this));
+                    return;
+                }
+
+                // timeout
+                impl.timed_out = true;
+                close_socket(get_lowest_layer(impl.stream));
+                return;
+
+            case status::closing:
+                impl.timed_out = true;
+                close_socket(get_lowest_layer(impl.stream));
+                return;
+
+            case status::closed:
+            case status::failed:
+                // nothing to do?
+                return;
+            }
+        }
+    };
 };
 
+//--------------------------------------------------------------------------
+//
+// client
+//
+//--------------------------------------------------------------------------
+
+template<class NextLayer, bool deflateSupported>
+template<class Decorator>
+request_type
+stream<NextLayer, deflateSupported>::impl_type::
+build_request(
+    detail::sec_ws_key_type& key,
+    string_view host, string_view target,
+    Decorator const& decorator)
+{
+    request_type req;
+    req.target(target);
+    req.version(11);
+    req.method(http::verb::get);
+    req.set(http::field::host, host);
+    req.set(http::field::upgrade, "websocket");
+    req.set(http::field::connection, "upgrade");
+    detail::make_sec_ws_key(key);
+    req.set(http::field::sec_websocket_key, key);
+    req.set(http::field::sec_websocket_version, "13");
+    this->build_request_pmd(req);
+    decorator(req);
+    if(! req.count(http::field::user_agent))
+        req.set(http::field::user_agent,
+            BOOST_BEAST_VERSION_STRING);
+    return req;
+}
+
+// Called when the WebSocket Upgrade response is received
+template<class NextLayer, bool deflateSupported>
+void
+stream<NextLayer, deflateSupported>::impl_type::
+on_response(
+    response_type const& res,
+    detail::sec_ws_key_type const& key,
+    error_code& ec)
+{
+    auto const err =
+        [&](error e)
+        {
+            ec = e;
+        };
+    if(res.result() != http::status::switching_protocols)
+        return err(error::upgrade_declined);
+    if(res.version() != 11)
+        return err(error::bad_http_version);
+    {
+        auto const it = res.find(http::field::connection);
+        if(it == res.end())
+            return err(error::no_connection);
+        if(! http::token_list{it->value()}.exists("upgrade"))
+            return err(error::no_connection_upgrade);
+    }
+    {
+        auto const it = res.find(http::field::upgrade);
+        if(it == res.end())
+            return err(error::no_upgrade);
+        if(! http::token_list{it->value()}.exists("websocket"))
+            return err(error::no_upgrade_websocket);
+    }
+    {
+        auto const it = res.find(
+            http::field::sec_websocket_accept);
+        if(it == res.end())
+            return err(error::no_sec_accept);
+        detail::sec_ws_accept_type acc;
+        detail::make_sec_ws_accept(acc, key);
+        if(acc.compare(it->value()) != 0)
+            return err(error::bad_sec_accept);
+    }
+
+    ec = {};
+    this->on_response_pmd(res);
+    this->open(role_type::client);
+}
+
+//--------------------------------------------------------------------------
+
+template<class NextLayer, bool deflateSupported>
+template<class Body, class Allocator, class Decorator>
+response_type
+stream<NextLayer, deflateSupported>::impl_type::
+build_response(
+    http::request<Body,
+        http::basic_fields<Allocator>> const& req,
+    Decorator const& decorator,
+    error_code& result)
+{
+    auto const decorate =
+        [&decorator](response_type& res)
+        {
+            decorator(res);
+            if(! res.count(http::field::server))
+            {
+                BOOST_STATIC_ASSERT(sizeof(BOOST_BEAST_VERSION_STRING) < 20);
+                static_string<20> s(BOOST_BEAST_VERSION_STRING);
+                res.set(http::field::server, s);
+            }
+        };
+    auto err =
+        [&](error e)
+        {
+            result = e;
+            response_type res;
+            res.version(req.version());
+            res.result(http::status::bad_request);
+            res.body() = result.message();
+            res.prepare_payload();
+            decorate(res);
+            return res;
+        };
+    if(req.version() != 11)
+        return err(error::bad_http_version);
+    if(req.method() != http::verb::get)
+        return err(error::bad_method);
+    if(! req.count(http::field::host))
+        return err(error::no_host);
+    {
+        auto const it = req.find(http::field::connection);
+        if(it == req.end())
+            return err(error::no_connection);
+        if(! http::token_list{it->value()}.exists("upgrade"))
+            return err(error::no_connection_upgrade);
+    }
+    {
+        auto const it = req.find(http::field::upgrade);
+        if(it == req.end())
+            return err(error::no_upgrade);
+        if(! http::token_list{it->value()}.exists("websocket"))
+            return err(error::no_upgrade_websocket);
+    }
+    string_view key;
+    {
+        auto const it = req.find(http::field::sec_websocket_key);
+        if(it == req.end())
+            return err(error::no_sec_key);
+        key = it->value();
+        if(key.size() > detail::sec_ws_key_type::max_size_n)
+            return err(error::bad_sec_key);
+    }
+    {
+        auto const it = req.find(http::field::sec_websocket_version);
+        if(it == req.end())
+            return err(error::no_sec_version);
+        if(it->value() != "13")
+        {
+            response_type res;
+            res.result(http::status::upgrade_required);
+            res.version(req.version());
+            res.set(http::field::sec_websocket_version, "13");
+            result = error::bad_sec_version;
+            res.body() = result.message();
+            res.prepare_payload();
+            decorate(res);
+            return res;
+        }
+    }
+
+    response_type res;
+    res.result(http::status::switching_protocols);
+    res.version(req.version());
+    res.set(http::field::upgrade, "websocket");
+    res.set(http::field::connection, "upgrade");
+    {
+        detail::sec_ws_accept_type acc;
+        detail::make_sec_ws_accept(acc, key);
+        res.set(http::field::sec_websocket_accept, acc);
+    }
+    this->build_response_pmd(res, req);
+    decorate(res);
+    result = {};
+    return res;
+}
+
+//------------------------------------------------------------------------------
+
+// Attempt to read a complete frame header.
+// Returns `false` if more bytes are needed
+template<class NextLayer, bool deflateSupported>
+template<class DynamicBuffer>
+bool
+stream<NextLayer, deflateSupported>::impl_type::
+parse_fh(
+    detail::frame_header& fh,
+    DynamicBuffer& b,
+    error_code& ec)
+{
+    if(buffer_size(b.data()) < 2)
+    {
+        // need more bytes
+        ec = {};
+        return false;
+    }
+    buffers_suffix<typename
+        DynamicBuffer::const_buffers_type> cb{
+            b.data()};
+    std::size_t need;
+    {
+        std::uint8_t tmp[2];
+        cb.consume(net::buffer_copy(
+            net::buffer(tmp), cb));
+        fh.len = tmp[1] & 0x7f;
+        switch(fh.len)
+        {
+            case 126: need = 2; break;
+            case 127: need = 8; break;
+            default:
+                need = 0;
+        }
+        fh.mask = (tmp[1] & 0x80) != 0;
+        if(fh.mask)
+            need += 4;
+        if(buffer_size(cb) < need)
+        {
+            // need more bytes
+            ec = {};
+            return false;
+        }
+        fh.op   = static_cast<
+            detail::opcode>(tmp[0] & 0x0f);
+        fh.fin  = (tmp[0] & 0x80) != 0;
+        fh.rsv1 = (tmp[0] & 0x40) != 0;
+        fh.rsv2 = (tmp[0] & 0x20) != 0;
+        fh.rsv3 = (tmp[0] & 0x10) != 0;
+    }
+    switch(fh.op)
+    {
+    case detail::opcode::binary:
+    case detail::opcode::text:
+        if(rd_cont)
+        {
+            // new data frame when continuation expected
+            ec = error::bad_data_frame;
+            return false;
+        }
+        if(fh.rsv2 || fh.rsv3 ||
+            ! this->rd_deflated(fh.rsv1))
+        {
+            // reserved bits not cleared
+            ec = error::bad_reserved_bits;
+            return false;
+        }
+        break;
+
+    case detail::opcode::cont:
+        if(! rd_cont)
+        {
+            // continuation without an active message
+            ec = error::bad_continuation;
+            return false;
+        }
+        if(fh.rsv1 || fh.rsv2 || fh.rsv3)
+        {
+            // reserved bits not cleared
+            ec = error::bad_reserved_bits;
+            return false;
+        }
+        break;
+
+    default:
+        if(detail::is_reserved(fh.op))
+        {
+            // reserved opcode
+            ec = error::bad_opcode;
+            return false;
+        }
+        if(! fh.fin)
+        {
+            // fragmented control message
+            ec = error::bad_control_fragment;
+            return false;
+        }
+        if(fh.len > 125)
+        {
+            // invalid length for control message
+            ec = error::bad_control_size;
+            return false;
+        }
+        if(fh.rsv1 || fh.rsv2 || fh.rsv3)
+        {
+            // reserved bits not cleared
+            ec = error::bad_reserved_bits;
+            return false;
+        }
+        break;
+    }
+    if(role == role_type::server && ! fh.mask)
+    {
+        // unmasked frame from client
+        ec = error::bad_unmasked_frame;
+        return false;
+    }
+    if(role == role_type::client && fh.mask)
+    {
+        // masked frame from server
+        ec = error::bad_masked_frame;
+        return false;
+    }
+    if(detail::is_control(fh.op) &&
+        buffer_size(cb) < need + fh.len)
+    {
+        // Make the entire control frame payload
+        // get read in before we return `true`
+        return false;
+    }
+    switch(fh.len)
+    {
+    case 126:
+    {
+        std::uint8_t tmp[2];
+        BOOST_ASSERT(buffer_size(cb) >= sizeof(tmp));
+        cb.consume(net::buffer_copy(net::buffer(tmp), cb));
+        fh.len = detail::big_uint16_to_native(&tmp[0]);
+        if(fh.len < 126)
+        {
+            // length not canonical
+            ec = error::bad_size;
+            return false;
+        }
+        break;
+    }
+    case 127:
+    {
+        std::uint8_t tmp[8];
+        BOOST_ASSERT(buffer_size(cb) >= sizeof(tmp));
+        cb.consume(net::buffer_copy(net::buffer(tmp), cb));
+        fh.len = detail::big_uint64_to_native(&tmp[0]);
+        if(fh.len < 65536)
+        {
+            // length not canonical
+            ec = error::bad_size;
+            return false;
+        }
+        break;
+    }
+    }
+    if(fh.mask)
+    {
+        std::uint8_t tmp[4];
+        BOOST_ASSERT(buffer_size(cb) >= sizeof(tmp));
+        cb.consume(net::buffer_copy(net::buffer(tmp), cb));
+        fh.key = detail::little_uint32_to_native(&tmp[0]);
+        detail::prepare_key(rd_key, fh.key);
+    }
+    else
+    {
+        // initialize this otherwise operator== breaks
+        fh.key = 0;
+    }
+    if(! detail::is_control(fh.op))
+    {
+        if(fh.op != detail::opcode::cont)
+        {
+            rd_size = 0;
+            rd_op = fh.op;
+        }
+        else
+        {
+            if(rd_size > (std::numeric_limits<
+                std::uint64_t>::max)() - fh.len)
+            {
+                // message size exceeds configured limit
+                ec = error::message_too_big;
+                return false;
+            }
+        }
+        if(! this->rd_deflated())
+        {
+            if(rd_msg_max && beast::detail::sum_exceeds(
+                rd_size, fh.len, rd_msg_max))
+            {
+                // message size exceeds configured limit
+                ec = error::message_too_big;
+                return false;
+            }
+        }
+        rd_cont = ! fh.fin;
+        rd_remain = fh.len;
+    }
+    b.consume(b.size() - buffer_size(cb));
+    ec = {};
+    return true;
+}
+
+template<class NextLayer, bool deflateSupported>
+template<class DynamicBuffer>
+void
+stream<NextLayer, deflateSupported>::impl_type::
+write_ping(DynamicBuffer& db,
+    detail::opcode code, ping_data const& data)
+{
+    detail::frame_header fh;
+    fh.op = code;
+    fh.fin = true;
+    fh.rsv1 = false;
+    fh.rsv2 = false;
+    fh.rsv3 = false;
+    fh.len = data.size();
+    fh.mask = role == role_type::client;
+    if(fh.mask)
+        fh.key = create_mask();
+    detail::write(db, fh);
+    if(data.empty())
+        return;
+    detail::prepared_key key;
+    if(fh.mask)
+        detail::prepare_key(key, fh.key);
+    auto mb = db.prepare(data.size());
+    net::buffer_copy(mb,
+        net::const_buffer(
+            data.data(), data.size()));
+    if(fh.mask)
+        detail::mask_inplace(mb, key);
+    db.commit(data.size());
+}
+
+template<class NextLayer, bool deflateSupported>
+template<class DynamicBuffer>
+void
+stream<NextLayer, deflateSupported>::impl_type::
+write_close(DynamicBuffer& db, close_reason const& cr)
+{
+    using namespace boost::endian;
+    detail::frame_header fh;
+    fh.op = detail::opcode::close;
+    fh.fin = true;
+    fh.rsv1 = false;
+    fh.rsv2 = false;
+    fh.rsv3 = false;
+    fh.len = cr.code == close_code::none ?
+        0 : 2 + cr.reason.size();
+    if(role == role_type::client)
+    {
+        fh.mask = true;
+        fh.key = create_mask();
+    }
+    else
+    {
+        fh.mask = false;
+    }
+    detail::write(db, fh);
+    if(cr.code != close_code::none)
+    {
+        detail::prepared_key key;
+        if(fh.mask)
+            detail::prepare_key(key, fh.key);
+        {
+            std::uint8_t tmp[2];
+            ::new(&tmp[0]) big_uint16_buf_t{
+                (std::uint16_t)cr.code};
+            auto mb = db.prepare(2);
+            net::buffer_copy(mb,
+                net::buffer(tmp));
+            if(fh.mask)
+                detail::mask_inplace(mb, key);
+            db.commit(2);
+        }
+        if(! cr.reason.empty())
+        {
+            auto mb = db.prepare(cr.reason.size());
+            net::buffer_copy(mb,
+                net::const_buffer(
+                    cr.reason.data(), cr.reason.size()));
+            if(fh.mask)
+                detail::mask_inplace(mb, key);
+            db.commit(cr.reason.size());
+        }
+    }
+}
+
 } // websocket
 } // beast
 } // boost

include/boost/beast/websocket/impl/write.hpp

@@ -24,6 +24,7 @@
 #include <boost/beast/core/detail/clamp.hpp>
 #include <boost/beast/core/detail/config.hpp>
 #include <boost/beast/websocket/detail/frame.hpp>
+#include <boost/beast/websocket/impl/stream_impl.hpp>
 #include <boost/asio/coroutine.hpp>
 #include <boost/assert.hpp>
 #include <boost/config.hpp>
@@ -51,7 +52,7 @@ class stream<NextLayer, deflateSupported>::write_some_op
         do_deflate
     };
 
-    stream& ws_;
+    boost::weak_ptr<impl_type> wp_;
     buffers_suffix<Buffers> cb_;
     detail::frame_header fh_;
     detail::prepared_key key_;
@@ -69,17 +70,18 @@ public:
     template<class Handler_>
     write_some_op(
         Handler_&& h,
-        stream<NextLayer, deflateSupported>& ws,
+        boost::shared_ptr<impl_type> const& sp,
         bool fin,
         Buffers const& bs)
         : beast::async_op_base<Handler,
             beast::executor_type<stream>>(
-                std::forward<Handler_>(h), ws.get_executor())
-        , ws_(ws)
+                std::forward<Handler_>(h),
+                    sp->stream.get_executor())
+        , wp_(sp)
         , cb_(bs)
         , fin_(fin)
     {
-        auto& impl = *ws_.impl_;
+        auto& impl = *sp;
 
         // Set up the outgoing frame header
         if(! impl.wr_cont)
@@ -157,7 +159,11 @@ operator()(
     using beast::detail::clamp;
     std::size_t n;
     net::mutable_buffer b;
-    auto& impl = *ws_.impl_;
+    auto sp = wp_.lock();
+    if(! sp)
+        return this->invoke(cont,
+            net::error::operation_aborted, 0);
+    auto& impl = *sp;
     BOOST_ASIO_CORO_REENTER(*this)
     {
         // Acquire the write lock
@@ -165,7 +171,7 @@ operator()(
         {
         do_suspend:
             BOOST_ASIO_CORO_YIELD
-            impl.paused_wr.emplace(std::move(*this));
+            impl.op_wr.emplace(std::move(*this));
             impl.wr_block.lock(this);
             BOOST_ASIO_CORO_YIELD
             net::post(std::move(*this));
@@ -228,9 +234,10 @@ operator()(
                 // Give up the write lock in between each frame
                 // so that outgoing control frames might be sent.
                 impl.wr_block.unlock(this);
-                if( impl.paused_close.maybe_invoke() ||
-                    impl.paused_rd.maybe_invoke() ||
-                    impl.paused_ping.maybe_invoke())
+                if( impl.op_close.maybe_invoke()
+                    || impl.op_idle_ping.maybe_invoke()
+                    || impl.op_rd.maybe_invoke()
+                    || impl.op_ping.maybe_invoke())
                 {
                     BOOST_ASSERT(impl.wr_block.is_locked());
                     goto do_suspend;
@@ -248,7 +255,7 @@ operator()(
             remain_ = beast::buffer_size(cb_);
             fh_.fin = fin_;
             fh_.len = remain_;
-            fh_.key = ws_.create_mask();
+            fh_.key = impl.create_mask();
             detail::prepare_key(key_, fh_.key);
             impl.wr_fb.clear();
             detail::write<flat_static_buffer_base>(
@@ -302,7 +309,7 @@ operator()(
                 n = clamp(remain_, impl.wr_buf_size);
                 remain_ -= n;
                 fh_.len = n;
-                fh_.key = ws_.create_mask();
+                fh_.key = impl.create_mask();
                 fh_.fin = fin_ ? remain_ == 0 : false;
                 detail::prepare_key(key_, fh_.key);
                 net::buffer_copy(net::buffer(
@@ -330,9 +337,10 @@ operator()(
                 // Give up the write lock in between each frame
                 // so that outgoing control frames might be sent.
                 impl.wr_block.unlock(this);
-                if( impl.paused_close.maybe_invoke() ||
-                    impl.paused_rd.maybe_invoke() ||
-                    impl.paused_ping.maybe_invoke())
+                if( impl.op_close.maybe_invoke()
+                    || impl.op_idle_ping.maybe_invoke()
+                    || impl.op_rd.maybe_invoke()
+                    || impl.op_ping.maybe_invoke())
                 {
                     BOOST_ASSERT(impl.wr_block.is_locked());
                     goto do_suspend;
@@ -365,7 +373,7 @@ operator()(
                 }
                 if(fh_.mask)
                 {
-                    fh_.key = ws_.create_mask();
+                    fh_.key = impl.create_mask();
                     detail::prepared_key key;
                     detail::prepare_key(key, fh_.key);
                     detail::mask_inplace(b, key);
@@ -391,9 +399,10 @@ operator()(
                     // Give up the write lock in between each frame
                     // so that outgoing control frames might be sent.
                     impl.wr_block.unlock(this);
-                    if( impl.paused_close.maybe_invoke() ||
-                        impl.paused_rd.maybe_invoke() ||
-                        impl.paused_ping.maybe_invoke())
+                    if( impl.op_close.maybe_invoke()
+                        || impl.op_idle_ping.maybe_invoke()
+                        || impl.op_rd.maybe_invoke()
+                        || impl.op_ping.maybe_invoke())
                     {
                         BOOST_ASSERT(impl.wr_block.is_locked());
                         goto do_suspend;
@@ -413,15 +422,10 @@ operator()(
 
     upcall:
         impl.wr_block.unlock(this);
-        impl.paused_close.maybe_invoke()
-            || impl.paused_rd.maybe_invoke()
-            || impl.paused_ping.maybe_invoke();
-        if(! cont)
-        {
-            BOOST_ASIO_CORO_YIELD
-            net::post(bind_front_handler(
-                std::move(*this), ec, bytes_transferred_));
-        }
+        impl.op_close.maybe_invoke()
+            || impl.op_idle_ping.maybe_invoke()
+            || impl.op_rd.maybe_invoke()
+            || impl.op_ping.maybe_invoke();
         this->invoke(cont, ec, bytes_transferred_);
     }
 }
@@ -507,7 +511,7 @@ write_some(bool fin,
             }
             if(fh.mask)
             {
-                fh.key = this->create_mask();
+                fh.key = this->impl_->create_mask();
                 detail::prepared_key key;
                 detail::prepare_key(key, fh.key);
                 detail::mask_inplace(b, key);
@@ -581,7 +585,7 @@ write_some(bool fin,
         // mask, no autofrag
         fh.fin = fin;
         fh.len = remain;
-        fh.key = this->create_mask();
+        fh.key = this->impl_->create_mask();
         detail::prepared_key key;
         detail::prepare_key(key, fh.key);
         detail::fh_buffer fh_buf;
@@ -629,7 +633,7 @@ write_some(bool fin,
             ConstBufferSequence> cb(buffers);
         for(;;)
         {
-            fh.key = this->create_mask();
+            fh.key = this->impl_->create_mask();
             detail::prepared_key key;
             detail::prepare_key(key, fh.key);
             auto const n =
@@ -676,7 +680,7 @@ async_write_some(bool fin,
         WriteHandler, void(error_code, std::size_t));
     write_some_op<ConstBufferSequence, BOOST_ASIO_HANDLER_TYPE(
         WriteHandler, void(error_code, std::size_t))>(
-            std::move(init.completion_handler), *this, fin, bs);
+            std::move(init.completion_handler), impl_, fin, bs);
     return init.result.get();
 }
 
@@ -731,7 +735,7 @@ async_write(
         WriteHandler, void(error_code, std::size_t));
     write_some_op<ConstBufferSequence, BOOST_ASIO_HANDLER_TYPE(
         WriteHandler, void(error_code, std::size_t))>(
-            std::move(init.completion_handler), *this, true, bs);
+            std::move(init.completion_handler), impl_, true, bs);
     return init.result.get();
 }
 

include/boost/beast/websocket/stream.hpp

@@ -15,15 +15,17 @@
 #include <boost/beast/websocket/option.hpp>
 #include <boost/beast/websocket/role.hpp>
 #include <boost/beast/websocket/rfc6455.hpp>
+#include <boost/beast/websocket/stream_base.hpp>
 #include <boost/beast/websocket/stream_fwd.hpp>
-#include <boost/beast/websocket/detail/pmd_extension.hpp>
-#include <boost/beast/websocket/detail/stream_base.hpp>
 #include <boost/beast/websocket/detail/hybi13.hpp>
+#include <boost/beast/websocket/detail/impl_base.hpp>
+#include <boost/beast/websocket/detail/pmd_extension.hpp>
 #include <boost/beast/core/string.hpp>
 #include <boost/beast/core/detail/type_traits.hpp>
 #include <boost/beast/http/detail/type_traits.hpp>
 #include <boost/asio/async_result.hpp>
 #include <boost/asio/error.hpp>
+#include <boost/shared_ptr.hpp>
 #include <algorithm>
 #include <cstdint>
 #include <functional>
@@ -81,17 +83,16 @@ class frame_test;
     operations are performed within the same implicit or explicit strand.
 
     @par Example
-
-    To use the @ref stream template with an `ip::tcp::socket`,
-    you would write:
-
+    To declare the @ref stream object with a @ref tcp_stream in a
+    multi-threaded asynchronous program using a strand, you may write:
     @code
-    websocket::stream<ip::tcp::socket> ws{io_context};
+    websocket::stream<tcp_stream<
+        net::io_context::strand>> ws{net::io_context::strand(ioc)};
     @endcode
-    Alternatively, you can write:
+    Alternatively, for a single-threaded or synchronous application
+    you may write:
     @code
-    ip::tcp::socket sock{io_context};
-    websocket::stream<ip::tcp::socket&> ws{sock};
+    websocket::stream<tcp_stream<net::io_context::executor_type>> ws(ioc);
     @endcode
 
     @tparam NextLayer The type representing the next layer, to which
@@ -130,14 +131,14 @@ template<
     bool deflateSupported>
 class stream
 #if ! BOOST_BEAST_DOXYGEN
-    : private detail::stream_base
+    : private stream_base
 #endif
 {
     friend class close_test;
     friend class frame_test;
     friend class ping_test;
-    friend class read1_test;
     friend class read2_test;
+    friend class read3_test;
     friend class stream_test;
     friend class write_test;
 
@@ -153,7 +154,7 @@ class stream
 
     struct impl_type;
 
-    std::shared_ptr<impl_type> impl_;
+    boost::shared_ptr<impl_type> impl_;
 
     using time_point = typename
         std::chrono::steady_clock::time_point;
@@ -351,6 +352,20 @@ public:
     //
     //--------------------------------------------------------------------------
 
+#if BOOST_BEAST_DOXYGEN
+    template<class Option>
+    void
+    get_option(Option& opt);
+
+    template<class Option>
+    void
+    set_option(Option const& opt);
+#else
+
+    void get_option(timeout& opt);
+    void set_option(timeout const& opt);
+#endif
+
     /** Set the permessage-deflate extension options
 
         @throws invalid_argument if `deflateSupported == false`, and either
@@ -3478,6 +3493,7 @@ private:
     template<class>         class close_op;
     template<class>         class handshake_op;
     template<class>         class ping_op;
+    template<class>         class auto_ping_op;
     template<class, class>  class read_some_op;
     template<class, class>  class read_op;
     template<class>         class response_op;
@@ -3487,55 +3503,14 @@ private:
     static void default_decorate_req(request_type&) {}
     static void default_decorate_res(response_type&) {}
 
-    template<class DynamicBuffer>
-    bool
-    parse_fh(
-        detail::frame_header& fh,
-        DynamicBuffer& b,
-        error_code& ec);
-
-    template<class DynamicBuffer>
-    void
-    write_close(DynamicBuffer& b, close_reason const& rc);
-
-    template<class DynamicBuffer>
-    void
-    write_ping(DynamicBuffer& b,
-        detail::opcode op, ping_data const& data);
-
-    //
-    // upgrade
-    //
-
-    template<class Decorator>
-    request_type
-    build_request(detail::sec_ws_key_type& key,
-        string_view host,
-            string_view target,
-                Decorator const& decorator);
-
-    template<
-        class Body, class Allocator, class Decorator>
-    response_type
-    build_response(
-        http::request<Body,
-            http::basic_fields<Allocator>> const& req,
-        Decorator const& decorator,
-        error_code& ec);
-
-    void
-    on_response(
-        response_type const& res,
-        detail::sec_ws_key_type const& key,
-        error_code& ec);
-
     //
     // accept / handshake
     //
 
-    template<class Decorator>
+    template<class Buffers, class Decorator>
     void
     do_accept(
+        Buffers const& buffers,
         Decorator const& decorator,
         error_code& ec);
 

include/boost/beast/websocket/stream_base.hpp

@@ -0,0 +1,139 @@
+//
+// 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_STREAM_BASE_HPP
+#define BOOST_BEAST_WEBSOCKET_STREAM_BASE_HPP
+
+#include <boost/beast/core/detail/config.hpp>
+#include <boost/beast/websocket/role.hpp>
+#include <chrono>
+
+namespace boost {
+namespace beast {
+namespace websocket {
+
+/** This class is used as a base for the @ref websocket::stream class template to group common types and constants.
+*/
+struct stream_base
+{
+    /// The type used to represent durations
+    using duration =
+        std::chrono::steady_clock::duration;
+
+    /// The type used to represent time points
+    using time_point =
+        std::chrono::steady_clock::time_point;
+
+    /// Returns the special time_point value meaning "never"
+    static
+    constexpr
+    time_point
+    never() noexcept
+    {
+        return (time_point::max)();
+    }
+
+    /// Returns the special duration value meaning "none"
+    static
+    constexpr
+    duration
+    none() noexcept
+    {
+        return (duration::max)();
+    }
+
+    /** Stream option to control the behavior of websocket timeouts.
+
+        Timeout features are available for asynchronous operations only.
+    */
+    struct timeout
+    {
+        /** Time limit on handshake, accept, and close operations:
+
+            This value whether or not there is a time limit, and the
+            duration of that time limit, for asynchronous handshake,
+            accept, and close operations. If this is equal to the
+            value @ref none then there will be no time limit. Otherwise,
+            if any of the applicable operations takes longer than this
+            amount of time, the operation will be canceled and a
+            timeout error delivered to the completion handler.
+        */
+        duration handshake_timeout = none();
+
+        /** The time limit after which a connection is considered idle.
+        */
+        duration idle_timeout = none();
+
+        /** Automatic ping setting.
+
+            If the idle interval is set, this setting affects the
+            behavior of the stream when no data is received for the
+            timeout interval as follows:
+
+            @li When `keep_alive_pings` is `true`, an idle ping will be
+            sent automatically. If another timeout interval elapses
+            with no received data then the connection will be closed.
+            An outstanding read operation must be pending, which will
+            complete immediately the error @ref beast::error::timeout.
+
+            @li When `keep_alive_pings` is `false`, the connection will be closed.
+            An outstanding read operation must be pending, which will
+            complete immediately the error @ref beast::error::timeout.
+        */
+        bool keep_alive_pings = false;
+    };
+
+    /** Construct timeout settings with suggested values for a role.
+
+        This constructs the timeout settings with a predefined set
+        of values which varies depending on the desired role. The
+        values are selected upon construction, regardless of the
+        current or actual role in use on the stream.
+
+        @param role The role of the websocket stream
+        (@ref role_type::client or @ref role_type::server).
+    */
+    static
+    timeout
+    suggested_settings(role_type role) noexcept
+    {
+        timeout opt;
+        switch(role)
+        {
+        case role_type::client:
+            opt.handshake_timeout = std::chrono::seconds(30);
+            opt.idle_timeout = none();
+            opt.keep_alive_pings = false;
+            break;
+
+        case role_type::server:
+            opt.handshake_timeout = std::chrono::seconds(30);
+            opt.idle_timeout = std::chrono::seconds(300);
+            opt.keep_alive_pings = true;
+            break;
+        }
+    }
+
+protected:
+    enum class status
+    {
+        //none,
+        handshake,
+        open,
+        closing,
+        closed,
+        failed // VFALCO Is this needed?
+    };
+};
+
+} // websocket
+} // beast
+} // boost
+
+#endif