Merge subtree Beast 1.0.0-b4:

Merge commit 'c0952e54db7bd519440dc0611db7347cb048296d' into

src/beast/test/websocket/stream.cpp

@@ -11,158 +11,33 @@
 #include "websocket_async_echo_peer.hpp"
 #include "websocket_sync_echo_peer.hpp"
 
-#include <beast/bind_handler.hpp>
-#include <beast/streambuf.hpp>
-#include <beast/detail/unit_test/suite.hpp>
+#include <beast/core/streambuf.hpp>
+#include <beast/core/to_string.hpp>
+#include <beast/test/fail_stream.hpp>
+#include <beast/test/string_stream.hpp>
+#include <beast/test/yield_to.hpp>
+#include <beast/unit_test/suite.hpp>
 #include <boost/asio.hpp>
 #include <boost/asio/spawn.hpp>
 #include <boost/optional.hpp>
-#include <condition_variable>
-#include <functional>
-#include <mutex>
-#include <thread>
-
-#include <beast/http/parser_v1.hpp>
 
 namespace beast {
 namespace websocket {
 
-class stream_test : public beast::detail::unit_test::suite
+class stream_test
+    : public beast::unit_test::suite
+    , public test::enable_yield_to
 {
-    boost::asio::io_service ios_;
-    boost::optional<boost::asio::io_service::work> work_;
-    std::thread thread_;
-    std::mutex m_;
-    std::condition_variable cv_;
-    bool running_ = false;;
-
 public:
     using endpoint_type = boost::asio::ip::tcp::endpoint;
     using address_type = boost::asio::ip::address;
     using socket_type = boost::asio::ip::tcp::socket;
 
-    // meets the requirements of AsyncStream, SyncStream
-    class string_Stream
+    void testClamp()
     {
-        std::string s_;
-        boost::asio::io_service& ios_;
-
-    public:
-        string_Stream(boost::asio::io_service& ios,
-                std::string s)
-            : s_(s)
-            , ios_(ios)
-        {
-        }
-
-        boost::asio::io_service&
-        get_io_service()
-        {
-            return ios_;
-        }
-
-        template<class MutableBufferSequence>
-        std::size_t
-        read_some(MutableBufferSequence const& buffers)
-        {
-            error_code ec;
-            auto const n = read_some(buffers, ec);
-            if(ec)
-                throw boost::system::system_error{ec};
-            return n;
-        }
-
-        template<class MutableBufferSequence>
-        std::size_t
-        read_some(MutableBufferSequence const& buffers,
-            error_code& ec)
-        {
-            auto const n = boost::asio::buffer_copy(
-                buffers, boost::asio::buffer(s_));
-            s_.erase(0, n);
-            return n;
-        }
-
-        template<class MutableBufferSequence, class ReadHandler>
-        typename async_completion<ReadHandler,
-            void(error_code, std::size_t)>::result_type
-        async_read_some(MutableBufferSequence const& buffers,
-            ReadHandler&& handler)
-        {
-            auto const n = boost::asio::buffer_copy(
-                buffers, boost::asio::buffer(s_));
-            s_.erase(0, n);
-            async_completion<ReadHandler,
-                void(error_code, std::size_t)> completion(handler);
-            ios_.post(bind_handler(
-                completion.handler, error_code{}, n));
-            return completion.result.get();
-        }
-
-        template<class ConstBufferSequence>
-        std::size_t
-        write_some(ConstBufferSequence const& buffers)
-        {
-            error_code ec;
-            auto const n = write_some(buffers, ec);
-            if(ec)
-                throw boost::system::system_error{ec};
-            return n;
-        }
-
-        template<class ConstBufferSequence>
-        std::size_t
-        write_some(ConstBufferSequence const& buffers,
-            error_code&)
-        {
-            return boost::asio::buffer_size(buffers);
-        }
-
-        template<class ConstBuffeSequence, class WriteHandler>
-        typename async_completion<WriteHandler,
-            void(error_code, std::size_t)>::result_type
-        async_write_some(ConstBuffeSequence const& buffers,
-            WriteHandler&& handler)
-        {
-            async_completion<WriteHandler,
-                void(error_code, std::size_t)> completion(handler);
-            ios_.post(bind_handler(completion.handler,
-                error_code{}, boost::asio::buffer_size(buffers)));
-            return completion.result.get();
-        }
-    };
-
-    stream_test()
-        : work_(ios_)
-        , thread_([&]{ ios_.run(); })
-    {
-    }
-
-    ~stream_test()
-    {
-        work_ = boost::none;
-        thread_.join();
-    }
-
-    template<class Function>
-    void exec(Function&& f)
-    {
-        {
-            std::lock_guard<std::mutex> lock(m_);
-            running_ = true;
-        }
-        boost::asio::spawn(ios_,
-            [&](boost::asio::yield_context do_yield)
-            {
-                f(do_yield);
-                std::lock_guard<std::mutex> lock(m_);
-                running_ = false;
-                cv_.notify_all();
-            }
-            , boost::coroutines::attributes(2 * 1024 * 1024));
-
-        std::unique_lock<std::mutex> lock(m_);
-        cv_.wait(lock, [&]{ return ! running_; });
+        expect(detail::clamp(
+            std::numeric_limits<std::uint64_t>::max()) ==
+                std::numeric_limits<std::size_t>::max());
     }
 
     void testSpecialMembers()
@@ -175,6 +50,7 @@ public:
             stream<socket_type> ws2(ios_);
             ws = std::move(ws2);
         }
+        expect(&ws.get_io_service() == &ios_);
         pass();
     }
 
@@ -185,7 +61,15 @@ public:
         ws.set_option(read_buffer_size(8192));
         ws.set_option(read_message_max(1 * 1024 * 1024));
         ws.set_option(write_buffer_size(2048));
-        pass();
+        try
+        {
+            ws.set_option(message_type(opcode::close));
+            fail();
+        }
+        catch(std::exception const&)
+        {
+            pass();
+        }
     }
 
     template<std::size_t N>
