add websocket support & replace promise/future with coroutine

CMakeLists.txt

@@ -1,6 +1,6 @@
 cmake_minimum_required(VERSION 3.10)
 
-project(bserv)
+project(bserv_main)
 
 set(CMAKE_CXX_STANDARD 17)
 set(CMAKE_CXX_STANDARD_REQUIRED True)
@@ -13,12 +13,7 @@ set(CMAKE_CXX_FLAGS "-Wall -Wextra")
 set(CMAKE_CXX_FLAGS_DEBUG "-g")
 set(CMAKE_CXX_FLAGS_RELEASE "-O3")
 
-add_executable(bserv main.cpp)
+add_subdirectory(bserv)
-target_link_libraries(bserv
+
-                      pthread
+add_executable(main main.cpp)
-                      boost_thread
+target_link_libraries(main bserv)
-                      boost_log
-                      boost_log_setup
-                      pqxx
-                      pq
-                      cryptopp)

bserv/CMakeLists.txt

@@ -0,0 +1,26 @@
+cmake_minimum_required(VERSION 3.10)
+
+project(bserv)
+
+set(CMAKE_CXX_STANDARD 17)
+set(CMAKE_CXX_STANDARD_REQUIRED True)
+
+if(NOT CMAKE_BUILD_TYPE)
+  set(CMAKE_BUILD_TYPE Release)
+endif()
+
+set(CMAKE_CXX_FLAGS "-Wall -Wextra")
+set(CMAKE_CXX_FLAGS_DEBUG "-g")
+set(CMAKE_CXX_FLAGS_RELEASE "-O3")
+
+add_library(bserv server.cpp)
+target_link_libraries(bserv
+                      pthread
+                      boost_thread
+                      boost_coroutine
+                      boost_log
+                      boost_log_setup
+                      boost_json
+                      pqxx
+                      pq
+                      cryptopp)

bserv/client.hpp

@@ -2,8 +2,9 @@
 #define _CLIENT_HPP
 
 #include <boost/beast.hpp>
+#include <boost/asio/spawn.hpp>
 #include <boost/asio.hpp>
-#include <boost/json/src.hpp>
+#include <boost/json.hpp>
 
 #include <iostream>
 #include <string>
@@ -29,139 +30,74 @@ using response_type = http::response<http::string_body>;
 class request_failed_exception
     : public std::exception {
 private:
-    std::string msg_;
+    const std::string msg_;
 public:
     request_failed_exception(const std::string& msg) : msg_{msg} {}
     const char* what() const noexcept { return msg_.c_str(); }
 };
 
 // https://www.boost.org/doc/libs/1_75_0/libs/beast/example/http/client/async/http_client_async.cpp
+// https://www.boost.org/doc/libs/1_75_0/libs/beast/example/http/client/coro/http_client_coro.cpp
 
 // sends one async request to a remote server
