bart
/
network-experiment
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

WIP: apparently pipelining is not a thing anymore, and this now works, sessions are being cleaned up.

This commit is contained in:
Bart Beumer 2025-11-09 23:00:09 +00:00
parent a52b092e43
commit 63be32de7c
12 changed files with 282 additions and 184 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
applications/http-mandelbrot/html/index.html
View File

@ -31,12 +31,12 @@
<div id="map" style="width: 1500px; height: 1500px;"></div> <div id="map" style="width: 100%; height: 100%;"></div>
<script> <script>
const map = L.map('map', {crs: L.CRS.Simple}).setView([0.0, 0.0], 1); const map = L.map('map', {crs: L.CRS.Simple}).setView([0.0, 0.0], 1);
const tiles = L.tileLayer('/render_{x}_{y}_{z}', { const tiles = L.tileLayer('http://localhost:9800/render_{x}_{y}_{z}', {
maxZoom: 1000 maxZoom: 1000
}).addTo(map); }).addTo(map);

35
applications/http-mandelbrot/src/main.cpp
View File

@ -9,8 +9,7 @@
#include <bmrshared/server.hpp> #include <bmrshared/server.hpp>
#include <bmrshared/request_handler_interface.hpp> #include <bmrshared/request_handler_interface.hpp>
#include <bmrshared/directory_request_handler.hpp> #include <bmrshared/directory_request_handler.hpp>
#include <bmrshared/response_promise.hpp> #include <bmrshared/request_response.hpp>
#include <boost/asio/signal_set.hpp> #include <boost/asio/signal_set.hpp>
#include <boost/gil/image.hpp> #include <boost/gil/image.hpp>
#include <boost/gil/image_view.hpp> #include <boost/gil/image_view.hpp>
@ -24,7 +23,7 @@
namespace namespace
{ {
const std::regex regex_renderpath(R"REGEX(\/render_(\d+)_(\d+)_(\d+))REGEX"); const std::regex regex_renderpath(R"REGEX(\/render_(-?\d+)_(-?\d+)_(\d+))REGEX");
@ -106,9 +105,9 @@ void fractal(
} }
constexpr unsigned int config_max_simultanious_requests = 50; constexpr unsigned int config_max_simultanious_requests = 10;
constexpr uint64_t config_request_body_limit = (10 * 1024); constexpr uint64_t config_request_body_limit = (10 * 1024);
constexpr std::chrono::seconds config_request_timeout(30); constexpr std::chrono::seconds config_request_timeout(3);
const boost::asio::ip::tcp::endpoint listen_endpoint(boost::asio::ip::tcp::v4(), 9800); const boost::asio::ip::tcp::endpoint listen_endpoint(boost::asio::ip::tcp::v4(), 9800);
@ -124,15 +123,15 @@ void fractal(
~split_req() override = default; ~split_req() override = default;
void handle_request_http(const address_type& addr, const request_type& req, bmrshared::web::response_promise promise) override void handle_request_http(bmrshared::web::request_response rs) override
{ {
auto& req = rs.get_request();
std::string target = req.target(); std::string target = req.target();
std::cout << "HTTP GET: " << target << std::endl;
if (req.method() != boost::beast::http::verb::get) if (req.method() != boost::beast::http::verb::get)
{ {
// Other methods are not supported for directory acces // Other methods are not supported for directory acces
// //
auto& bad_request = promise.CreateResponse<boost::beast::http::response<boost::beast::http::string_body>>(boost::beast::http::status::bad_request, req.version()); auto& bad_request = rs.create_response<boost::beast::http::response<boost::beast::http::string_body>>(boost::beast::http::status::bad_request, req.version());
bad_request.set(boost::beast::http::field::server, BOOST_BEAST_VERSION_STRING); bad_request.set(boost::beast::http::field::server, BOOST_BEAST_VERSION_STRING);
bad_request.set(boost::beast::http::field::content_type, "text/plain"); bad_request.set(boost::beast::http::field::content_type, "text/plain");
bad_request.prepare_payload(); bad_request.prepare_payload();
@ -149,9 +148,8 @@ void fractal(
double offset_x = std::stoi(base_match[2]); double offset_x = std::stoi(base_match[2]);
double z = std::stoi(base_match[3]); double z = std::stoi(base_match[3]);
auto renderfn = [offset_y, offset_x, z, req, target, promise]() mutable auto renderfn = [offset_y, offset_x, z, req, target, rs, tmp]() mutable
{ {
std::cout << "HTTP inside render function " << target << std::endl;
constexpr int pixel_width = 256; constexpr int pixel_width = 256;
constexpr int pixel_height = 256; constexpr int pixel_height = 256;
constexpr int max_iterations = 255; constexpr int max_iterations = 255;
@ -166,7 +164,7 @@ void fractal(
if(x >= 0 && y >= 0 && x < pixel_width && y < pixel_height) if(x >= 0 && y >= 0 && x < pixel_width && y < pixel_height)
{ {
const auto iter = view.at({x, y}); const auto iter = view.at({x, y});
boost::gil::color_convert(boost::gil::gray8_pixel_t(num_iter%256), *iter); boost::gil::color_convert(boost::gil::gray8_pixel_t(num_iter), *iter);
} }
}; };
@ -186,23 +184,21 @@ void fractal(
std::stringstream out_buffer( std::ios_base::out | std::ios_base::binary ); std::stringstream out_buffer( std::ios_base::out | std::ios_base::binary );
boost::gil::write_view(out_buffer, boost::gil::write_view(out_buffer,
boost::gil::view(image), boost::gil::view(image),
boost::gil::image_write_info<boost::gil::jpeg_tag>(95)); boost::gil::image_write_info<boost::gil::jpeg_tag>(70));
auto& ok = promise.CreateResponse<boost::beast::http::response<boost::beast::http::string_body>>(boost::beast::http::status::ok, req.version()); auto& ok = rs.create_response<boost::beast::http::response<boost::beast::http::string_body>>(boost::beast::http::status::ok, req.version());
ok.set(boost::beast::http::field::server, BOOST_BEAST_VERSION_STRING); ok.set(boost::beast::http::field::server, BOOST_BEAST_VERSION_STRING);
ok.set(boost::beast::http::field::content_type, "image/jpeg"); ok.set(boost::beast::http::field::content_type, "image/jpeg");
ok.body() = out_buffer.str(); ok.body() = out_buffer.str();
ok.keep_alive(req.keep_alive()); ok.keep_alive(req.keep_alive());
ok.prepare_payload(); ok.prepare_payload();
std::cout << " DONE HTTP inside render function " << target << std::endl;
}; };
boost::asio::post(m_ioc, renderfn); boost::asio::post(m_ioc, renderfn);
} }
else else
{ {
m_dir.handle_request_http(addr, req, promise); m_dir.handle_request_http(rs);
} }
} }
@ -246,8 +242,10 @@ int main(int argc, char **argv)
request_handler); request_handler);
}); });
std::vector<std::jthread> threads; std::vector<std::jthread> threads;
while(threads.size() < 4) while(threads.size() < 16)
{ {
threads.emplace_back([&ioc]{ioc.run();}); threads.emplace_back([&ioc]{ioc.run();});
} }
@ -255,4 +253,7 @@ int main(int argc, char **argv)
ioc.run(); ioc.run();
threads.clear(); threads.clear();
} }

