feat: update sse-contract-tests to use foxy's server_session (#43)

Updates sse-contract-tests to use foxy::server_session and coroutines.

Author: cwaldren-ld

apps/sse-contract-tests/CMakeLists.txt

@@ -8,7 +8,14 @@ project(
         LANGUAGES CXX
 )
 
+set(Boost_USE_STATIC_LIBS ON)
+set(Boost_USE_MULTITHREADED ON)
+set(Boost_USE_STATIC_RUNTIME OFF)
+find_package(Boost 1.80 REQUIRED COMPONENTS coroutine)
+
+
 include(${CMAKE_FILES}/json.cmake)
+include(${CMAKE_FILES}/foxy.cmake)
 
 add_executable(sse-tests
         src/main.cpp
@@ -21,7 +28,9 @@ add_executable(sse-tests
 target_link_libraries(sse-tests PRIVATE
         launchdarkly::sse
         launchdarkly::common
+        foxy
         nlohmann_json::nlohmann_json
+        Boost::coroutine
         )
 
 target_include_directories(sse-tests PUBLIC include)

apps/sse-contract-tests/include/server.hpp

@@ -7,44 +7,42 @@
 #include "entity_manager.hpp"
 #include "logger.hpp"
 
+#include <foxy/listener.hpp>
+
 namespace net = boost::asio;  // from <boost/asio.hpp>
 
 #include <vector>
 
 using tcp = boost::asio::ip::tcp;  // from <boost/asio/ip/tcp.hpp>
 
-class server : public std::enable_shared_from_this<server> {
-    net::io_context& ioc_;
-    tcp::acceptor acceptor_;
+class server {
+    foxy::listener listener_;
     EntityManager entity_manager_;
     std::vector<std::string> caps_;
     launchdarkly::Logger& logger_;
 
    public:
     /**
      * Constructs a server, which stands up a REST API at the given
-     * port and address.
+     * port and address. The server is ready to accept connections upon
+     * construction.
      * @param ioc IO context.
      * @param address Address to bind.
      * @param port Port to bind.
      * @param logger Logger.
      */
     server(net::io_context& ioc,
            std::string const& address,
-           std::string const& port,
+           unsigned short port,
            launchdarkly::Logger& logger);
     /**
      * Advertise an optional test-harness capability, such as "comments".
      * @param cap
      */
     void add_capability(std::string cap);
+
     /**
-     * Begins an async operation to start accepting requests.
+     * Shuts down the server.
      */
-    void run();
-
-   private:
-    void do_accept();
-    void on_accept(boost::system::error_code const& ec, tcp::socket socket);
-    void fail(boost::beast::error_code ec, char const* what);
+    void shutdown();
 };

apps/sse-contract-tests/include/session.hpp

@@ -5,65 +5,90 @@
 
 #include <boost/beast.hpp>
 #include <boost/beast/http.hpp>
+#include <foxy/server_session.hpp>
 #include <vector>
 
+#include <boost/asio/yield.hpp>
+
 namespace beast = boost::beast;    // from <boost/beast.hpp>
 namespace http = beast::http;      // from <boost/beast/http.hpp>
 namespace net = boost::asio;       // from <boost/asio.hpp>
 using tcp = boost::asio::ip::tcp;  // from <boost/asio/ip/tcp.hpp>
 
-class Session : public std::enable_shared_from_this<Session> {
-    beast::tcp_stream stream_;
-    beast::flat_buffer buffer_{8192};
-    http::request<http::string_body> request_;
-    EntityManager& manager_;
-    std::vector<std::string> capabilities_;
-    std::function<void()> on_shutdown_cb_;
-    bool shutdown_requested_;
-    launchdarkly::Logger& logger_;
-
+class Session : boost::asio::coroutine {
    public:
+    using Request = http::request<http::string_body>;
+    using Response = http::response<http::string_body>;
+
+    struct Frame {
+        Request request_;
+        Response resp_;
+    };
+
     /**
      * Constructs a session, which provides a REST API.
-     * @param socket Connected socket.
+     * @param session The HTTP session.
      * @param manager Manager through which entities can be created/destroyed.
      * @param caps Test service capabilities to advertise.
      * @param logger Logger.
      */
-    explicit Session(tcp::socket&& socket,
-                     EntityManager& manager,
-                     std::vector<std::string> caps,
-                     launchdarkly::Logger& logger);
+    Session(foxy::server_session& session,
+            EntityManager& manager,
+            std::vector<std::string>& caps,
+            launchdarkly::Logger& logger);
 
-    ~Session();
+    template <class Self>
+    auto operator()(Self& self,
+                    boost::system::error_code ec = {},
+                    std::size_t const bytes_transferred = 0) -> void {
+        using launchdarkly::LogLevel;
+        auto& f = *frame_;
 
-    /**
-     * Set a callback to be invoked when a REST client requests shutdown.
-     */
-    template <typename Callback>
-    void on_shutdown(Callback cb) {
-        on_shutdown_cb_ = cb;
-    }
-    /**
-     * Begin waiting for requests.
-     */
-    void start();
-    /**
-     * Stop waiting for requests and close the session.
-     */
-    void stop();
+        reenter(*this) {
+            while (true) {
+                f.resp_ = {};
+                f.request_ = {};
 
-   private:
-    http::message_generator handle_request(
-        http::request<http::string_body>&& req);
-    void do_read();
+                yield session_.async_read(f.request_, std::move(self));
+                if (ec) {
+                    LD_LOG(logger_, LogLevel::kWarn)
+                        << "session: read: " << ec.what();
+                    break;
+                }
+
+                if (auto response = generate_response(f.request_)) {
+                    f.resp_ = *response;
+                } else {
+                    LD_LOG(logger_, LogLevel::kWarn)
+                        << "session: shutdown requested by client";
+                    std::exit(0);
+                }
 
-    void do_stop(char const* reason);
-    void on_read(beast::error_code ec, std::size_t bytes_transferred);
+                yield session_.async_write(f.resp_, std::move(self));
 
-    void send_response(http::message_generator&& msg);
+                if (ec) {
+                    LD_LOG(logger_, LogLevel::kWarn)
+                        << "session: write: " << ec.what();
+                    break;
+                }
+
+                if (!f.request_.keep_alive()) {
+                    break;
+                }
+            }
+
+            return self.complete({}, 0);
+        }
+    }
 
-    void on_write(bool keep_alive,
-                  beast::error_code ec,
-                  std::size_t bytes_transferred);
+    std::optional<Response> generate_response(Request& req);
+
+   private:
+    foxy::server_session& session_;
+    std::unique_ptr<Frame> frame_;
+    EntityManager& manager_;
+    std::vector<std::string>& caps_;
+    launchdarkly::Logger& logger_;
 };
+
+#include <boost/asio/unyield.hpp>

apps/sse-contract-tests/src/main.cpp

@@ -4,7 +4,9 @@
 
 #include <boost/asio/io_context.hpp>
 #include <boost/asio/signal_set.hpp>
+#include <boost/asio/spawn.hpp>
 #include <boost/beast.hpp>
+#include <boost/lexical_cast.hpp>
 
 #include <memory>
 
@@ -19,31 +21,43 @@ int main(int argc, char* argv[]) {
     launchdarkly::Logger logger{
         std::make_unique<ConsoleBackend>("sse-contract-tests")};
 
-    std::string port = "8123";
+    const std::string default_port = "8123";
+    std::string port = default_port;
     if (argc == 2) {
         port =
             argv[1];  // NOLINT(cppcoreguidelines-pro-bounds-pointer-arithmetic)
     }
+
     try {
         net::io_context ioc{1};
 
-        auto srv = std::make_shared<server>(ioc, "0.0.0.0", port, logger);
-        srv->add_capability("headers");
-        srv->add_capability("comments");
-        srv->add_capability("report");
-        srv->add_capability("post");
-        srv->add_capability("read-timeout");
-        srv->run();
+        auto p = boost::lexical_cast<unsigned short>(port);
+        server srv(ioc, "0.0.0.0", p, logger);
+
+        srv.add_capability("headers");
+        srv.add_capability("comments");
+        srv.add_capability("report");
+        srv.add_capability("post");
+        srv.add_capability("read-timeout");
 
         net::signal_set signals{ioc, SIGINT, SIGTERM};
-        signals.async_wait([&](beast::error_code const&, int) {
+
+        boost::asio::spawn(ioc.get_executor(), [&](auto yield) mutable {
+            signals.async_wait(yield);
             LD_LOG(logger, LogLevel::kInfo) << "shutting down..";
-            ioc.stop();
+            srv.shutdown();
         });
 
         ioc.run();
         LD_LOG(logger, LogLevel::kInfo) << "bye!";
 
+    } catch (boost::bad_lexical_cast&) {
+        LD_LOG(logger, LogLevel::kError)
+            << "invalid port (" << port
+            << "), provide a number (no arguments defaults "
+               "to port "
+            << default_port << ")";
+        return EXIT_FAILURE;
     } catch (std::exception const& e) {
         LD_LOG(logger, LogLevel::kError) << e.what();
         return EXIT_FAILURE;

apps/sse-contract-tests/src/server.cpp

@@ -2,6 +2,7 @@
 #include "session.hpp"
 
 #include <boost/asio/dispatch.hpp>
+#include <boost/asio/ip/address.hpp>
 #include <boost/asio/placeholders.hpp>
 #include <boost/asio/strand.hpp>
 #include <boost/bind.hpp>
@@ -11,49 +12,17 @@ using launchdarkly::LogLevel;
 
 server::server(net::io_context& ioc,
                std::string const& address,
-               std::string const& port,
+               unsigned short port,
                launchdarkly::Logger& logger)
-    : ioc_{ioc},
-      acceptor_{ioc},
+    : listener_{ioc.get_executor(),
+                tcp::endpoint(boost::asio::ip::make_address(address), port)},
       entity_manager_{ioc.get_executor(), logger},
-      caps_{},
       logger_{logger} {
-    beast::error_code ec;
-
-    tcp::resolver resolver{ioc_};
-    tcp::endpoint endpoint = *resolver.resolve(address, port, ec).begin();
-    if (ec) {
-        fail(ec, "resolve");
-        return;
-    }
-    acceptor_.open(endpoint.protocol(), ec);
-    if (ec) {
-        fail(ec, "open");
-        return;
-    }
-    acceptor_.set_option(tcp::acceptor::reuse_address(true), ec);
-    if (ec) {
-        fail(ec, "set_option");
-        return;
-    }
-    acceptor_.bind(endpoint, ec);
-    if (ec) {
-        fail(ec, "bind");
-        return;
-    }
-    acceptor_.listen(net::socket_base::max_listen_connections, ec);
-    if (ec) {
-        fail(ec, "listen");
-        return;
-    }
-
     LD_LOG(logger_, LogLevel::kInfo)
         << "server: listening on " << address << ":" << port;
-}
-
-void server::fail(beast::error_code ec, char const* what) {
-    LD_LOG(logger_, LogLevel::kError)
-        << "server: " << what << ": " << ec.message();
+    listener_.async_accept([this](auto& server) {
+        return Session(server, entity_manager_, caps_, logger_);
+    });
 }
 
 void server::add_capability(std::string cap) {
@@ -62,39 +31,6 @@ void server::add_capability(std::string cap) {
     caps_.push_back(std::move(cap));
 }
 
-void server::run() {
-    LD_LOG(logger_, LogLevel::kDebug) << "server: run requested";
-    net::dispatch(
-        acceptor_.get_executor(),
-        beast::bind_front_handler(&server::do_accept, shared_from_this()));
-}
-
-void server::do_accept() {
-    LD_LOG(logger_, LogLevel::kDebug) << "server: waiting for connection";
-    acceptor_.async_accept(
-        net::make_strand(ioc_),
-        beast::bind_front_handler(&server::on_accept, shared_from_this()));
-}
-
-void server::on_accept(boost::system::error_code const& ec,
-                       tcp::socket socket) {
-    if (!acceptor_.is_open()) {
-        return;
-    }
-    if (ec) {
-        fail(ec, "accept");
-        return;
-    }
-
-    auto session = std::make_shared<Session>(std::move(socket), entity_manager_,
-                                             caps_, logger_);
-
-    session->on_shutdown([this]() {
-        LD_LOG(logger_, LogLevel::kDebug) << "server: terminating";
-        ioc_.stop();
-    });
-
-    session->start();
-
-    do_accept();
+void server::shutdown() {
+    listener_.shutdown();
 }