[FOLD]

Author: vinniefalco

CMakeLists.txt

@@ -60,6 +60,7 @@ set(BOOST_WS_IO_DEPENDENCIES
     Boost::system
     Boost::url
     Boost::ws_proto
+    Boost::http_io
     )
 
 foreach (BOOST_WS_IO_DEPENDENCY ${BOOST_WS_IO_DEPENDENCIES})

test/unit/server.cpp

@@ -7,8 +7,6 @@
 // Official repository: https://github.com/boostorg/beast
 //
 
-#include "server.hpp"
-
 #if 0
 
 #include <boost/beast/core.hpp>
@@ -61,7 +59,7 @@ class session : public std::enable_shared_from_this<session>
         // on the I/O objects in this session. Although not strictly necessary
         // for single-threaded contexts, this example code is written to be
         // thread-safe by default.
-        net::dispatch(ws_.get_executor(),
+        asio::dispatch(ws_.get_executor(),
             beast::bind_front_handler(
                 &session::on_run,
                 shared_from_this()));
@@ -158,12 +156,12 @@ class session : public std::enable_shared_from_this<session>
 // Accepts incoming connections and launches the sessions
 class listener : public std::enable_shared_from_this<listener>
 {
-    net::io_context& ioc_;
+    asio::io_context& ioc_;
     tcp::acceptor acceptor_;
 
 public:
     listener(
-        net::io_context& ioc,
+        asio::io_context& ioc,
         tcp::endpoint endpoint)
         : ioc_(ioc)
         , acceptor_(ioc)
@@ -179,7 +177,7 @@ class listener : public std::enable_shared_from_this<listener>
         }
 
         // Allow address reuse
-        acceptor_.set_option(net::socket_base::reuse_address(true), ec);
+        acceptor_.set_option(asio::socket_base::reuse_address(true), ec);
         if(ec)
         {
             fail(ec, "set_option");
@@ -196,7 +194,7 @@ class listener : public std::enable_shared_from_this<listener>
 
         // Start listening for connections
         acceptor_.listen(
-            net::socket_base::max_listen_connections, ec);
+            asio::socket_base::max_listen_connections, ec);
         if(ec)
         {
             fail(ec, "listen");
@@ -217,7 +215,7 @@ class listener : public std::enable_shared_from_this<listener>
     {
         // The new connection gets its own strand
         acceptor_.async_accept(
-            net::make_strand(ioc_),
+            asio::make_strand(ioc_),
             beast::bind_front_handler(
                 &listener::on_accept,
                 shared_from_this()));
@@ -254,12 +252,12 @@ int main(int argc, char* argv[])
             "    websocket-server-async 0.0.0.0 8080 1\n";
         return EXIT_FAILURE;
     }
-    auto const address = net::ip::make_address(argv[1]);
+    auto const address = asio::ip::make_address(argv[1]);
     auto const port = static_cast<unsigned short>(std::atoi(argv[2]));
     auto const threads = std::max<int>(1, std::atoi(argv[3]));
 
     // The io_context is required for all I/O
-    net::io_context ioc{threads};
+    asio::io_context ioc{threads};
 
     // Create and launch a listening port
     std::make_shared<listener>(ioc, tcp::endpoint{address, port})->run();
@@ -280,35 +278,152 @@ int main(int argc, char* argv[])
 
 #endif
 
+#include "test/unit/server.hpp"
 #include <boost/asio/io_context.hpp>
 #include <boost/asio/executor_work_guard.hpp>
+#include <boost/beast/websocket/stream.hpp>
+#include <boost/assert.hpp>
+#include <boost/smart_ptr/enable_shared_from.hpp>
+#include <memory>
 #include <thread>
+#include <utility>
+
+#include "test_suite.hpp"
 
 namespace boost {
 namespace ws_io {
 namespace test {
 
+//------------------------------------------------
+
+template<class MF, class T, class... Args0>
+struct bind_wrapper
+{
+    MF mf_;
+    boost::shared_ptr<T> self_;
+    //std::tuple<Args0...> args0_;
+
+    template<class... Args>
+    void
+    operator()(Args&&... args) const
+    {
+        (self_.get()->*mf_)(std::forward<Args>(args)...);
+    }
+};
+
+template<class MF, class Arg0, class... Argn>
+bind_wrapper<MF, Arg0, Argn...>
+bind_front(
+    MF&& mf,
+    Arg0* arg0,
+    Argn&&... argn)
+{
+    return bind_wrapper<MF, Arg0, Argn...>{ mf, shared_from(arg0) };
+}
+
+//------------------------------------------------
+
+class session
+    : public boost::enable_shared_from
+{
+    beast::websocket::stream<socket_type> ws_;
+
+public:
+    session(socket_type&& sock)
+        : ws_(std::move(sock))
+    {
+    }
+
+    void
+    run()
+    {
+        ws_.async_accept(
+            bind_front(&session::on_accept, this));
+    }
+
+    void
+    on_accept(
+        system::error_code ec)
+    {
+        if(! BOOST_TEST(! ec.failed()))
+            return;
+        BOOST_TEST_PASS();
+
+    }
+};
+
+//------------------------------------------------
+
+/** Connect two TCP sockets together.
+*/
+template<class Executor>
+bool
+connect(
+    asio::basic_stream_socket<asio::ip::tcp, Executor>& s1,
+    asio::basic_stream_socket<asio::ip::tcp, Executor>& s2)
+
+{
+    BOOST_ASSERT(s1.get_executor() == s2.get_executor());
+    try
+    {
+        asio::basic_socket_acceptor<
+            asio::ip::tcp, Executor> a(s1.get_executor());
+        auto ep = asio::ip::tcp::endpoint(
+            asio::ip::make_address_v4("127.0.0.1"), 0);
+        a.open(ep.protocol());
+        a.set_option(
+            asio::socket_base::reuse_address(true));
+        a.bind(ep);
+        a.listen(0);
+        ep = a.local_endpoint();
+        a.async_accept(s2, success_handler());
+        s1.async_connect(ep, success_handler());
+        s1.get_executor().context().restart();
+        s1.get_executor().context().run();
+        if(! BOOST_TEST_EQ(s1.remote_endpoint(), s2.local_endpoint()))
+            return false;
+        if(! BOOST_TEST_EQ(s2.remote_endpoint(), s1.local_endpoint()))
+            return false;
+    }
+    catch(std::exception const&)
+    {
+        BOOST_TEST_FAIL();
+        return false;
+    }
+
+    return true;
+}
+
+//------------------------------------------------
+
 struct server::impl
 {
     asio::io_context ioc_;
-    asio::executor_work_guard<
-        asio::io_context::executor_type> wg_;
-    std::thread t_;
 
     impl()
-        : wg_(ioc_.get_executor())
-        , t_(
-            [&]
-            {
-               ioc_.run(); 
-            })
     {
     }
 
     ~impl()
     {
-        wg_.reset();
-        t_.join();
+    }
+
+    socket_type
+    connect()
+    {
+        socket_type s0(ioc_.get_executor());
+        socket_type s1(ioc_.get_executor());
+        test::connect(s0, s1);
+        auto s = boost::make_shared<session>(std::move(s0));
+        s->run();
+        return s1;
+    }
+
+    void
+    run()
+    {
+        ioc_.restart();
+        ioc_.run();
     }
 };
 
@@ -324,6 +439,20 @@ server::
     delete impl_;
 }
 
+socket_type
+server::
+connect()
+{
+    return impl_->connect();
+}
+
+void
+server::
+run()
+{
+    return impl_->run();
+}
+
 } // test
 } // ws_io
 } // boost

test/unit/server.hpp

@@ -7,21 +7,50 @@
 // Official repository: https://github.com/cppalliance/ws_io
 //
 
+#include <boost/asio/ip/tcp.hpp>
+#include "test_suite.hpp"
+
 namespace boost {
 namespace ws_io {
 namespace test {
 
+//------------------------------------------------
+
+using socket_type =
+    asio::basic_stream_socket<
+        asio::ip::tcp,
+        asio::io_context::executor_type>;
+
 class server
 {
 public:
     server();
     ~server();
 
+    socket_type
+    connect();
+
+    void
+    run();
+
 private:
     struct impl;
     impl* impl_;
 };
 
+//------------------------------------------------
+
+struct success_handler
+{
+    bool pass = false;
+
+    void
+    operator()(system::error_code ec, ...)
+    {
+        pass = BOOST_TEST(! ec.failed());
+    }
+};
+
 } // test
 } // ws_io
 } // boost

test/unit/stream.cpp

@@ -8,6 +8,10 @@
 //
 
 #include <boost/ws_proto/handshake.hpp>
+#include <boost/http_proto/context.hpp>
+#include <boost/http_proto/response_parser.hpp>
+#include <boost/http_io/read.hpp>
+#include <boost/asio/write.hpp>
 #include "test/unit/server.hpp"
 #include "test_suite.hpp"
 
@@ -24,7 +28,20 @@ struct stream_test
     void
     run()
     {
-        auto req = ws_proto::make_upgrade("");
+        test::server srv;
+        auto sock = srv.connect();
+        auto req = ws_proto::make_upgrade("/");
+        asio::write(sock, asio::buffer(req.buffer()));
+        srv.run();
+        http_proto::context ctx;
+        http_proto::parser::config_base cfg;
+        http_proto::install_parser_service(ctx, cfg);
+        http_proto::response_parser pr(ctx);
+        pr.reset();
+        pr.start();
+        http_io::async_read_header(sock, pr, test::success_handler());
+        srv.run();
+        auto res = pr.get();
     }
 };