2
bmrshared-web/include/bmrshared/directory_request_handler.hpp
View File

@ -17,7 +17,7 @@ class directory_request_handler : public request_handler_interface
explicit directory_request_handler(const std::filesystem::path& document_root); explicit directory_request_handler(const std::filesystem::path& document_root);
~directory_request_handler() override; ~directory_request_handler() override;
void handle_request_http(const address_type& address, const request_type& req, response_promise promise) override; void handle_request_http(request_response rs) override;
void handle_request_websocket_upgrade(const address_type& address, void handle_request_websocket_upgrade(const address_type& address,
const request_type& req, const request_type& req,

4
bmrshared-web/include/bmrshared/request_handler_interface.hpp
View File

@ -7,10 +7,10 @@
// //
#pragma once #pragma once
#include "server.hpp" #include "server.hpp"
#include <bmrshared/request_response.hpp>
namespace bmrshared::web namespace bmrshared::web
{ {
class response_promise;
class request_handler_interface class request_handler_interface
{ {
@ -20,7 +20,7 @@ class request_handler_interface
virtual ~request_handler_interface() = default; virtual ~request_handler_interface() = default;
virtual void handle_request_http(const address_type&, const request_type&, response_promise) = 0; virtual void handle_request_http(request_response rs) = 0;
virtual void handle_request_websocket_upgrade(const address_type&, const request_type&, boost::asio::ip::tcp::socket) = 0; virtual void handle_request_websocket_upgrade(const address_type&, const request_type&, boost::asio::ip::tcp::socket) = 0;
}; };
} // namespace bmrshared::web } // namespace bmrshared::web

104
bmrshared-web/include/bmrshared/request_response.hpp Normal file
View File

@ -0,0 +1,104 @@
// GNU Lesser General Public License v3.0
// Copyright (c) 2025 Bart Beumer <bart@4beumer.nl>
//
// This program is free software; you can redistribute it and/or modify it
// under the terms of the GNU Lesser General Public License v3.0 as published by
// the Free Software Foundation.
//
#pragma once
#include "server.hpp"
#include <boost/beast.hpp>
#include <functional>
#include <memory>
namespace bmrshared::web
{
class request_response final
{
public:
using response_sender = std::function<void(boost::beast::tcp_stream& stream)>;
using request_type = bmrshared::web::server::request_type;
using callback_on_finalize = std::function<void(response_sender send_response)>;
private:
class response_writer_interface
{
public:
virtual ~response_writer_interface();
virtual void write_response(boost::beast::tcp_stream&) = 0;
};
template<typename response_type>
class response_writer final : public response_writer_interface
{
public:
template<typename... TArgs>
response_writer(TArgs&&... args)
: m_response(std::forward<TArgs>(args)...)
{}
~response_writer() override = default;
void write_response(boost::beast::tcp_stream& stream) override
{
boost::beast::http::write(stream, m_response);
}
response_type& response()
{
return m_response;
}
private:
response_type m_response;
};
class internal_request_response final
{
public:
// Disabling copying and moving.
internal_request_response() = delete;
internal_request_response(const internal_request_response&) = delete;
internal_request_response(internal_request_response&&) = delete;
internal_request_response& operator=(const internal_request_response&) = delete;
internal_request_response& operator=(internal_request_response&&) = delete;
internal_request_response(request_type request, callback_on_finalize cb_finalize);
~internal_request_response();
const request_type& get_request() const;
void set_response_sender(std::unique_ptr<response_writer_interface> response_writer);
private:
request_type m_request;
callback_on_finalize m_cb_finalize;
std::unique_ptr<response_writer_interface> m_response_writer;
};
public:
request_response() = delete;
request_response(const request_response&) = default;
request_response(request_response&&) = default;
request_response& operator=(const request_response&) = default;
request_response& operator=(request_response&&) = default;
request_response(request_type request, callback_on_finalize cb_finalize);
~request_response();
template<typename response_type, typename... arg_types>
response_type& create_response(arg_types&&... args)
{
auto created = std::make_unique<response_writer<response_type>>(std::forward<arg_types>(args)...);
auto& resp = created->response();
m_internal->set_response_sender(std::move(created));
return resp;
}
const request_type& get_request() const;
private:
std::shared_ptr<internal_request_response> m_internal;
};
}

75
bmrshared-web/include/bmrshared/response_promise.hpp
View File

@ -1,75 +0,0 @@
// GNU Lesser General Public License v3.0
// Copyright (c) 2025 Bart Beumer <bart@4beumer.nl>
//
// This program is free software; you can redistribute it and/or modify it
// under the terms of the GNU Lesser General Public License v3.0 as published by
// the Free Software Foundation.
//
#pragma once
#include <boost/asio/io_context.hpp>
#include <boost/asio/ip/tcp.hpp>
#include <boost/beast.hpp>
#include <functional>
#include <utility>
namespace bmrshared::web
{
class response_promise final
{
private:
class response_writer_interface
{
public:
virtual ~response_writer_interface() = default;
virtual void write_response(boost::beast::tcp_stream&) = 0;
};
template<typename TResponse>
class response_writer : public response_writer_interface
{
public:
template<typename... TArgs>
response_writer(TArgs... args)
: m_response(args...)
{}
~response_writer() override = default;
void write_response(boost::beast::tcp_stream& stream) override
{
boost::beast::http::write(stream, m_response);
}
TResponse& response()
{
return m_response;
}
private:
TResponse m_response;
};
public:
using response_sender = std::function<void(boost::beast::tcp_stream& stream)>;
using callback_on_response = std::function<void(response_sender fnSendResponse)>;
response_promise() = delete;
explicit response_promise(callback_on_response cbOnResponse);
~response_promise();
template<typename TResponse, typename... TArgs>
TResponse& CreateResponse(TArgs&&... args)
{
auto created = std::make_unique<response_writer<TResponse>>(std::forward<TArgs>(args)...);
auto& resp = created->response();
m_response_writer = std::move(created);
return resp;
}
private:
callback_on_response m_call_on_response;
std::shared_ptr<response_writer_interface> m_response_writer;
};
} // namespace bmrshared::web

2
bmrshared-web/lib/CMakeLists.txt
View File

@ -7,7 +7,7 @@ add_library(
${PROJECT_NAME} ${PROJECT_NAME}
directory_request_handler.cpp directory_request_handler.cpp
internal_server.cpp internal_server.cpp
response_promise.cpp request_response.cpp
server.cpp server.cpp
) )

24
bmrshared-web/lib/directory_request_handler.cpp
View File

@ -9,7 +9,6 @@
#include <algorithm> #include <algorithm>
#include <array> #include <array>
#include <bmrshared/magic_file_info.hpp> #include <bmrshared/magic_file_info.hpp>
#include <bmrshared/response_promise.hpp>
#include <filesystem> #include <filesystem>
#include <future> #include <future>
#include <iomanip> #include <iomanip>
@ -42,15 +41,14 @@ directory_request_handler::directory_request_handler(const std::filesystem::path
directory_request_handler::~directory_request_handler() = default; directory_request_handler::~directory_request_handler() = default;
void directory_request_handler::handle_request_http(const address_type& address, void directory_request_handler::handle_request_http(request_response rs)
const request_type& req,
bmrshared::web::response_promise promise)
{ {
const auto& req = rs.get_request();
if ((req.method() != http::verb::get) && (req.method() != http::verb::head)) if ((req.method() != http::verb::get) && (req.method() != http::verb::head))
{ {
// Other methods are not supported for directory acces // Other methods are not supported for directory acces
// //
auto& bad_request = promise.CreateResponse<http::response<http::string_body>>(http::status::bad_request, req.version()); auto& bad_request = rs.create_response<http::response<http::string_body>>(http::status::bad_request, req.version());
bad_request.set(http::field::server, BOOST_BEAST_VERSION_STRING); bad_request.set(http::field::server, BOOST_BEAST_VERSION_STRING);
bad_request.set(http::field::content_type, "text/plain"); bad_request.set(http::field::content_type, "text/plain");
bad_request.prepare_payload(); bad_request.prepare_payload();
@ -85,7 +83,7 @@ void directory_request_handler::handle_request_http(const address_type& address,
if (!found_file) if (!found_file)
{ {
auto& not_found = promise.CreateResponse<http::response<http::string_body>>(http::status::not_found, req.version()); auto& not_found = rs.create_response<http::response<http::string_body>>(http::status::not_found, req.version());
not_found.set(http::field::server, BOOST_BEAST_VERSION_STRING); not_found.set(http::field::server, BOOST_BEAST_VERSION_STRING);
not_found.set(http::field::content_type, "text/plain"); not_found.set(http::field::content_type, "text/plain");
not_found.prepare_payload(); not_found.prepare_payload();
@ -100,7 +98,7 @@ void directory_request_handler::handle_request_http(const address_type& address,
body.open(found_file->native().c_str(), beast::file_mode::scan, ec); body.open(found_file->native().c_str(), beast::file_mode::scan, ec);
if (ec) if (ec)
{ {
auto& internal = promise.CreateResponse<http::response<http::string_body>>(http::status::internal_server_error, req.version()); auto& internal = rs.create_response<http::response<http::string_body>>(http::status::internal_server_error, req.version());
internal.set(http::field::server, BOOST_BEAST_VERSION_STRING); internal.set(http::field::server, BOOST_BEAST_VERSION_STRING);
internal.set(http::field::content_type, "text/plain"); internal.set(http::field::content_type, "text/plain");
internal.prepare_payload(); internal.prepare_payload();
@ -116,7 +114,7 @@ void directory_request_handler::handle_request_http(const address_type& address,
if (req.count(http::field::if_modified_since) != 0 && (req[http::field::if_modified_since] == last_modified)) if (req.count(http::field::if_modified_since) != 0 && (req[http::field::if_modified_since] == last_modified))
{ {
auto& head = promise.CreateResponse<http::response<http::string_body>>(http::status::not_modified, req.version()); auto& head = rs.create_response<http::response<http::string_body>>(http::status::not_modified, req.version());
head.set(http::field::server, BOOST_BEAST_VERSION_STRING); head.set(http::field::server, BOOST_BEAST_VERSION_STRING);
head.keep_alive(req.keep_alive()); head.keep_alive(req.keep_alive());
return; return;
@ -126,7 +124,7 @@ void directory_request_handler::handle_request_http(const address_type& address,
// Okay, we have a file so let us send the correct response. // Okay, we have a file so let us send the correct response.
if (req.method() == http::verb::head) if (req.method() == http::verb::head)
{ {
auto& head = promise.CreateResponse<http::response<http::string_body>>(http::status::ok, req.version()); auto& head = rs.create_response<http::response<http::string_body>>(http::status::ok, req.version());
head.set(http::field::server, BOOST_BEAST_VERSION_STRING); head.set(http::field::server, BOOST_BEAST_VERSION_STRING);
head.set(http::field::content_type, mime_type); head.set(http::field::content_type, mime_type);
head.set(http::field::last_modified, last_modified); head.set(http::field::last_modified, last_modified);
@ -136,8 +134,8 @@ void directory_request_handler::handle_request_http(const address_type& address,
return; return;
} }
else if (req.method() == http::verb::get) else if (req.method() == http::verb::get)
{/* {
auto& head = promise.CreateResponse<http::response<http::file_body>>(std::piecewise_construct, http::response<http::file_body> head(std::piecewise_construct,
std::make_tuple(std::move(body)), std::make_tuple(std::move(body)),
std::make_tuple(http::status::ok, req.version())); std::make_tuple(http::status::ok, req.version()));
@ -146,8 +144,10 @@ void directory_request_handler::handle_request_http(const address_type& address,
head.set(http::field::last_modified, last_modified); head.set(http::field::last_modified, last_modified);
head.set(http::field::cache_control, "public, max-age=360"); head.set(http::field::cache_control, "public, max-age=360");
head.set(http::field::content_length, std::to_string(file_size)); head.set(http::field::content_length, std::to_string(file_size));
head.keep_alive(req.keep_alive());*/ head.keep_alive(req.keep_alive());
rs.create_response<http::response<http::file_body>>(std::move(head));
return; return;
} }
} }

104
bmrshared-web/lib/internal_server.cpp
View File

@ -8,7 +8,7 @@
#include "internal_server.h" #include "internal_server.h"
#include <bmrshared/request_handler_interface.hpp> #include <bmrshared/request_handler_interface.hpp>
#include <algorithm> #include <algorithm>
#include <iostream>
using bmrshared::web::detail::internal_server; using bmrshared::web::detail::internal_server;
@ -19,6 +19,7 @@ internal_server::internal_server(boost::asio::io_context& io_context,
unsigned int max_simultaneous_requests, unsigned int max_simultaneous_requests,
request_handler_interface& handler) request_handler_interface& handler)
: m_io_context(io_context) : m_io_context(io_context)
, m_strand(io_context)
, m_endPoint(listen_endpoint) , m_endPoint(listen_endpoint)
, m_incoming_request_timeout(incoming_request_timeout) , m_incoming_request_timeout(incoming_request_timeout)
, m_max_simultaneous_requests(max_simultaneous_requests) , m_max_simultaneous_requests(max_simultaneous_requests)
@ -50,7 +51,14 @@ void internal_server::async_accept()
auto shared_server = weak_server.lock(); auto shared_server = weak_server.lock();
if(shared_server) if(shared_server)
{ {
shared_server->do_accept(ec, std::move(socket)); boost::asio::post(shared_server->m_strand, [weak_server, ec, s = std::make_shared<boost::asio::ip::tcp::socket>(std::move(socket))]
{
auto shared_server = weak_server.lock();
if (shared_server)
{
shared_server->do_accept(ec, std::move(*s));
}
});
} }
}); });
} }
@ -64,46 +72,48 @@ void internal_server::do_accept(boost::system::error_code ec, boost::asio::ip::t
} }
else else
{ {
m_sessions.push_back(std::make_shared<http_session>(std::move(socket))); auto sessionptr = std::make_shared<http_session>(std::move(socket));
session_process_queued(); m_sessions.push_back(sessionptr);
async_accept();
} }
async_accept();
session_process_queued();
} }
void internal_server::session_process_queued() void internal_server::session_process_queued()
{ {
// Determine how many outstanding requests we have.
std::size_t count_responding = 0;
for(const auto& session : m_sessions)
{
if (session->m_state == session_state::wait_response)
{
count_responding++;
}
}
// Send out all processed responses. // Send out all processed responses.
for (auto& session : m_sessions) for (auto& session : m_sessions)
{ {
if (session) if (session)
{ {
http_session& s = *session; http_session& s = *session;
auto& our = s.m_outstanding_request;
auto& our = s.m_outstanding_requests; if (our && our->m_response_sender)
while (!our.empty() && our.front() && our.front()->m_response_sender)
{ {
auto& resp = our.front(); (*our->m_response_sender)(s.stream);
(*resp->m_response_sender)(s.stream); our.reset();
our.pop(); s.m_state = session_state::queue_reading;
} }
} }
} }
// Determine how many outstanding requests we have.
std::size_t count_responding = 0;
for(const auto& session : m_sessions)
{
count_responding += session->m_outstanding_requests.size();
}
// Allow for more outstanding requests unless we are already over the maximum. // Allow for more outstanding requests unless we are already over the maximum.
if (count_responding < m_max_simultaneous_requests) if (count_responding < m_max_simultaneous_requests)
{ {
for(auto& session : m_sessions) for(auto& session : m_sessions)
{ {
if (!session || (count_responding > m_max_simultaneous_requests)) if (!session)
{ {
// do nothing. // do nothing.
} }
@ -116,12 +126,19 @@ void internal_server::session_process_queued()
s.stream.expires_after(m_incoming_request_timeout); s.stream.expires_after(m_incoming_request_timeout);
auto handler = [weak_server = weak_from_this(), session](boost::beast::error_code ec, std::size_t size) auto handler = [weak_server = weak_from_this(), session](boost::beast::error_code ec, std::size_t size)
{
auto shared_server = weak_server.lock();
if (shared_server)
{
boost::asio::post(shared_server->m_strand, [weak_server, session, ec, size]
{ {
auto shared_server = weak_server.lock(); auto shared_server = weak_server.lock();
if (shared_server) if (shared_server)
{ {
shared_server->session_on_read(session, ec, size); shared_server->session_on_read(session, ec, size);
} }
});
}
}; };
boost::beast::http::async_read(s.stream, s.buffer, *s.parser, handler); boost::beast::http::async_read(s.stream, s.buffer, *s.parser, handler);
} }
@ -135,13 +152,7 @@ void internal_server::session_on_read(std::shared_ptr<http_session> session,
{ {
http_session& s = *session; http_session& s = *session;
if (ec == boost::beast::http::error::end_of_stream) if (ec)
{
// End of stream. Shutdown and remove http session
s.stream.socket().shutdown(boost::asio::ip::tcp::socket::shutdown_send, ec);
session_remove(session);
}
else if (ec)
{ {
// Any other kind of error, also shutdown and remove http_session // Any other kind of error, also shutdown and remove http_session
s.stream.socket().shutdown(boost::asio::ip::tcp::socket::shutdown_send, ec); s.stream.socket().shutdown(boost::asio::ip::tcp::socket::shutdown_send, ec);
@ -156,29 +167,42 @@ void internal_server::session_on_read(std::shared_ptr<http_session> session,
} }
else else
{ {
std::shared_ptr<http_request> req = s.m_outstanding_requests.emplace(std::make_shared<http_request>()); std::shared_ptr<http_request> req = std::make_shared<http_request>();
s.m_outstanding_request = req;
// A regular HTTP request. // A regular HTTP request.
auto finalize_response = [session, weak_server = weak_from_this(), req](const response_promise::response_sender& rs) auto finalize_response = [session, weak_server = weak_from_this(), req](const request_response::response_sender& rs)
{ {
// We have a response, add it to our spot in the queue. // We have a response, add it to our spot in the queue.
auto shared_server = weak_server.lock(); auto shared_server = weak_server.lock();
if (shared_server) if (shared_server)
{ {
// We have a response, add it to our spot in the queue.
req->m_response_sender = rs;
// Queue processing of the queue so we can continue. // Queue processing of the queue so we can continue.
boost::asio::post(shared_server->m_io_context,[shared_server]{shared_server->session_process_queued();}); boost::asio::post(shared_server->m_strand,
[shared_server, req, rs]
{
req->m_response_sender = rs;
shared_server->session_process_queued();
});
} }
}; };
const auto& address = session->stream.socket().remote_endpoint().address(); const auto& address = session->stream.socket().remote_endpoint().address();
m_handler.handle_request_http(address, session->parser->get(), response_promise(finalize_response)); const auto& request = session->parser->get();
s.m_state = session_state::queue_reading; auto rs = request_response(request, finalize_response);
session_process_queued(); auto& internal = rs.create_response<boost::beast::http::response<boost::beast::http::string_body>>(boost::beast::http::status::internal_server_error, request.version());
internal.set(boost::beast::http::field::server, BOOST_BEAST_VERSION_STRING);
internal.set(boost::beast::http::field::content_type, "text/plain");
internal.prepare_payload();
internal.body() = "Internal server error.";
internal.keep_alive(request.keep_alive());
m_handler.handle_request_http(rs);
s.m_state = session_state::wait_response;
} }
session_process_queued();
} }
void internal_server::session_finalize_response(std::shared_ptr<http_session> session) void internal_server::session_finalize_response(std::shared_ptr<http_session> session)
@ -192,5 +216,11 @@ void internal_server::session_finalize_response(std::shared_ptr<http_session> se
void internal_server::session_remove(std::shared_ptr<http_session> session) void internal_server::session_remove(std::shared_ptr<http_session> session)
{ {
m_sessions.erase(std::remove(m_sessions.begin(), m_sessions.end(), session), m_sessions.end()); m_sessions.erase(std::remove_if(m_sessions.begin(),
m_sessions.end(),
[session](const std::shared_ptr<http_session>& s)
{
return session.get() == s.get();
}),
m_sessions.end());
} }

21
bmrshared-web/lib/internal_server.h
View File

@ -6,35 +6,43 @@
// the Free Software Foundation. // the Free Software Foundation.
// //
#pragma once #pragma once
#include <bmrshared/response_promise.hpp>
#include <bmrshared/server.hpp> #include <bmrshared/server.hpp>
#include <bmrshared/request_response.hpp>
#include <boost/asio/io_context.hpp> #include <boost/asio/io_context.hpp>
#include <boost/asio/strand.hpp>
#include <boost/asio/ip/tcp.hpp> #include <boost/asio/ip/tcp.hpp>
#include <chrono> #include <chrono>
#include <memory> #include <memory>
#include <queue> #include <queue>
#include <iostream>
namespace bmrshared::web { namespace bmrshared::web {
namespace detail { namespace detail {
enum class session_state { enum class session_state {
queue_reading, queue_reading,
reading reading,
wait_response
}; };
struct http_request struct http_request
{ {
std::optional<response_promise::response_sender> m_response_sender; std::optional<request_response::response_sender> m_response_sender;
}; };
struct http_session { struct http_session {
explicit http_session(boost::asio::ip::tcp::socket s) explicit http_session(boost::asio::ip::tcp::socket s)
: stream(std::move(s)) : stream(std::move(s))
{} {
}
~http_session()
{
}
boost::beast::tcp_stream stream; boost::beast::tcp_stream stream;
boost::beast::flat_buffer buffer; boost::beast::flat_buffer buffer;
boost::optional<boost::beast::http::request_parser<boost::beast::http::string_body>> parser; boost::optional<boost::beast::http::request_parser<boost::beast::http::string_body>> parser;
session_state m_state; session_state m_state = session_state::queue_reading;
std::queue<std::shared_ptr<http_request>> m_outstanding_requests; std::shared_ptr<http_request> m_outstanding_request;
}; };
class internal_server final : public std::enable_shared_from_this<internal_server> { class internal_server final : public std::enable_shared_from_this<internal_server> {
@ -64,6 +72,7 @@ namespace detail {
private: private:
boost::asio::io_context& m_io_context; boost::asio::io_context& m_io_context;
boost::asio::io_context::strand m_strand;
boost::asio::ip::tcp::endpoint m_endPoint; boost::asio::ip::tcp::endpoint m_endPoint;
std::chrono::seconds m_incoming_request_timeout; std::chrono::seconds m_incoming_request_timeout;
unsigned int m_max_simultaneous_requests; unsigned int m_max_simultaneous_requests;

56
bmrshared-web/lib/request_response.cpp Normal file
View File

@ -0,0 +1,56 @@
// GNU Lesser General Public License v3.0
// Copyright (c) 2025 Bart Beumer <bart@4beumer.nl>
//
// This program is free software; you can redistribute it and/or modify it
// under the terms of the GNU Lesser General Public License v3.0 as published by
// the Free Software Foundation.
//
#include <bmrshared/request_response.hpp>
#include <iostream>
using bmrshared::web::request_response;
request_response::response_writer_interface::~response_writer_interface() = default;
request_response::internal_request_response::internal_request_response(request_type request, callback_on_finalize cb_finalize)
: m_request(std::move(request))
, m_cb_finalize(std::move(cb_finalize))
, m_response_writer()
{
}
request_response::internal_request_response::~internal_request_response()
{
if (m_response_writer)
{
std::shared_ptr<response_writer_interface> shared_writer(std::move(m_response_writer));
m_cb_finalize(
[shared_writer](boost::beast::tcp_stream& stream)
{
shared_writer->write_response(stream);
});
}
}
const request_response::request_type& request_response::internal_request_response::get_request() const
{
return m_request;
}
void request_response::internal_request_response::set_response_sender(std::unique_ptr<response_writer_interface> response_writer)
{
m_response_writer = std::move(response_writer);
}
request_response::request_response(request_type request, callback_on_finalize cb_finalize)
: m_internal(std::make_unique<internal_request_response>(std::move(request), std::move(cb_finalize)))
{
}
request_response::~request_response() = default;
const request_response::request_type& request_response::get_request() const
{
return m_internal->get_request();
}

27
bmrshared-web/lib/response_promise.cpp
View File

@ -1,27 +0,0 @@
// GNU Lesser General Public License v3.0
// Copyright (c) 2025 Bart Beumer <bart@4beumer.nl>
//
// This program is free software; you can redistribute it and/or modify it
// under the terms of the GNU Lesser General Public License v3.0 as published by
// the Free Software Foundation.
//
#include <bmrshared/response_promise.hpp>
using bmrshared::web::response_promise;
response_promise::response_promise(callback_on_response cbOnResponse)
: m_call_on_response(std::move(cbOnResponse))
{
}
response_promise::~response_promise()
{
if (m_response_writer)
{
m_call_on_response(
[r = m_response_writer](boost::beast::tcp_stream& stream)
{
r->write_response(stream);
});
}
}