Add wrap_executor for capy::executor integration

- Add wrap_executor() to adapt Asio executors to capy::executor
- Integrate capy::executor into route_params and spawn for executor affinity
- Refactor http_stream, plain_worker, and worker_ssl for new executor model
- Add comprehensive wrap_executor unit tests
- Remove obsolete route_handler_asio.hpp and post_work files

Author: vinniefalco

CLAUDE.md

@@ -8,7 +8,6 @@
 //
 
 #include "certificate.hpp"
-#include "post_work.hpp"
 #include "serve_detached.hpp"
 #include "serve_log_admin.hpp"
 #include <boost/beast2/asio_io_context.hpp>
@@ -17,9 +16,12 @@
 #include <boost/beast2/server/serve_static.hpp>
 #include <boost/beast2/error.hpp>
 #include <boost/capy/application.hpp>
+#include <boost/capy/async_result.hpp>
+#include <boost/capy/thread_pool.hpp>
 #include <boost/http_proto/request_parser.hpp>
 #include <boost/http_proto/serializer.hpp>
 #include <boost/http_proto/server/cors.hpp>
+#include <boost/capy/bcrypt.hpp>
 #include <boost/capy/brotli/decode.hpp>
 #include <boost/capy/brotli/encode.hpp>
 #include <boost/capy/zlib/deflate.hpp>
@@ -30,6 +32,8 @@
 namespace boost {
 namespace beast2 {
 
+capy::thread_pool g_tp;
+
 void install_services(capy::application& app)
 {
 #ifdef BOOST_CAPY_HAS_BROTLI
@@ -102,7 +106,7 @@ struct do_json_rpc
             return http::route::next;
         return rp.read_body(
             json_sink(),
-            [this, &rp](
+            [this](
                 json::value jv) ->
                     http::route_result
             {
@@ -121,6 +125,46 @@ struct do_json_rpc
 
 };
 
+#ifdef BOOST_BEAST2_HAS_CORO
+auto
+my_coro(
+    http::route_params& rp) ->
+        capy::task<http::route_result>
+{
+    (void)rp;
+    asio::thread_pool tp(1);
+    co_await capy::make_async_result<void>(
+        [&tp](auto&& handler)
+        {
+            asio::post(tp.get_executor(),
+                [handler = std::move(handler)]() mutable
+                {
+                    // Simulate some asynchronous work
+                    std::this_thread::sleep_for(std::chrono::seconds(1));
+                    handler();
+                });
+        });
+    co_return http::route::next;
+}
+
+auto
+do_bcrypt(
+    http::route_params& rp) ->
+        capy::task<http::route_result>
+{
+    std::string password = "boost";
+    //auto rv = co_await capy::bcrypt::async_hash(password);
+    co_await g_tp.get_executor().submit(
+        []()
+        {
+            std::this_thread::sleep_for(std::chrono::seconds(1));
+        });
+    co_return http::route::next;
+}
+
+
+#endif
+
 int server_main( int argc, char* argv[] )
 {
     try
@@ -150,6 +194,14 @@ int server_main( int argc, char* argv[] )
             http::cors(opts),
             do_json_rpc()
         );
+#ifdef BOOST_BEAST2_HAS_CORO
+        srv.wwwroot.use(
+            "/spawn",
+            http::co_route(my_coro));
+        srv.wwwroot.use(
+            "/bcrypt",
+            http::co_route(do_bcrypt));
+#endif
         srv.wwwroot.use("/", serve_static( argv[3] ));
 
         app.start();

example/server/main.cpp

@@ -34,6 +34,7 @@
 #include <boost/beast2/client.hpp>
 #include <boost/beast2/endpoint.hpp>
 #include <boost/beast2/error.hpp>
+#include <boost/beast2/wrap_executor.hpp>
 #include <boost/beast2/format.hpp>
 #include <boost/beast2/log_service.hpp>
 #include <boost/beast2/logger.hpp>

example/server/post_work.cpp

@@ -39,9 +39,9 @@ namespace beast2 {
 //------------------------------------------------
 
 #if defined(__cpp_lib_coroutine) && __cpp_lib_coroutine >= 201902L
-# define BOOST_BEAST_HAS_CORO 1
+# define BOOST_BEAST2_HAS_CORO 1
 #elif defined(__cpp_impl_coroutine) && __cpp_impl_coroutines >= 201902L
-# define BOOST_BEAST_HAS_CORO 1
+# define BOOST_BEAST2_HAS_CORO 1
 #endif
 
 //------------------------------------------------

example/server/post_work.hpp

@@ -12,13 +12,14 @@
 
 #include <boost/beast2/detail/config.hpp>
 #include <boost/beast2/log_service.hpp>
+#include <boost/beast2/error.hpp>
 #include <boost/beast2/format.hpp>
 #include <boost/beast2/read.hpp>
-#include <boost/beast2/write.hpp>
 #include <boost/beast2/server/any_lambda.hpp>
 #include <boost/beast2/server/route_handler_asio.hpp>
 #include <boost/beast2/server/router_asio.hpp>
-#include <boost/beast2/error.hpp>
+#include <boost/beast2/wrap_executor.hpp>
+#include <boost/beast2/write.hpp>
 #include <boost/beast2/detail/except.hpp>
 #include <boost/capy/application.hpp>
 #include <boost/http_proto/request_parser.hpp>
@@ -101,7 +102,8 @@ class http_stream
         std::size_t bytes_transferred);
     void on_complete();
     http::resumer do_suspend() override;
-    void do_resume(http::route_result const& ec) override;
+    void do_resume(http::route_result const& rv) override;
+    void do_resume(std::exception_ptr ep) override;
     void do_close();
     void do_fail(core::string_view s,
         system::error_code const& ec);
@@ -182,6 +184,7 @@ http_stream(
 
     rp_.serializer = http::serializer(app);
     rp_.suspend = http::suspender(*this);
+    rp_.ex = wrap_executor(stream_.get_executor());
 }
 
 // called to start a new HTTP session.
@@ -444,6 +447,40 @@ do_resume(
         });
 }
 
+// called by resume(ep)
+template<class AsyncStream>
+void
+http_stream<AsyncStream>::
+do_resume(
+    std::exception_ptr ep)
+{
+    asio::dispatch(
+        stream_.get_executor(),
+        [this, ep]
+        {
+            BOOST_ASSERT(pwg_.get() != nullptr);
+            pwg_.reset();
+
+            rp_.status(http::status::internal_server_error);
+            try
+            {
+                std::rethrow_exception(ep);
+            }
+            catch(std::exception const& e)
+            {
+                std::string s;
+                format_to(s, "An internal server error occurred: {}", e.what());
+                rp_.set_body(s);
+            }
+            catch(...)
+            {
+                rp_.set_body("An internal server error occurred");
+            }
+            rp_.res.set_keep_alive(false);
+            do_write();
+        });
+}
+
 // called when a non-recoverable error occurs
 template<class AsyncStream>
 void