-template <typename ResponseType>
+inline http::response<http::string_body> http_client_send(
-class http_client_session
-    : public std::enable_shared_from_this<
-        http_client_session<ResponseType>> {
-private:
-    tcp::resolver resolver_;
-    beast::tcp_stream stream_;
-    // must persist between reads
-    beast::flat_buffer buffer_;
-    http::request<http::string_body> req_;
-    http::response<http::string_body> res_;
-    std::promise<ResponseType> promise_;
-    void failed(const beast::error_code& ec, const std::string& what) {
-        promise_.set_exception(
-            std::make_exception_ptr(
-                request_failed_exception{what + ": " + ec.message()}));
-    }
-public:
-    http_client_session(
         asio::io_context& ioc,
-        const http::request<http::string_body>& req)
+        asio::yield_context& yield,
-    : resolver_{asio::make_strand(ioc)},
-    stream_{asio::make_strand(ioc)}, req_{req} {}
-    std::future<ResponseType> send(
         const std::string& host,
-        const std::string& port) {
+        const std::string& port,
-        resolver_.async_resolve(
+        const http::request<http::string_body>& req) {
-            host, port,
+    beast::error_code ec;
-            beast::bind_front_handler(
+    tcp::resolver resolver{ioc};
-                &http_client_session::on_resolve,
+    const auto results = resolver.async_resolve(host, port, yield[ec]);
-                http_client_session<ResponseType>::shared_from_this()));
+    if (ec) {
-        return promise_.get_future();
+        throw request_failed_exception{"http_client_session::resolver resolve: " + ec.message()};
     }
-    void on_resolve(
+    beast::tcp_stream stream{ioc};
-        beast::error_code ec,
+    // sets a timeout on the operation
-        tcp::resolver::results_type results) {
+    stream.expires_after(std::chrono::seconds(EXPIRY_TIME));
-        if (ec) {
+    // makes the connection on the IP address we get from a lookup
-            failed(ec, "http_client_session::resolver resolve");
+    stream.async_connect(results, yield[ec]);
-            return;
+    if (ec) {
-        }
+        throw request_failed_exception{"http_client_session::stream connect: " + ec.message()};
-        // sets a timeout on the operation
-        stream_.expires_after(std::chrono::seconds(EXPIRY_TIME));
-        // makes the connection on the IP address we get from a lookup
-        stream_.async_connect(
-            results,
-            beast::bind_front_handler(
-                &http_client_session::on_connect,
-                http_client_session<ResponseType>::shared_from_this()));
     }
-    void on_connect(
+    // sets a timeout on the operation
-        beast::error_code ec,
+    stream.expires_after(std::chrono::seconds(EXPIRY_TIME));
-        tcp::resolver::results_type::endpoint_type) {
+    // sends the HTTP request to the remote host
-        if (ec) {
+    http::async_write(stream, req, yield[ec]);
-            failed(ec, "http_client_session::stream connect");
+    if (ec) {
-            return;
+        throw request_failed_exception{"http_client_session::stream write: " + ec.message()};
-        }
-        // sets a timeout on the operation
-        stream_.expires_after(std::chrono::seconds(EXPIRY_TIME));
-        // sends the HTTP request to the remote host
-        http::async_write(
-            stream_, req_,
-            beast::bind_front_handler(
-                &http_client_session::on_write,
-                http_client_session<ResponseType>::shared_from_this()));
     }
-    void on_write(
+    beast::flat_buffer buffer;
-        beast::error_code ec,
+    http::response<http::string_body> res;
-        std::size_t bytes_transferred) {
+    // receives the HTTP response
-        boost::ignore_unused(bytes_transferred);
+    http::async_read(stream, buffer, res, yield[ec]);
-        if (ec) {
+    if (ec) {
-            failed(ec, "http_client_session::stream write");
+        throw request_failed_exception{"http_client_session::stream read: " + ec.message()};
-            return;
-        }
-        // receives the HTTP response
-        http::async_read(
-            stream_, buffer_, res_,
-            beast::bind_front_handler(
-                &http_client_session::on_read, 
-                http_client_session<ResponseType>::shared_from_this()));
     }
-    static_assert(std::is_same<ResponseType, http::response<http::string_body>>::value
+    // gracefully close the socket
-        || std::is_same<ResponseType, boost::json::value>::value,
+    stream.socket().shutdown(tcp::socket::shutdown_both, ec);
-        "unsupported `ResponseType`");
+    // `not_connected` happens sometimes so don't bother reporting it
-    void on_read(
+    if (ec && ec != beast::errc::not_connected) {
-        beast::error_code ec,
+        // reports the error to the log!
-        std::size_t bytes_transferred) {
+        fail(ec, "http_client_session::stream::socket shutdown");
-        boost::ignore_unused(bytes_transferred);
+        // return;
-        if (ec) {
-            failed(ec, "http_client_session::stream read");
-            return;
-        }
-        if constexpr (std::is_same<ResponseType, http::response<http::string_body>>::value) {
-            promise_.set_value(std::move(res_));
-        } else if constexpr (std::is_same<ResponseType, boost::json::value>::value) {
-            promise_.set_value(boost::json::parse(res_.body()));
-        } else { // this should never happen
-            promise_.set_exception(
-                std::make_exception_ptr(
-                    request_failed_exception{"unsupported `ResponseType`"}));
-        }
-        // gracefully close the socket
-        stream_.socket().shutdown(tcp::socket::shutdown_both, ec);
-        // `not_connected` happens sometimes so don't bother reporting it
-        if (ec && ec != beast::errc::not_connected) {
-            // reports the error to the log!
-            fail(ec, "http_client_session::stream::socket shutdown");
-            return;
-        }
-        // if we get here then the connection is closed gracefully
     }
-};
+    // if we get here then the connection is closed gracefully
+    return res;
+}
 
-request_type get_request(
+inline request_type get_request(
     const std::string& host,
     const std::string& target,
     const http::verb& method,
-    const boost::json::object& obj) {
+    const boost::json::value& val) {
     request_type req;
     req.method(method);
     req.target(target);
     req.set(http::field::host, host);
     req.set(http::field::user_agent, NAME);
     req.set(http::field::content_type, "application/json");
-    req.body() = boost::json::serialize(obj);
+    req.body() = boost::json::serialize(val);
     req.prepare_payload();
     return req;
 }
@@ -169,99 +105,97 @@ request_type get_request(
 class http_client {
 private:
     asio::io_context& ioc_;
+    asio::yield_context& yield_;
 public:
-    http_client(asio::io_context& ioc)
+    http_client(asio::io_context& ioc, asio::yield_context& yield)
-    : ioc_{ioc} {}
+    : ioc_{ioc}, yield_{yield} {}
-    std::future<http::response<http::string_body>> request(
+    http::response<http::string_body> request(
         const std::string& host,
         const std::string& port,
         const http::request<http::string_body>& req) {
-        return std::make_shared<
+        return http_client_send(ioc_, yield_, host, port, req);
-            http_client_session<http::response<http::string_body>>
-        >(ioc_, req)->send(host, port);
     }
-    std::future<boost::json::value> request_for_object(
+    boost::json::value request_for_value(
         const std::string& host,
         const std::string& port,
         const http::request<http::string_body>& req) {
-        return std::make_shared<
+        return boost::json::parse(request(host, port, req).body());
-            http_client_session<boost::json::value>
-        >(ioc_, req)->send(host, port);
     }
 
-    std::future<response_type> send(
+    response_type send(
         const std::string& host,
         const std::string& port,
         const std::string& target,
         const http::verb& method,
-        const boost::json::object& obj) {
+        const boost::json::value& val) {
-        request_type req = get_request(host, target, method, obj);
+        request_type req = get_request(host, target, method, val);
         return request(host, port, req);
     }
-    std::future<boost::json::value> send_for_object(
+    boost::json::value send_for_value(
         const std::string& host,
         const std::string& port,
         const std::string& target,
         const http::verb& method,
-        const boost::json::object& obj) {
+        const boost::json::value& val) {
-        request_type req = get_request(host, target, method, obj);
+        request_type req = get_request(host, target, method, val);
-        return request_for_object(host, port, req);
+        return request_for_value(host, port, req);
     }
-    std::future<response_type> get(
+
+    response_type get(
         const std::string& host,
         const std::string& port,
         const std::string& target,
-        const boost::json::object& obj) {
+        const boost::json::value& val) {
-        return send(host, port, target, http::verb::get, obj);
+        return send(host, port, target, http::verb::get, val);
     }
-    std::future<boost::json::value> get_for_object(
+    boost::json::value get_for_value(
         const std::string& host,
         const std::string& port,
         const std::string& target,
-        const boost::json::object& obj) {
+        const boost::json::value& val) {
-        return send_for_object(host, port, target, http::verb::get, obj);
+        return send_for_value(host, port, target, http::verb::get, val);
     }
-    std::future<response_type> put(
+    response_type put(
         const std::string& host,
         const std::string& port,
         const std::string& target,
-        const boost::json::object& obj) {
+        const boost::json::value& val) {
-        return send(host, port, target, http::verb::put, obj);
+        return send(host, port, target, http::verb::put, val);
     }
-    std::future<boost::json::value> put_for_object(
+    boost::json::value put_for_value(
         const std::string& host,
         const std::string& port,
         const std::string& target,
-        const boost::json::object& obj) {
+        const boost::json::value& val) {
-        return send_for_object(host, port, target, http::verb::put, obj);
+        return send_for_value(host, port, target, http::verb::put, val);
     }
-    std::future<response_type> post(
+    response_type post(
         const std::string& host,
         const std::string& port,
         const std::string& target,
-        const boost::json::object& obj) {
+        const boost::json::value& val) {
-        return send(host, port, target, http::verb::post, obj);
+        return send(host, port, target, http::verb::post, val);
     }
-    std::future<boost::json::value> post_for_object(
+    boost::json::value post_for_value(
         const std::string& host,
         const std::string& port,
         const std::string& target,
-        const boost::json::object& obj) {
+        const boost::json::value& val) {
-        return send_for_object(host, port, target, http::verb::post, obj);
+        return send_for_value(host, port, target, http::verb::post, val);
     }
-    std::future<response_type> delete_(
+    response_type delete_(
         const std::string& host,
         const std::string& port,
         const std::string& target,
-        const boost::json::object& obj) {
+        const boost::json::value& val) {
-        return send(host, port, target, http::verb::delete_, obj);
+        return send(host, port, target, http::verb::delete_, val);
     }
-    std::future<boost::json::value> delete_for_object(
+    boost::json::value delete_for_value(
         const std::string& host,
         const std::string& port,
         const std::string& target,
-        const boost::json::object& obj) {
+        const boost::json::value& val) {
-        return send_for_object(host, port, target, http::verb::delete_, obj);
+        return send_for_value(host, port, target, http::verb::delete_, val);
     }
 };
 

bserv/common.hpp

@@ -9,5 +9,6 @@
 #include "server.hpp"
 #include "session.hpp"
 #include "utils.hpp"
+#include "websocket.hpp"
 
 #endif  // _COMMON_HPP

bserv/database.hpp

@@ -1,7 +1,7 @@
 #ifndef _DATABASE_HPP
 #define _DATABASE_HPP
 
-#include <boost/json/src.hpp>
+#include <boost/json.hpp>
 
 #include <cstddef>
 #include <string>
@@ -78,7 +78,7 @@ public:
     }
 };
 
-db_connection::~db_connection() {
+inline db_connection::~db_connection() {
     std::lock_guard<std::mutex> lg{mgr_.queue_lock_};
     mgr_.queue_.emplace(conn_);
     // if this is the first available connection back to the queue,
@@ -157,12 +157,12 @@ std::shared_ptr<db_parameter> convert_parameter(
     return std::make_shared<db_value<Param>>(param);
 }
 
-std::shared_ptr<db_parameter> convert_parameter(
+inline std::shared_ptr<db_parameter> convert_parameter(
     const char* param) {
     return std::make_shared<db_value<std::string>>(param);
 }
 
-std::shared_ptr<db_parameter> convert_parameter(
+inline std::shared_ptr<db_parameter> convert_parameter(
     const db_name& param) {
     return std::make_shared<db_name>(param);
 }

bserv/logging.hpp

@@ -20,7 +20,7 @@ namespace keywords = boost::log::keywords;
 namespace src = boost::log::sources;
 
 // this function should be called before logging is used
-void init_logging(const server_config& config) {
+inline void init_logging(const server_config& config) {
     logging::add_file_log(
         keywords::file_name = config.get_log_path() + "_%Y%m%d_%H-%M-%S.%N.log",
         keywords::rotation_size = config.get_log_rotation_size(),
@@ -39,7 +39,7 @@ void init_logging(const server_config& config) {
 #define lgerror BOOST_LOG_TRIVIAL(error)
 #define lgfatal BOOST_LOG_TRIVIAL(fatal)
 
-void fail(const boost::system::error_code& ec, const char* what) {
+inline void fail(const boost::system::error_code& ec, const char* what) {
     lgerror << what << ": " << ec.message() << std::endl;
 }
 

bserv/router.hpp

@@ -1,8 +1,10 @@
 #ifndef _ROUTER_HPP
 #define _ROUTER_HPP
 
+#include <boost/asio/spawn.hpp>
+#include <boost/asio.hpp>
 #include <boost/beast.hpp>
-#include <boost/json/src.hpp>
+#include <boost/json.hpp>
 
 #include <string>
 #include <regex>
@@ -19,6 +21,7 @@
 #include "session.hpp"
 #include "utils.hpp"
 #include "config.hpp"
+#include "websocket.hpp"
 
 namespace bserv {
 
@@ -28,7 +31,6 @@ namespace http = beast::http;
 struct server_resources {
     std::shared_ptr<session_manager_base> session_mgr;
     std::shared_ptr<db_connection_manager> db_conn_mgr;
-    std::shared_ptr<http_client> http_client_ptr;
 };
 
 namespace placeholders {
@@ -62,6 +64,8 @@ constexpr placeholder<-4> json_params;
 constexpr placeholder<-5> db_connection_ptr;
 // std::shared_ptr<bserv::http_client>
 constexpr placeholder<-6> http_client_ptr;
+// std::shared_ptr<bserv::websocket_server>
+constexpr placeholder<-7> websocket_server_ptr;
 
 }  // placeholders
 
@@ -123,6 +127,8 @@ struct get_parameter<0, Head, Tail...> {
 template <typename Type>
 Type&& get_parameter_data(
     server_resources&,
+    asio::io_context&, asio::yield_context&,
+    std::shared_ptr<websocket_session>,
     const std::vector<std::string>&,
     request_type&, response_type&, Type&& val) {
     return static_cast<Type&&>(val);
@@ -131,14 +137,18 @@ Type&& get_parameter_data(
 template <int N, std::enable_if_t<(N >= 0), int> = 0>
 const std::string& get_parameter_data(
     server_resources&,
+    asio::io_context&, asio::yield_context&,
+    std::shared_ptr<websocket_session>,
     const std::vector<std::string>& url_params,
     request_type&, response_type&,
     placeholders::placeholder<N>) {
     return url_params[N];
 }
 
-std::shared_ptr<session_type> get_parameter_data(
+inline std::shared_ptr<session_type> get_parameter_data(
     server_resources& resources,
+    asio::io_context&, asio::yield_context&,
+    std::shared_ptr<websocket_session>,
     const std::vector<std::string>&,
     request_type& request, response_type& response,
     placeholders::placeholder<-1>) {
@@ -157,24 +167,30 @@ std::shared_ptr<session_type> get_parameter_data(
     return session_ptr;
 }
 
-request_type& get_parameter_data(
+inline request_type& get_parameter_data(
     server_resources&,
+    asio::io_context&, asio::yield_context&,
+    std::shared_ptr<websocket_session>,
     const std::vector<std::string>&,
     request_type& request, response_type&,
     placeholders::placeholder<-2>) {
     return request;
 }
 
-response_type& get_parameter_data(
+inline response_type& get_parameter_data(
     server_resources&,
+    asio::io_context&, asio::yield_context&,
+    std::shared_ptr<websocket_session>,
     const std::vector<std::string>&,
     request_type&, response_type& response,
     placeholders::placeholder<-3>) {
     return response;
 }
 
-boost::json::object get_parameter_data(
+inline boost::json::object get_parameter_data(
     server_resources&,
+    asio::io_context&, asio::yield_context&,
+    std::shared_ptr<websocket_session>,
     const std::vector<std::string>&,
     request_type& request, response_type&,
     placeholders::placeholder<-4>) {
@@ -206,20 +222,34 @@ boost::json::object get_parameter_data(
     return body;
 }
 
-std::shared_ptr<db_connection> get_parameter_data(
+inline std::shared_ptr<db_connection> get_parameter_data(
     server_resources& resources,
+    asio::io_context&, asio::yield_context&,
+    std::shared_ptr<websocket_session>,
     const std::vector<std::string>&,
     request_type&, response_type&,
     placeholders::placeholder<-5>) {
     return resources.db_conn_mgr->get_or_block();
 }
 
-std::shared_ptr<http_client> get_parameter_data(
+inline std::shared_ptr<http_client> get_parameter_data(
-    server_resources& resources,
+    server_resources&,
+    asio::io_context& ioc, asio::yield_context& yield,
+    std::shared_ptr<websocket_session>,
     const std::vector<std::string>&,
     request_type&, response_type&,
     placeholders::placeholder<-6>) {
-    return resources.http_client_ptr;
+    return std::make_shared<http_client>(ioc, yield);
+}
+
+inline std::shared_ptr<websocket_server> get_parameter_data(
+    server_resources&,
+    asio::io_context&, asio::yield_context& yield,
+    std::shared_ptr<websocket_session> ws_session,
+    const std::vector<std::string>&,
+    request_type&, response_type&,
+    placeholders::placeholder<-7>) {
+    return std::make_shared<websocket_server>(*ws_session, yield);
 }
 
 template <int Idx, typename Func, typename Params, typename ...Args>
@@ -228,6 +258,8 @@ struct path_handler;
 template <int Idx, typename Ret, typename ...Args, typename ...Params>
 struct path_handler<Idx, Ret (*)(Args ...), parameter_pack<Params...>> {
     Ret invoke(server_resources& resources,
+        asio::io_context& ioc, asio::yield_context& yield,
+        std::shared_ptr<websocket_session> ws_session,
         Ret (*pf)(Args ...), parameter_pack<Params...>& params,
         const std::vector<std::string>& url_params,
         request_type& request, response_type& response) {
@@ -235,8 +267,8 @@ struct path_handler<Idx, Ret (*)(Args ...), parameter_pack<Params...>> {
         else return static_cast<path_handler<
             Idx - 1, Ret (*)(Args ...), parameter_pack<Params...>,
             typename get_parameter<Idx - 1, Params...>::type>*
-            >(this)->invoke2(resources, pf, params, url_params, request, response,
+            >(this)->invoke2(resources, ioc, yield, ws_session, pf, params, url_params, request, response,
-                get_parameter_data(resources, url_params, request, response,
+                get_parameter_data(resources, ioc, yield, ws_session, url_params, request, response,
                     get_parameter_value<Idx - 1>(params)));
     }
 };
@@ -251,6 +283,8 @@ struct path_handler<Idx, Ret (*)(Args ...),
         typename Head2, typename ...Tail2,
         std::enable_if_t<sizeof...(Tail2) == sizeof...(Tail), int> = 0>
     Ret invoke2(server_resources& resources,
+        asio::io_context& ioc, asio::yield_context& yield,
+        std::shared_ptr<websocket_session> ws_session,
         Ret (*pf)(Args ...), parameter_pack<Params...>& params, 
         const std::vector<std::string>& url_params,
         request_type& request, response_type& response,
@@ -261,8 +295,8 @@ struct path_handler<Idx, Ret (*)(Args ...),
         else return static_cast<path_handler<
             Idx - 1, Ret (*)(Args ...), parameter_pack<Params...>,
             typename get_parameter<Idx - 1, Params...>::type, Head, Tail...>*
-            >(this)->invoke2(resources, pf, params, url_params, request, response,
+            >(this)->invoke2(resources, ioc, yield, ws_session, pf, params, url_params, request, response,
-                get_parameter_data(resources, url_params, request, response,
+                get_parameter_data(resources, ioc, yield, ws_session, url_params, request, response,
                     get_parameter_value<Idx - 1>(params)),
                 static_cast<Head2&&>(head2), static_cast<Tail2&&>(tail2)...);
     }
@@ -274,7 +308,7 @@ const std::vector<std::pair<std::regex, std::string>> url_regex_mapping{
     {std::regex{"<path>"}, R"(([A-Za-z0-9_/\.\-]+))"}
 };
 
-std::string get_re_url(const std::string& url) {
+inline std::string get_re_url(const std::string& url) {
     std::string re_url = url;
     for (auto& [r, s] : url_regex_mapping)
         re_url = std::regex_replace(re_url, r, s);
@@ -289,6 +323,8 @@ struct path_holder : std::enable_shared_from_this<path_holder> {
         std::vector<std::string>&) const = 0;
     virtual std::optional<boost::json::value> invoke(
         server_resources&,
+        asio::io_context&, asio::yield_context&,
+        std::shared_ptr<websocket_session>,
         const std::vector<std::string>&,
         request_type&, response_type&) = 0;
 };
@@ -320,10 +356,12 @@ public:
     }
     std::optional<boost::json::value> invoke(
         server_resources& resources,
+        asio::io_context& ioc, asio::yield_context& yield,
+        std::shared_ptr<websocket_session> ws_session,
         const std::vector<std::string>& url_params,
         request_type& request, response_type& response) {
         return handler_.invoke(
-            resources,
+            resources, ioc, yield, ws_session,
             pf_, params_, url_params,
             request, response);
     }
@@ -369,11 +407,13 @@ public:
         resources_ = resources;
     }
     std::optional<boost::json::value> operator()(
+        asio::io_context& ioc, asio::yield_context& yield,
+        std::shared_ptr<websocket_session> ws_session,
         const std::string& url, request_type& request, response_type& response) {
         std::vector<std::string> url_params;
         for (auto& ptr : paths_) {
             if (ptr->match(url, url_params))
-                return ptr->invoke(*resources_, url_params, request, response);
+                return ptr->invoke(*resources_, ioc, yield, ws_session, url_params, request, response);
         }
         throw url_not_found_exception{};
     }

bserv/server.cpp

@@ -0,0 +1,503 @@
+#include "server.hpp"
+
+#include "logging.hpp"
+#include "utils.hpp"
+#include "client.hpp"
+#include "websocket.hpp"
+
+#include <iostream>
+#include <string>
+#include <cstddef>
+#include <cstdlib>
+#include <vector>
+#include <optional>
+#include <functional>
+#include <thread>
+#include <chrono>
+
+namespace bserv {
+
+
+std::string get_address(const tcp::socket& socket) {
+    tcp::endpoint end_point = socket.remote_endpoint();
+    std::string addr = end_point.address().to_string()
+        + ':' + std::to_string(end_point.port());
+    return addr;
+}
+
+http::response<http::string_body> handle_request(
+        http::request<http::string_body>& req, router& routes,
+        std::shared_ptr<websocket_session> ws_session,
+        asio::io_context& ioc, asio::yield_context& yield) {
+    
+    const auto bad_request = [&req](beast::string_view why) {
+        http::response<http::string_body> res{
+            http::status::bad_request, req.version()};
+        res.set(http::field::server, NAME);
+        res.set(http::field::content_type, "text/html");
+        res.keep_alive(req.keep_alive());
+        res.body() = std::string{why};
+        res.prepare_payload();
+        return res;
+    };
+
+    const auto not_found = [&req](beast::string_view target) {
+        http::response<http::string_body> res{
+            http::status::not_found, req.version()};
+        res.set(http::field::server, NAME);
+        res.set(http::field::content_type, "text/html");
+        res.keep_alive(req.keep_alive());
+        res.body() = "The requested url '" 
+            + std::string{target} + "' does not exist.";
+        res.prepare_payload();
+        return res;
+    };
+
+    const auto server_error = [&req](beast::string_view what) {
+        http::response<http::string_body> res{
+            http::status::internal_server_error, req.version()};
+        res.set(http::field::server, NAME);
+        res.set(http::field::content_type, "text/html");
+        res.keep_alive(req.keep_alive());
+        res.body() = "Internal server error: " + std::string{what};
+        res.prepare_payload();
+        return res;
+    };
+
+    boost::string_view target = req.target();
+    auto pos = target.find('?');
+    boost::string_view url;
+    if (pos == boost::string_view::npos) url = target;
+    else url = target.substr(0, pos);
+
+    http::response<http::string_body> res{
+        http::status::ok, req.version()};
+    res.set(http::field::server, NAME);
+    res.set(http::field::content_type, "application/json");
+    res.keep_alive(req.keep_alive());
+    
+    std::optional<boost::json::value> val;
+    try {
+        val = routes(ioc, yield, ws_session, std::string{url}, req, res);
+    } catch (const url_not_found_exception& e) {
+        return not_found(url);
+    } catch (const bad_request_exception& e) {
+        return bad_request("Request body is not a valid JSON string.");
+    } catch (const std::exception& e) {
+        return server_error(e.what());
+    } catch (...) {
+        return server_error("Unknown exception.");
+    }
+
+    if (val.has_value()) {
+        res.body() = json::serialize(val.value());
+        res.prepare_payload();
+    }
+
+    return res;
+}
+
+class websocket_session_server;
+
+void handle_websocket_request(
+        std::shared_ptr<websocket_session_server>,
+        std::shared_ptr<websocket_session> session,
+        http::request<http::string_body>& req, router& routes,
+        asio::io_context& ioc, asio::yield_context yield);
+
+class websocket_session_server
+    : public std::enable_shared_from_this<websocket_session_server> {
+private:
+    friend websocket_server;
+    std::string address_;
+    std::shared_ptr<websocket_session> session_;
+    http::request<http::string_body> req_;
+    router& routes_;
+    void on_accept(beast::error_code ec) {
+        if (ec) {
+            fail(ec, "websocket_session_server accept");
+            return;
+        }
+        // handles request here
+        asio::spawn(
+            session_->ioc_,
+            std::bind(
+                &handle_websocket_request,
+                shared_from_this(),
+                session_,
+                std::ref(req_),
+                std::ref(routes_),
+                std::ref(session_->ioc_),
+                std::placeholders::_1));
+    }
+public:
+    explicit websocket_session_server(
+        asio::io_context& ioc,
+        tcp::socket&& socket,
+        http::request<http::string_body>&& req,
+        router& routes)
+    : address_{get_address(socket)},
+        session_{std::make_shared<
+            websocket_session>(address_, ioc, std::move(socket))},
+        req_{std::move(req)}, routes_{routes} {
+        lgtrace << "websocket_session_server opened: " << address_;
+    }
+    ~websocket_session_server() {
+        lgtrace << "websocket_session_server closed: " << address_;
+    }
+    // starts the asynchronous accept operation
+    void do_accept() {
+        // sets suggested timeout settings for the websocket
+        session_->ws_.set_option(
+            websocket::stream_base::timeout::suggested(
+                beast::role_type::server));
+        // sets a decorator to change the Server of the handshake
+        session_->ws_.set_option(
+            websocket::stream_base::decorator(
+                [](websocket::response_type& res) {
+                    res.set(
+                        http::field::server,
+                        std::string{BOOST_BEAST_VERSION_STRING} + " websocket-server");
+                }));
+        // accepts the websocket handshake
+        session_->ws_.async_accept(
+            req_,
+            beast::bind_front_handler(
+                &websocket_session_server::on_accept,
+                shared_from_this()));
+    }
+};
+
+void handle_websocket_request(
+        std::shared_ptr<websocket_session_server>,
+        std::shared_ptr<websocket_session> session,
+        http::request<http::string_body>& req, router& routes,
+        asio::io_context& ioc, asio::yield_context yield) {
+    handle_request(req, routes, session, ioc, yield);
+}
+
+std::string websocket_server::read() {
+    beast::error_code ec;
+    beast::flat_buffer buffer;
+    // reads a message into the buffer
+    session_.ws_.async_read(buffer, yield_[ec]);
+    lgtrace << "websocket_server: read from " << session_.address_;
+    // this indicates that the session was closed
+    if (ec == websocket::error::closed) {
+        throw websocket_closed{};
+    }
+    if (ec) {
+        fail(ec, "websocket_server read");
+        throw websocket_io_exception{"websocket_server read: " + ec.message()};
+    }
+    // lgtrace << "websocket_server: received text? " << ws_.got_text() << " from " << address_;
+    return beast::buffers_to_string(buffer.data());
+}
+
+void websocket_server::write(const std::string& data) {
+    beast::error_code ec;
+    // ws_.text(ws_.got_text());
+    session_.ws_.async_write(asio::buffer(data), yield_[ec]);
+    lgtrace << "websocket_server: write to " << session_.address_;
+    if (ec) {
+        fail(ec, "websocket_server write");
+        throw websocket_io_exception{"websocket_server write: " + ec.message()};
+    }
+}
+
+
+class http_session;
+
+// this function produces an HTTP response for the given
+// request. The type of the response object depends on the
+// contents of the request, so the interface requires the
+// caller to pass a generic lambda for receiving the response.
+// NOTE: `send` should be called only once!
+template <class Send>
+void handle_http_request(
+        std::shared_ptr<http_session>,
+        http::request<http::string_body> req,
+        Send& send, router& routes, asio::io_context& ioc, asio::yield_context yield) {
+    send(handle_request(req, routes, nullptr, ioc, yield));
+}
+
+// handles an HTTP server connection
+class http_session
+    : public std::enable_shared_from_this<http_session> {
+private:
+    // the function object is used to send an HTTP message.
+    class send_lambda {
+    private:
+        http_session& self_;
+    public:
+        send_lambda(http_session& self)
+            : self_{self} {}
+        template <bool isRequest, class Body, class Fields>
+        void operator()(
+            http::message<isRequest, Body, Fields>&& msg) const {
+            // the lifetime of the message has to extend
+            // for the duration of the async operation so
+            // we use a shared_ptr to manage it.
+            auto sp = std::make_shared<
+                http::message<isRequest, Body, Fields>>(
+                    std::move(msg));
+            // stores a type-erased version of the shared
+            // pointer in the class to keep it alive.
+            self_.res_ = sp;
+            // writes the response
+            http::async_write(
+                self_.stream_, *sp,
+                beast::bind_front_handler(
+                    &http_session::on_write,
+                    self_.shared_from_this(),
+                    sp->need_eof()));
+        }
+    } lambda_;
+    asio::io_context& ioc_;
+    beast::tcp_stream stream_;
+    beast::flat_buffer buffer_;
+    boost::optional<
+        http::request_parser<http::string_body>> parser_;
+    std::shared_ptr<void> res_;
+    router& routes_;
+    router& ws_routes_;
+    const std::string address_;
+    void do_read() {
+        // constructs a new parser for each message
+        parser_.emplace();
+        // applies a reasonable limit to the allowed size
+        // of the body in bytes to prevent abuse.
+        parser_->body_limit(PAYLOAD_LIMIT);
+        // sets the timeout.
+        stream_.expires_after(std::chrono::seconds(EXPIRY_TIME));
+        // reads a request using the parser-oriented interface
+        http::async_read(
+            stream_, buffer_, *parser_,
+            beast::bind_front_handler(
+                &http_session::on_read,
+                shared_from_this()));
+    }
+    void on_read(
+        beast::error_code ec,
+        std::size_t bytes_transferred) {
+        boost::ignore_unused(bytes_transferred);
+        lgtrace << "received " << bytes_transferred << " byte(s) from: " << address_;
+        // this means they closed the connection
+        if (ec == http::error::end_of_stream) {
+            do_close();
+            return;
+        }
+        if (ec) {
+            fail(ec, "http_session async_read");
+            return;
+        }
+
+        // sees if it is a websocket upgrade
+        if (websocket::is_upgrade(parser_->get())) {
+            // creates a websocket session, transferring ownership
+            // of both the socket and the http request
+            std::make_shared<websocket_session_server>(
+                ioc_,
+                stream_.release_socket(),
+                parser_->release(),
+                ws_routes_
+            )->do_accept();
+            return;
+        }
+
+        // handles the request and sends the response
+
+        asio::spawn(
+            ioc_,
+            std::bind(
+                &handle_http_request<send_lambda>,
+                shared_from_this(),
+                parser_->release(),
+                std::ref(lambda_),
+                std::ref(routes_),
+                std::ref(ioc_),
+                std::placeholders::_1));
+        // handle_request(parser_->release(), lambda_, routes_);
+
+        // at this point the parser can be reset
+    }
+    void on_write(
+        bool close, beast::error_code ec,
+        std::size_t bytes_transferred) {
+        boost::ignore_unused(bytes_transferred);
+        // we're done with the response so delete it
+        res_.reset();
+        if (ec) {
+            fail(ec, "http_session async_write");
+            return;
+        }
+        lgtrace << "sent " << bytes_transferred << " byte(s) to: " << address_;
+        if (close) {
+            // this means we should close the connection, usually because
+            // the response indicated the "Connection: close" semantic.
+            do_close();
+            return;
+        }
+        // reads another request
+        do_read();
+    }
+    void do_close() {
+        // sends a TCP shutdown
+        beast::error_code ec;
+        stream_.socket().shutdown(tcp::socket::shutdown_send, ec);
+        // at this point the connection is closed gracefully
+        lgtrace << "socket connection closed: " << address_;
+    }
+public:
+    http_session(
+        asio::io_context& ioc,
+        tcp::socket&& socket,
+        router& routes,
+        router& ws_routes)
+        : lambda_{*this},
+        ioc_{ioc},
+        stream_{std::move(socket)},
+        routes_{routes},
+        ws_routes_{ws_routes},
+        address_{get_address(stream_.socket())} {
+        lgtrace << "http session opened: " << address_;
+    }
+    ~http_session() {
+        lgtrace << "http session closed: " << address_;
+    }
+    void run() {
+        asio::dispatch(
+            stream_.get_executor(),
+            beast::bind_front_handler(
+                &http_session::do_read,
+                shared_from_this()));
+    }
+};
+
+// accepts incoming connections and launches the sessions
+class listener
+    : public std::enable_shared_from_this<listener> {
+private:
+    asio::io_context& ioc_;
+    tcp::acceptor acceptor_;
+    router& routes_;
+    router& ws_routes_;
+    void do_accept() {
+        acceptor_.async_accept(
+            asio::make_strand(ioc_),
+            beast::bind_front_handler(
+                &listener::on_accept,
+                shared_from_this()));
+    }
+    void on_accept(beast::error_code ec, tcp::socket socket) {
+        if (ec) {
+            fail(ec, "listener::acceptor async_accept");
+        } else {
+            lgtrace << "listener accepts: " << get_address(socket);
+            std::make_shared<http_session>(
+                ioc_, std::move(socket), routes_, ws_routes_)->run();
+        }
+        do_accept();
+    }
+public:
+    listener(
+        asio::io_context& ioc,
+        tcp::endpoint endpoint,
+        router& routes,
+        router& ws_routes)
+        : ioc_{ioc},
+        acceptor_{asio::make_strand(ioc)},
+        routes_{routes},
+        ws_routes_{ws_routes} {
+        beast::error_code ec;
+        acceptor_.open(endpoint.protocol(), ec);
+        if (ec) {
+            fail(ec, "listener::acceptor open");
+            exit(EXIT_FAILURE);
+            return;
+        }
+        acceptor_.set_option(
+            asio::socket_base::reuse_address(true), ec);
+        if (ec) {
+            fail(ec, "listener::acceptor set_option");
+            exit(EXIT_FAILURE);
+            return;
+        }
+        acceptor_.bind(endpoint, ec);
+        if (ec) {
+            fail(ec, "listener::acceptor bind");
+            exit(EXIT_FAILURE);
+            return;
+        }
+        acceptor_.listen(
+            asio::socket_base::max_listen_connections, ec);
+        if (ec) {
+            fail(ec, "listener::acceptor listen");
+            exit(EXIT_FAILURE);
+            return;
+        }
+    }
+    void run() {
+        asio::dispatch(
+            acceptor_.get_executor(),
+            beast::bind_front_handler(
+                &listener::do_accept,
+                shared_from_this()));
+    }
+};
+
+
+server::server(const server_config& config, router&& routes, router&& ws_routes)
+        : ioc_{config.get_num_threads()},
+        routes_{std::move(routes)},
+        ws_routes_{std::move(ws_routes)} {
+    init_logging(config);
+
+    // database connection
+    try {
+        db_conn_mgr_ = std::make_shared<
+            db_connection_manager>(config.get_db_conn_str(), config.get_num_db_conn());
+    } catch (const std::exception& e) {
+        lgfatal << "db connection initialization failed: " << e.what() << std::endl;
+        exit(EXIT_FAILURE);
+    }
+    session_mgr_ = std::make_shared<memory_session_manager>();
+
+    std::shared_ptr<server_resources> resources_ptr = std::make_shared<server_resources>();
+    resources_ptr->session_mgr = session_mgr_;
+    resources_ptr->db_conn_mgr = db_conn_mgr_;
+
+    routes_.set_resources(resources_ptr);
+    ws_routes_.set_resources(resources_ptr);
+
+    // creates and launches a listening port
+    std::make_shared<listener>(
+        ioc_, tcp::endpoint{tcp::v4(), config.get_port()}, routes_, ws_routes_)->run();
+
+    // captures SIGINT and SIGTERM to perform a clean shutdown
+    asio::signal_set signals{ioc_, SIGINT, SIGTERM};
+    signals.async_wait(
+        [&](const boost::system::error_code&, int) {
+        // stops the `io_context`. This will cause `run()`
+        // to return immediately, eventually destroying the
+        // `io_context` and all of the sockets in it.
+        ioc_.stop();
+    });
+
+    lginfo << config.get_name() << " started";
+
+    // runs the I/O service on the requested number of threads
+    std::vector<std::thread> v;
+    v.reserve(config.get_num_threads() - 1);
+    for (int i = 1; i < config.get_num_threads(); ++i)
+        v.emplace_back([&]{ ioc_.run(); });
+    ioc_.run();
+
+    // if we get here, it means we got a SIGINT or SIGTERM
+    lginfo << "exiting " << config.get_name();
+
+    // blocks until all the threads exit
+    for (auto & t : v) t.join();
+}
+
+}  // bserv

bserv/server.hpp

@@ -3,8 +3,12 @@
  * 
  * reference:
  * https://www.boost.org/doc/libs/1_75_0/libs/beast/example/http/server/async/http_server_async.cpp
+ * https://www.boost.org/doc/libs/1_75_0/libs/beast/example/http/server/coro/http_server_coro.cpp
  * https://www.boost.org/doc/libs/1_75_0/libs/beast/example/advanced/server/advanced_server.cpp
  * 
+ * websocket:
+ * https://www.boost.org/doc/libs/1_75_0/libs/beast/example/websocket/server/async/websocket_server_async.cpp
+ * 
  */
 
 #ifndef _SERVER_HPP
@@ -12,370 +16,36 @@
 
 #include <boost/beast.hpp>
 #include <boost/asio.hpp>
+#include <boost/asio/spawn.hpp>
 #include <boost/optional.hpp>
-#include <boost/json/src.hpp>
+#include <boost/json.hpp>
 
-#include <iostream>
-#include <string>
-#include <cstddef>
-#include <cstdlib>
-#include <vector>
-#include <optional>
 #include <memory>
-#include <thread>
-#include <chrono>
 
 #include "config.hpp"
-#include "logging.hpp"
-#include "utils.hpp"
 #include "router.hpp"
 #include "database.hpp"
 #include "session.hpp"
-#include "client.hpp"
 
 namespace bserv {
 
 namespace beast = boost::beast;
 namespace http = beast::http;
+namespace websocket = beast::websocket;
 namespace asio = boost::asio;
 namespace json = boost::json;
 using asio::ip::tcp;
 
-// this function produces an HTTP response for the given
-// request. The type of the response object depends on the
-// contents of the request, so the interface requires the
-// caller to pass a generic lambda for receiving the response.
-// NOTE: `send` should be called only once!
-template <class Body, class Allocator, class Send>
-void handle_request(
-    http::request<Body, http::basic_fields<Allocator>>&& req,
-    Send&& send, router& routes) {
-    
-    const auto bad_request = [&req](beast::string_view why) {
-        http::response<http::string_body> res{
-            http::status::bad_request, req.version()};
-        res.set(http::field::server, NAME);
-        res.set(http::field::content_type, "text/html");
-        res.keep_alive(req.keep_alive());
-        res.body() = std::string{why};
-        res.prepare_payload();
-        return res;
-    };
-
-    const auto not_found = [&req](beast::string_view target) {
-        http::response<http::string_body> res{
-            http::status::not_found, req.version()};
-        res.set(http::field::server, NAME);
-        res.set(http::field::content_type, "text/html");
-        res.keep_alive(req.keep_alive());
-        res.body() = "The requested url '" 
-            + std::string{target} + "' does not exist.";
-        res.prepare_payload();
-        return res;
-    };
-
-    const auto server_error = [&req](beast::string_view what) {
-        http::response<http::string_body> res{
-            http::status::internal_server_error, req.version()};
-        res.set(http::field::server, NAME);
-        res.set(http::field::content_type, "text/html");
-        res.keep_alive(req.keep_alive());
-        res.body() = "Internal server error: " + std::string{what};
-        res.prepare_payload();
-        return res;
-    };
-
-    boost::string_view target = req.target();
-    auto pos = target.find('?');
-    boost::string_view url;
-    if (pos == boost::string_view::npos) url = target;
-    else url = target.substr(0, pos);
-
-    http::response<http::string_body> res{
-        http::status::ok, req.version()};
-    res.set(http::field::server, NAME);
-    res.set(http::field::content_type, "application/json");
-    res.keep_alive(req.keep_alive());
-    
-    std::optional<boost::json::value> val;
-    try {
-        val = routes(std::string{url}, req, res);
-    } catch (const url_not_found_exception& e) {
-        send(not_found(url));
-        return;
-    } catch (const bad_request_exception& e) {
-        send(bad_request("Request body is not a valid JSON string."));
-        return;
-    } catch (const std::exception& e) {
-        send(server_error(e.what()));
-        return;
-    } catch (...) {
-        send(server_error("Unknown exception."));
-        return;
-    }
-
-    if (val.has_value()) {
-        res.body() = json::serialize(val.value());
-        res.prepare_payload();
-    }
-
-    send(std::move(res));
-}
-
-std::string get_address(const tcp::socket& socket) {
-    tcp::endpoint end_point = socket.remote_endpoint();
-    std::string addr = end_point.address().to_string()
-        + ':' + std::to_string(end_point.port());
-    return addr;
-}
-
-// handles an HTTP server connection
-class http_session
-    : public std::enable_shared_from_this<http_session> {
-private:
-    // the function object is used to send an HTTP message.
-    class send_lambda {
-    private:
-        http_session& self_;
-    public:
-        send_lambda(http_session& self)
-            : self_{self} {}
-        template <bool isRequest, class Body, class Fields>
-        void operator()(
-            http::message<isRequest, Body, Fields>&& msg) const {
-            // the lifetime of the message has to extend
-            // for the duration of the async operation so
-            // we use a shared_ptr to manage it.
-            auto sp = std::make_shared<
-                http::message<isRequest, Body, Fields>>(
-                    std::move(msg));
-            // stores a type-erased version of the shared
-            // pointer in the class to keep it alive.
-            self_.res_ = sp;
-            // writes the response
-            http::async_write(
-                self_.stream_, *sp,
-                beast::bind_front_handler(
-                    &http_session::on_write,
-                    self_.shared_from_this(),
-                    sp->need_eof()));
-        }
-    } lambda_;
-    beast::tcp_stream stream_;
-    beast::flat_buffer buffer_;
-    boost::optional<
-        http::request_parser<http::string_body>> parser_;
-    std::shared_ptr<void> res_;
-    router& routes_;
-    const std::string address_;
-    void do_read() {
-        // constructs a new parser for each message
-        parser_.emplace();
-        // applies a reasonable limit to the allowed size
-        // of the body in bytes to prevent abuse.
-        parser_->body_limit(PAYLOAD_LIMIT);
-        // sets the timeout.
-        stream_.expires_after(std::chrono::seconds(EXPIRY_TIME));
-        // reads a request using the parser-oriented interface
-        http::async_read(
-            stream_, buffer_, *parser_,
-            beast::bind_front_handler(
-                &http_session::on_read,
-                shared_from_this()));
-    }
-    void on_read(
-        beast::error_code ec,
-        std::size_t bytes_transferred) {
-        boost::ignore_unused(bytes_transferred);
-        lgtrace << "received " << bytes_transferred << " byte(s) from: " << address_;
-        // this means they closed the connection
-        if (ec == http::error::end_of_stream) {
-            do_close();
-            return;
-        }
-        if (ec) {
-            fail(ec, "http_session async_read");
-            return;
-        }
-        // handles the request and sends the response
-        handle_request(parser_->release(), lambda_, routes_);
-        // at this point the parser can be reset
-    }
-    void on_write(
-        bool close, beast::error_code ec,
-        std::size_t bytes_transferred) {
-        boost::ignore_unused(bytes_transferred);
-        // we're done with the response so delete it
-        res_.reset();
-        if (ec) {
-            fail(ec, "http_session async_write");
-            return;
-        }
-        lgtrace << "sent " << bytes_transferred << " byte(s) to: " << address_;
-        if (close) {
-            // this means we should close the connection, usually because
-            // the response indicated the "Connection: close" semantic.
-            do_close();
-            return;
-        }
-        // reads another request
-        do_read();
-    }
-    void do_close() {
-        // sends a TCP shutdown
-        beast::error_code ec;
-        stream_.socket().shutdown(tcp::socket::shutdown_send, ec);
-        // at this point the connection is closed gracefully
-        lgtrace << "socket connection closed: " << address_;
-    }
-public:
-    http_session(tcp::socket&& socket, router& routes)
-        : lambda_{*this}, stream_{std::move(socket)}, routes_{routes},
-        address_{get_address(stream_.socket())} {
-        lgtrace << "http session opened: " << address_;
-    }
-    ~http_session() {
-        lgtrace << "http session closed: " << address_;
-    }
-    void run() {
-        asio::dispatch(
-            stream_.get_executor(),
-            beast::bind_front_handler(
-                &http_session::do_read,
-                shared_from_this()));
-    }
-};
-
-// accepts incoming connections and launches the sessions
-class listener
-    : public std::enable_shared_from_this<listener> {
-private:
-    asio::io_context& ioc_;
-    tcp::acceptor acceptor_;
-    router& routes_;
-    void do_accept() {
-        acceptor_.async_accept(
-            asio::make_strand(ioc_),
-            beast::bind_front_handler(
-                &listener::on_accept,
-                shared_from_this()));
-    }
-    void on_accept(beast::error_code ec, tcp::socket socket) {
-        if (ec) {
-            fail(ec, "listener::acceptor async_accept");
-        } else {
-            lgtrace << "listener accepts: " << get_address(socket);
-            std::make_shared<http_session>(
-                std::move(socket), routes_)->run();
-        }
-        do_accept();
-    }
-public:
-    listener(
-        asio::io_context& ioc,
-        tcp::endpoint endpoint,
-        router& routes)
-        : ioc_{ioc},
-        acceptor_{asio::make_strand(ioc)},
-        routes_{routes} {
-        beast::error_code ec;
-        acceptor_.open(endpoint.protocol(), ec);
-        if (ec) {
-            fail(ec, "listener::acceptor open");
-            exit(EXIT_FAILURE);
-            return;
-        }
-        acceptor_.set_option(
-            asio::socket_base::reuse_address(true), ec);
-        if (ec) {
-            fail(ec, "listener::acceptor set_option");
-            exit(EXIT_FAILURE);
-            return;
-        }
-        acceptor_.bind(endpoint, ec);
-        if (ec) {
-            fail(ec, "listener::acceptor bind");
-            exit(EXIT_FAILURE);
-            return;
-        }
-        acceptor_.listen(
-            asio::socket_base::max_listen_connections, ec);
-        if (ec) {
-            fail(ec, "listener::acceptor listen");
-            exit(EXIT_FAILURE);
-            return;
-        }
-    }
-    void run() {
-        asio::dispatch(
-            acceptor_.get_executor(),
-            beast::bind_front_handler(
-                &listener::do_accept,
-                shared_from_this()));
-    }
-};
-
 class server {
 private:
     // io_context for all I/O
     asio::io_context ioc_;
     router routes_;
+    router ws_routes_;
     std::shared_ptr<session_manager_base> session_mgr_;
     std::shared_ptr<db_connection_manager> db_conn_mgr_;
-    std::shared_ptr<http_client> http_client_ptr_;
 public:
-    server(const server_config& config, router&& routes)
+    server(const server_config& config, router&& routes, router&& ws_routes);
-        : ioc_{config.get_num_threads()},
-        routes_{std::move(routes)} {
-        init_logging(config);
-
-        // database connection
-        try {
-            db_conn_mgr_ = std::make_shared<
-                db_connection_manager>(config.get_db_conn_str(), config.get_num_db_conn());
-        } catch (const std::exception& e) {
-            lgfatal << "db connection initialization failed: " << e.what() << std::endl;
-            exit(EXIT_FAILURE);
-        }
-        session_mgr_ = std::make_shared<memory_session_manager>();
-        http_client_ptr_ = std::make_shared<http_client>(ioc_);
-
-        std::shared_ptr<server_resources> resources_ptr = std::make_shared<server_resources>();
-        resources_ptr->session_mgr = session_mgr_;
-        resources_ptr->db_conn_mgr = db_conn_mgr_;
-        resources_ptr->http_client_ptr = http_client_ptr_;
-
-        routes_.set_resources(resources_ptr);
-
-        // creates and launches a listening port
-        std::make_shared<listener>(
-            ioc_, tcp::endpoint{tcp::v4(), config.get_port()}, routes_)->run();
-
-        // captures SIGINT and SIGTERM to perform a clean shutdown
-        asio::signal_set signals{ioc_, SIGINT, SIGTERM};
-        signals.async_wait(
-            [&](const boost::system::error_code&, int) {
-            // stops the `io_context`. This will cause `run()`
-            // to return immediately, eventually destroying the
-            // `io_context` and all of the sockets in it.
-            ioc_.stop();
-        });
-
-        lginfo << config.get_name() << " started";
-
-        // runs the I/O service on the requested number of threads
-        std::vector<std::thread> v;
-        v.reserve(config.get_num_threads() - 1);
-        for (int i = 1; i < config.get_num_threads(); ++i)
-            v.emplace_back([&]{ ioc_.run(); });
-        ioc_.run();
-
-        // if we get here, it means we got a SIGINT or SIGTERM
-        lginfo << "exiting " << config.get_name();
-
-        // blocks until all the threads exit
-        for (auto & t : v) t.join();
-    }
 };
 
 }  // bserv

bserv/session.hpp

@@ -1,7 +1,7 @@
 #ifndef _SESSION_HPP
 #define _SESSION_HPP
 
-#include <boost/json/src.hpp>
+#include <boost/json.hpp>
 
 #include <cstddef>
 #include <string>

bserv/utils.hpp

@@ -25,10 +25,10 @@ namespace internal {
 //   it doesn't work with GNU GCC on Windows.
 // - for thread-safety, do not directly use it.
 //   use `get_rd_value` instead.
-std::random_device rd;
+inline std::random_device rd;
-std::mutex rd_mutex;
+inline std::mutex rd_mutex;
 
-auto get_rd_value() {
+inline auto get_rd_value() {
     std::lock_guard<std::mutex> lg{rd_mutex};
     return rd();
 }
@@ -51,7 +51,7 @@ const std::string url_safe_characters = "ABCDEFGHIJKLMNOPQRSTUVWXYZ"
 }  // internal
 
 // https://www.boost.org/doc/libs/1_75_0/libs/random/example/password.cpp
-std::string generate_random_string(std::size_t len) {
+inline std::string generate_random_string(std::size_t len) {
     std::string s;
     std::mt19937 rng{internal::get_rd_value()};
     std::uniform_int_distribution<> dist{0, (int) internal::chars.length() - 1};
@@ -62,7 +62,7 @@ std::string generate_random_string(std::size_t len) {
 namespace security {
 
 // https://codahale.com/a-lesson-in-timing-attacks/
-bool constant_time_compare(const std::string& a, const std::string& b) {
+inline bool constant_time_compare(const std::string& a, const std::string& b) {
     if (a.length() != b.length())
         return false;
     int result = 0;
@@ -72,7 +72,7 @@ bool constant_time_compare(const std::string& a, const std::string& b) {
 }
 
 // https://cryptopp.com/wiki/PKCS5_PBKDF2_HMAC
-std::string hash_password(
+inline std::string hash_password(
     const std::string& password,
     const std::string& salt,
     unsigned int iterations = 20000 /*320000*/) {
@@ -91,13 +91,13 @@ std::string hash_password(
     return result;
 }
 
-std::string encode_password(const std::string& password) {
+inline std::string encode_password(const std::string& password) {
     std::string salt = generate_random_string(16);
     std::string hashed_password = hash_password(password, salt);
     return salt + '$' + hashed_password;
 }
 
-bool check_password(const std::string& password,
+inline bool check_password(const std::string& password,
     const std::string& encoded_password) {
     std::string salt, hashed_password;
     std::string* a = &salt, * b = &hashed_password;
@@ -126,7 +126,7 @@ bool check_password(const std::string& password,
 
 // https://stackoverflow.com/questions/54060359/encoding-decoded-urls-in-c
 // there can be exceptions (std::stoi)!
-std::string decode_url(const std::string& s) {
+inline std::string decode_url(const std::string& s) {
     std::string r;
     for (std::size_t i = 0; i < s.length(); ++i) {
         if (s[i] == '%') {
@@ -139,7 +139,7 @@ std::string decode_url(const std::string& s) {
     return r;
 }
 
-std::string encode_url(const std::string& s) {
+inline std::string encode_url(const std::string& s) {
     std::ostringstream oss;
     for (auto& c : s) {
         if (internal::url_safe_characters.find(c) != std::string::npos) {
@@ -156,6 +156,7 @@ std::string encode_url(const std::string& s) {
 // where '&' can be any delimiter.
 // ki and vi will be converted if they are percent-encoded,
 // which is why the returned values are `string`, not `string_view`.
+inline
 std::pair<
     std::map<std::string, std::string>,
     std::map<std::string, std::vector<std::string>>>
@@ -218,6 +219,7 @@ parse_params(std::string& s, std::size_t start_pos = 0, char delimiter = '&') {
 // this function will convert ki and vi if they are percent-encoded.
 // NOTE: don't misuse this function, it's going to modify
 //       the parameter `s` in place!
+inline
 std::tuple<std::string,
            std::map<std::string, std::string>,
            std::map<std::string, std::vector<std::string>>>

bserv/websocket.hpp

@@ -0,0 +1,65 @@
+#ifndef _WEBSOCKET_HPP
+#define _WEBSOCKET_HPP
+
+#include <boost/beast.hpp>
+#include <boost/asio.hpp>
+#include <boost/json.hpp>
+
+#include <iostream>
+#include <string>
+#include <cstddef>
+#include <cstdlib>
+
+namespace bserv {
+
+namespace beast = boost::beast;
+namespace http = beast::http;
+namespace websocket = beast::websocket;
+namespace asio = boost::asio;
+namespace json = boost::json;
+using asio::ip::tcp;
+
+class websocket_closed
+    : public std::exception {
+public:
+    websocket_closed() {}
+    const char* what() const noexcept { return "websocket session has been closed"; }
+};
+
+class websocket_io_exception
+    : public std::exception {
+private:
+    const std::string msg_;
+public:
+    websocket_io_exception(const std::string& msg) : msg_{msg} {}
+    const char* what() const noexcept { return msg_.c_str(); }
+};
+
+struct websocket_session {
+    const std::string address_;
+    asio::io_context& ioc_;
+    websocket::stream<beast::tcp_stream> ws_;
+    websocket_session(
+        const std::string& address,
+        asio::io_context& ioc,
+        tcp::socket&& socket)
+        : address_{address},
+        ioc_{ioc}, ws_{std::move(socket)} {}
+};
+
+class websocket_server {
+private:
+    websocket_session& session_;
+    asio::yield_context& yield_;
+public:
+    websocket_server(websocket_session& session, asio::yield_context& yield)
+    : session_{session}, yield_{yield} {}
+    std::string read();
+    boost::json::value read_json() { return boost::json::parse(read()); }
+    void write(const std::string& data);
+    void write_json(const boost::json::value& val) { write(boost::json::serialize(val)); }
+};
+
+}  // bserv
+
+#endif  // _WEBSOCKET_HPP

handlers.hpp

@@ -1,7 +1,7 @@
 #ifndef _HANDLERS_HPP
 #define _HANDLERS_HPP
 
-#include <boost/json/src.hpp>
+#include <boost/json.hpp>
 
 #include <string>
 #include <memory>
@@ -209,11 +209,14 @@ boost::json::object user_logout(
     };
 }
 
-boost::json::object send_request(std::shared_ptr<bserv::http_client> client_ptr) {
+boost::json::object send_request(
+        std::shared_ptr<bserv::session_type> session,
+        std::shared_ptr<bserv::http_client> client_ptr,
+        boost::json::object&& params) {
     // post for response:
     // auto res = client_ptr->post(
     //     "localhost", "8080", "/echo", {{"msg", "request"}}
-    // ).get();
+    // );
     // return {{"response", boost::json::parse(res.body())}};
     // -------------------------------------------------------
     // - if it takes longer than 30 seconds (by default) to
@@ -221,15 +224,35 @@ boost::json::object send_request(std::shared_ptr<bserv::http_client> client_ptr)
     // -------------------------------------------------------
     // post for json response (json value, rather than json
     // object, is returned):
-    auto obj = client_ptr->post_for_object(
+    auto obj = client_ptr->post_for_value(
-        "localhost", "8080", "/echo", {{"msg", "request"}}
+        "localhost", "8080", "/echo", {{"request", params}}
-    ).get();
+    );
-    return {{"response", obj}};
+    if (session->count("cnt") == 0) {
+        (*session)["cnt"] = 0;
+    }
+    (*session)["cnt"] = (*session)["cnt"].as_int64() + 1;
+    return {{"response", obj}, {"cnt", (*session)["cnt"]}};
 }
 
 boost::json::object echo(
     boost::json::object&& params) {
-    return params;
+    return {{"echo", params}};
+}
+
+// websocket
+std::nullopt_t ws_echo(
+        std::shared_ptr<bserv::session_type> session,
+        std::shared_ptr<bserv::websocket_server> ws_server) {
+    ws_server->write_json((*session)["cnt"]);
+    while (true) {
+        try {
+            std::string data = ws_server->read();
+            ws_server->write(data);
+        } catch (bserv::websocket_closed&) {
+            break;
+        }
+    }
+    return std::nullopt;
 }
 
 #endif  // _HANDLERS_HPP

main.cpp

@@ -120,9 +120,15 @@ int main(int argc, char* argv[]) {
             bserv::placeholders::db_connection_ptr,
             bserv::placeholders::_1),
         bserv::make_path("/send", &send_request,
-            bserv::placeholders::http_client_ptr),
+            bserv::placeholders::session,
+            bserv::placeholders::http_client_ptr,
+            bserv::placeholders::json_params),
         bserv::make_path("/echo", &echo,
             bserv::placeholders::json_params)
+    }, {
+        bserv::make_path("/echo", &ws_echo,
+            bserv::placeholders::session,
+            bserv::placeholders::websocket_server_ptr)
     }};
     
     return EXIT_SUCCESS;

scripts/request_test.py

@@ -0,0 +1,48 @@
+import uuid
+
+import requests
+
+from multiprocessing import Process
+
+from pprint import pprint
+
+from time import time
+
+# session = requests.session()
+# pprint(session.post("http://localhost:8080/send", json={"id": "abc"}).json())
+# pprint(session.post("http://localhost:8080/send", json={"id": "def"}).json())
+# pprint(session.post("http://localhost:8080/send", json={"id": "ghi"}).json())
+# exit()
+
+P = 100  # number of concurrent processes
+N = 10  # for each process, the number of sessions
+R = 10  # for each session, the number of posts
+
+def test(i):
+    global C
+    # print(f'starting process {i}')
+    for _ in range(N):
+        session = requests.session()
+        for i in range(1, R + 1):
+            session_id = str(uuid.uuid4())
+            if {'cnt': i, 'response': {'echo': {'request': {'id': session_id}}}} \
+                    != session.post("http://localhost:8080/send", json={"id": session_id}).json():
+                print('test failed!')
+    # print(f'exiting process {i}')
+
+processes = [Process(target=test, args=(i, )) for i in range(P)]
+
+print('starting')
+
+start = time()
+
+for p in processes:
+    p.start()
+
+for p in processes:
+    p.join()
+
+end = time()
+
+print('test ended')
+print('elapsed: ', end - start)

scripts/ws_test.py

@@ -0,0 +1,62 @@
+import asyncio
+from multiprocessing import Process
+
+import requests
+import websockets
+
+import random
+import uuid
+from time import time
+
+from pprint import pprint
+
+P = 500
+
+def test():
+
+    async def fun(uri):
+        session = requests.session()
+        n = random.randint(1, 5)
+        # print(n)
+        for i in range(1, n + 1):
+            session_id = str(uuid.uuid4())
+            ret = session.post("http://localhost:8080/send", json={"id": session_id}).json()
+            # print(ret)
+            # print({'cnt': i, 'response': {'echo': {'request': {'id': session_id}}}})
+            if {'cnt': i, 'response': {'echo': {'request': {'id': session_id}}}} != ret:
+                print('post test failed!')
+        sess_id = session.cookies['bsessionid']
+        # print("session id:", sess_id)
+        async with websockets.connect(uri, extra_headers={'Cookie': f"bsessionid={sess_id}"}) as websocket:
+            cnt = int(await websocket.recv())
+            # print(cnt)
+            if cnt != n:
+                print('incorrect cnt')
+            m = random.randint(5, 10)
+            for _ in range(m):
+                session_id = str(uuid.uuid4())
+                await websocket.send(session_id)
+                ret = await websocket.recv()
+                if session_id != ret:
+                    print('ws test failed')
+
+    asyncio.get_event_loop().run_until_complete(
+        fun("ws://localhost:8080/echo")
+    )
+
+processes = [Process(target=test) for _ in range(P)]
+
+print('starting')
+
+start = time()
+
+for p in processes:
+    p.start()
+
+for p in processes:
+    p.join()
+
+end = time()
+
+print('test ended')
+print('elapsed: ', end - start)