@@ -196,10 +80,10 @@ public:
         return boost::asio::const_buffers_1(&s[0], N-1);
     }
 
-    void testAccept(boost::asio::yield_context do_yield)
+    void testAccept(yield_context do_yield)
     {
         {
-            stream<string_Stream> ws(ios_,
+            stream<test::string_stream> ws(ios_,
                 "GET / HTTP/1.1\r\n"
                 "Host: localhost:80\r\n"
                 "Upgrade: WebSocket\r\n"
@@ -218,7 +102,7 @@ public:
             }
         }
         {
-            stream<string_Stream> ws(ios_,
+            stream<test::string_stream> ws(ios_,
                 "GET / HTTP/1.1\r\n"
                 "\r\n");
             try
@@ -232,7 +116,7 @@ public:
             }
         }
         {
-            stream<string_Stream> ws(ios_,
+            stream<test::string_stream> ws(ios_,
                 "GET / HTTP/1.1\r\n"
                 "Host: localhost:80\r\n"
                 "Upgrade: WebSocket\r\n"
@@ -245,7 +129,7 @@ public:
             expect(! ec, ec.message());
         }
         {
-            stream<string_Stream> ws(ios_,
+            stream<test::string_stream> ws(ios_,
                 "GET / HTTP/1.1\r\n"
                 "\r\n");
             error_code ec;
@@ -253,7 +137,7 @@ public:
             expect(ec);
         }
         {
-            stream<string_Stream> ws(ios_,
+            stream<test::string_stream> ws(ios_,
                 "GET / HTTP/1.1\r\n"
                 "Host: localhost:80\r\n"
                 "Upgrade: WebSocket\r\n"
@@ -266,7 +150,7 @@ public:
             expect(! ec, ec.message());
         }
         {
-            stream<string_Stream> ws(ios_,
+            stream<test::string_stream> ws(ios_,
                 "GET / HTTP/1.1\r\n"
                 "\r\n");
             error_code ec;
@@ -274,7 +158,7 @@ public:
             expect(ec);
         }
         {
-            stream<string_Stream> ws(ios_,
+            stream<test::string_stream> ws(ios_,
                 "Host: localhost:80\r\n"
                 "Upgrade: WebSocket\r\n"
                 "Connection: upgrade\r\n"
@@ -293,7 +177,7 @@ public:
             }
         }
         {
-            stream<string_Stream> ws(ios_,
+            stream<test::string_stream> ws(ios_,
                 "\r\n");
             try
             {
@@ -307,7 +191,7 @@ public:
             }
         }
         {
-            stream<string_Stream> ws(ios_,
+            stream<test::string_stream> ws(ios_,
                 "Host: localhost:80\r\n"
                 "Upgrade: WebSocket\r\n"
                 "Connection: upgrade\r\n"
@@ -320,7 +204,7 @@ public:
             expect(! ec, ec.message());
         }
         {
-            stream<string_Stream> ws(ios_,
+            stream<test::string_stream> ws(ios_,
                 "GET / HTTP/1.1\r\n"
                 "\r\n");
             error_code ec;
@@ -328,7 +212,7 @@ public:
             expect(ec);
         }
         {
-            stream<string_Stream> ws(ios_,
+            stream<test::string_stream> ws(ios_,
                 "Host: localhost:80\r\n"
                 "Upgrade: WebSocket\r\n"
                 "Connection: upgrade\r\n"
@@ -341,7 +225,7 @@ public:
             expect(! ec, ec.message());
         }
         {
-            stream<string_Stream> ws(ios_,
+            stream<test::string_stream> ws(ios_,
                 "\r\n");
             error_code ec;
             ws.async_accept(strbuf(
@@ -351,7 +235,7 @@ public:
     }
 
     void testHandshake(endpoint_type const& ep,
-        boost::asio::yield_context do_yield)
+        yield_context do_yield)
     {
         {
             // disconnected socket
@@ -420,28 +304,331 @@ public:
         }
     }
 
+    void testErrorHandling(endpoint_type const& ep,
+        yield_context do_yield)
+    {
+        static std::size_t constexpr limit = 100;
+        std::size_t n;
+
+        // synchronous, exceptions
+        for(n = 0; n < limit; ++n)
+        {
+            error_code ec;
+            socket_type sock(ios_);
+            sock.connect(ep, ec);
+            if(! expect(! ec, ec.message()))
+                break;
+            stream<test::fail_stream<socket_type&>> ws(n, sock);
+            try
+            {
+                ws.handshake("localhost", "/");
+                ws.write(boost::asio::const_buffers_1(
+                    "Hello", 5));
+                opcode op;
+                streambuf sb;
+                ws.read(op, sb);
+                expect(op == opcode::text);
+                expect(to_string(sb.data()) == "Hello");
+                ws.close({});
+                try
+                {
+                    ws.read(op, sb);
+                }
+                catch(boost::system::system_error const& se)
+                {
+                    if(se.code() == error::closed)
+                        break;
+                    throw;
+                }
+                fail();
+                break;
+            }
+            catch(boost::system::system_error const&)
+            {
+            }
+        }
+        expect(n < limit);
+
+        // synchronous, error codes
+        for(n = 0; n < limit; ++n)
+        {
+            error_code ec;
+            socket_type sock(ios_);
+            sock.connect(ep, ec);
+            if(! expect(! ec, ec.message()))
+                break;
+            stream<test::fail_stream<socket_type&>> ws(n, sock);
+            ws.handshake("localhost", "/", ec);
+            if(ec)
+                continue;
+            ws.write(boost::asio::const_buffers_1(
+                "Hello", 5), ec);
+            if(ec)
+                continue;
+            opcode op;
+            streambuf sb;
+            ws.read(op, sb, ec);
+            if(ec)
+                continue;
+            expect(op == opcode::text);
+            expect(to_string(sb.data()) == "Hello");
+            ws.close({}, ec);
+            if(ec)
+                continue;
+            ws.read(op, sb, ec);
+            if(ec == error::closed)
+            {
+                pass();
+                break;
+            }
+        }
+        expect(n < limit);
+
+        // asynchronous
+        for(n = 0; n < limit; ++n)
+        {
+            error_code ec;
+            socket_type sock(ios_);
+            sock.connect(ep, ec);
+            if(! expect(! ec, ec.message()))
+                break;
+            stream<test::fail_stream<socket_type&>> ws(n, sock);
+            ws.async_handshake("localhost", "/", do_yield[ec]);
+            if(ec)
+                break;
+            ws.async_write(boost::asio::const_buffers_1(
+                "Hello", 5), do_yield[ec]);
+            if(ec)
+                continue;
+            opcode op;
+            streambuf sb;
+            ws.async_read(op, sb, do_yield[ec]);
+            if(ec)
+                continue;
+            expect(op == opcode::text);
+            expect(to_string(sb.data()) == "Hello");
+            ws.async_close({}, do_yield[ec]);
+            if(ec)
+                continue;
+            ws.async_read(op, sb, do_yield[ec]);
+            if(ec == error::closed)
+            {
+                pass();
+                break;
+            }
+        }
+        expect(n < limit);
+    }
+
+    void testMask(endpoint_type const& ep,
+        yield_context do_yield)
+    {
+        {
+            std::vector<char> v;
+            for(char n = 0; n < 20; ++n)
+            {
+                error_code ec;
+                socket_type sock(ios_);
+                sock.connect(ep, ec);
+                if(! expect(! ec, ec.message()))
+                    break;
+                stream<socket_type&> ws(sock);
+                ws.handshake("localhost", "/", ec);
+                if(! expect(! ec, ec.message()))
+                    break;
+                ws.write(boost::asio::buffer(v), ec);
+                if(! expect(! ec, ec.message()))
+                    break;
+                opcode op;
+                streambuf sb;
+                ws.read(op, sb, ec);
+                if(! expect(! ec, ec.message()))
+                    break;
+                expect(to_string(sb.data()) ==
+                    std::string{v.data(), v.size()});
+                v.push_back(n+1);
+            }
+        }
+        {
+            std::vector<char> v;
+            for(char n = 0; n < 20; ++n)
+            {
+                error_code ec;
+                socket_type sock(ios_);
+                sock.connect(ep, ec);
+                if(! expect(! ec, ec.message()))
+                    break;
+                stream<socket_type&> ws(sock);
+                ws.handshake("localhost", "/", ec);
+                if(! expect(! ec, ec.message()))
+                    break;
+                ws.async_write(boost::asio::buffer(v), do_yield[ec]);
+                if(! expect(! ec, ec.message()))
+                    break;
+                opcode op;
+                streambuf sb;
+                ws.async_read(op, sb, do_yield[ec]);
+                if(! expect(! ec, ec.message()))
+                    break;
+                expect(to_string(sb.data()) ==
+                    std::string{v.data(), v.size()});
+                v.push_back(n+1);
+            }
+        }
+    }
+
+    struct con
+    {
+        stream<socket_type> ws;
+
+        con(endpoint_type const& ep, boost::asio::io_service& ios)
+            : ws(ios)
+        {
+            ws.next_layer().connect(ep);
+            ws.handshake("localhost", "/");
+        }
+    };
+
+    template<std::size_t N>
+    class cbuf_helper
+    {
+        std::array<std::uint8_t, N> v_;
+        boost::asio::const_buffer cb_;
+
+    public:
+        using value_type = decltype(cb_);
+        using const_iterator = value_type const*;
+
+        template<class... Vn>
+        explicit
+        cbuf_helper(Vn... vn)
+            : v_({{ static_cast<std::uint8_t>(vn)... }})
+            , cb_(v_.data(), v_.size())
+        {
+        }
+
+        const_iterator
+        begin() const
+        {
+            return &cb_;
+        }
+
+        const_iterator
+        end() const
+        {
+            return begin()+1;
+        }
+    };
+
+    template<class... Vn>
+    cbuf_helper<sizeof...(Vn)>
+    cbuf(Vn... vn)
+    {
+        return cbuf_helper<sizeof...(Vn)>(vn...);
+    }
+
+    void testClose(endpoint_type const& ep, yield_context do_yield)
+    {
+        using boost::asio::buffer;
+        {
+            // payload length 1
+            con c(ep, ios_);
+            boost::asio::write(c.ws.next_layer(),
+                cbuf(0x88, 0x81, 0xff, 0xff, 0xff, 0xff, 0x00));
+        }
+        {
+            // invalid close code 1005
+            con c(ep, ios_);
+            boost::asio::write(c.ws.next_layer(),
+                cbuf(0x88, 0x82, 0xff, 0xff, 0xff, 0xff, 0xfc, 0x12));
+        }
+        {
+            // invalid utf8
+            con c(ep, ios_);
+            boost::asio::write(c.ws.next_layer(),
+                cbuf(0x88, 0x86, 0xff, 0xff, 0xff, 0xff, 0xfc, 0x15,
+                    0x0f, 0xd7, 0x73, 0x43));
+        }
+        {
+            // good utf8
+            con c(ep, ios_);
+            boost::asio::write(c.ws.next_layer(),
+                cbuf(0x88, 0x86, 0xff, 0xff, 0xff, 0xff, 0xfc, 0x15,
+                    'u', 't', 'f', '8'));
+        }
+    }
+
+    void testWriteFrame(endpoint_type const& ep)
+    {
+        for(;;)
+        {
+            boost::asio::io_service ios;
+            error_code ec;
+            socket_type sock(ios);
+            sock.connect(ep, ec);
+            if(! expect(! ec, ec.message()))
+                break;
+            stream<socket_type&> ws(sock);
+            ws.handshake("localhost", "/", ec);
+            if(! expect(! ec, ec.message()))
+                break;
+            ws.async_write_frame(false,
+                boost::asio::null_buffers{},
+                    [](error_code){ });
+            //
+            // Destruction of the io_service will cause destruction
+            // of the write_frame_op without invoking the final handler.
+            //
+            break;
+        }
+    }
+
     void run() override
     {
+        testClamp();
+
         testSpecialMembers();
 
         testOptions();
 
-        exec(std::bind(&stream_test::testAccept,
+        yield_to(std::bind(&stream_test::testAccept,
             this, std::placeholders::_1));
 
         auto const any = endpoint_type{
             address_type::from_string("127.0.0.1"), 0};
         {
             sync_echo_peer server(true, any);
-            exec(std::bind(&stream_test::testHandshake,
-                this, server.local_endpoint(),
-                    std::placeholders::_1));
+            auto const ep = server.local_endpoint();
+
+            yield_to(std::bind(&stream_test::testHandshake,
+                this, ep, std::placeholders::_1));
+
+            yield_to(std::bind(&stream_test::testErrorHandling,
+                this, ep, std::placeholders::_1));
+
+            yield_to(std::bind(&stream_test::testMask,
+                this, ep, std::placeholders::_1));
+
+            yield_to(std::bind(&stream_test::testClose,
+                this, ep, std::placeholders::_1));
+
+            testWriteFrame(ep);
         }
         {
             async_echo_peer server(true, any, 1);
-            exec(std::bind(&stream_test::testHandshake,
-                this, server.local_endpoint(),
-                    std::placeholders::_1));
+            auto const ep = server.local_endpoint();
+
+            yield_to(std::bind(&stream_test::testHandshake,
+                this, ep, std::placeholders::_1));
+
+            yield_to(std::bind(&stream_test::testErrorHandling,
+                this, ep, std::placeholders::_1));
+
+            yield_to(std::bind(&stream_test::testMask,
+                this, ep, std::placeholders::_1));
+
+            yield_to(std::bind(&stream_test::testClose,
+                this, ep, std::placeholders::_1));
         }
 
         pass();