3#include "connection_t.hpp"
4#include "../websocket/connection.hpp"
5#include "../../core/http/request.hpp"
6#include "../../core/http/generator.hpp"
8#include <boost/asio/dispatch.hpp>
9#include <boost/beast/core.hpp>
10#include <boost/beast/http.hpp>
11#include <boost/beast/http/detail/type_traits.hpp>
12#include <spdlog/logger.h>
28namespace malloy::server::http
39 template<
class Derived>
44 public std::enable_shared_from_this<request_generator>
49 using h_parser_t = std::unique_ptr<boost::beast::http::request_parser<boost::beast::http::empty_body>>;
50 using header_t = boost::beast::http::request_header<>;
54 header() {
return m_header; }
58 header()
const {
return m_header; }
60 template<
typename Body, std::invocable<malloy::http::request<Body>&&> Callback,
typename SetupCb>
62 body(Callback&& done, SetupCb&& setup)
64 using namespace boost::beast::http;
65 using body_t = std::decay_t<Body>;
66 auto parser = std::make_shared<boost::beast::http::request_parser<body_t>>(std::move(*m_parser));
67 parser->get().base() = m_header;
68 std::invoke(setup, parser->get().body());
70 boost::beast::http::async_read(
71 m_parent->derived().m_stream, m_buffer, *parser,
73 done = std::forward<Callback>(done),
74 p = parser, this_ = this->shared_from_this(),
76 ](
const auto& ec,
auto) {
77 if (ec && m_parent->m_logger) {
78 m_parent->m_logger->error(
"failed to read http request body: '{}'", ec.message());
86 template<
typename Body, std::invocable<malloy::http::request<Body>&&> Callback>
90 return body<Body>(std::forward<Callback>(done), [](
auto){});
94 request_generator(h_parser_t hparser, header_t header, std::shared_ptr<connection> parent, boost::beast::flat_buffer buff) :
95 m_buffer{ std::move(buff) },
96 m_parser{ std::move(hparser) },
97 m_header{ std::move(header) },
98 m_parent{ std::move(parent) }
103 boost::beast::flat_buffer m_buffer;
106 std::shared_ptr<connection> m_parent;
117 using http_conn_t =
const connection_t&;
118 using path = std::filesystem::path;
119 using req_t = std::shared_ptr<request_generator>;
126 websocket(
const path& root,
const req_t& req,
const std::shared_ptr<malloy::server::websocket::connection>&) = 0;
130 http(
const path& root,
const req_t& req, http_conn_t) = 0;
138 std::uint64_t request_body_limit = 100'000'000;
156 std::shared_ptr<spdlog::logger> logger,
157 boost::beast::flat_buffer buffer,
158 std::shared_ptr<handler>
router,
159 std::shared_ptr<const std::filesystem::path> http_doc_root
161 m_buffer(std::move(buffer)),
162 m_logger(std::move(logger)),
163 m_doc_root(std::move(http_doc_root)),
164 m_router(std::move(
router))
168 throw std::runtime_error(
"did not receive a valid logger instance.");
172 throw std::runtime_error(
"did not receive a valid router instance.");
183 std::shared_ptr<spdlog::logger>
197 template<
bool isRequest,
class Body,
class Fields>
199 do_write(boost::beast::http::message<isRequest, Body, Fields>&& msg)
204 auto sp = std::make_shared<boost::beast::http::message<isRequest, Body, Fields>>(std::move(msg));
211 boost::beast::http::async_write(
214 boost::beast::bind_front_handler(
215 &connection::on_write,
216 derived().shared_from_this(),
225 m_logger->trace(
"do_read()");
228 m_parser = std::make_unique<std::decay_t<
decltype(*m_parser)>>();
232 m_parser->body_limit(cfg.request_body_limit);
235 boost::beast::get_lowest_layer(derived().stream()).expires_after(std::chrono::seconds(30));
238 boost::beast::http::async_read_header(
242 boost::beast::bind_front_handler(
243 &connection::on_read,
244 derived().shared_from_this()
250 boost::beast::flat_buffer m_buffer;
255 m_logger->error(
"{}: {} (code: {})", context, ec.message(), ec.value());
259 friend class request_generator;
261 std::shared_ptr<spdlog::logger> m_logger;
262 std::shared_ptr<const std::filesystem::path> m_doc_root;
263 std::shared_ptr<handler> m_router;
264 std::shared_ptr<void> m_response;
267 typename request_generator::h_parser_t m_parser;
278 return static_cast<Derived&
>(*this);
282 on_read(boost::beast::error_code ec, std::size_t bytes_transferred)
284 m_logger->trace(
"on_read(): bytes read: {}", bytes_transferred);
287 if (ec == boost::beast::http::error::end_of_stream || ec == boost::beast::error::timeout)
292 m_logger->error(
"on_read(): {}", ec.message());
297 auto header = m_parser->get().base();
300 auto gen = std::shared_ptr<request_generator>{
new request_generator{ std::move(m_parser), std::move(header), derived().shared_from_this(), std::move(m_buffer) }};
303 if (boost::beast::websocket::is_upgrade(gen->header())) {
304 m_logger->info(
"upgrading HTTP connection to WS connection");
308 boost::beast::get_lowest_layer(derived().
stream()).expires_never();
316 m_router->websocket(*m_doc_root, gen, ws_connection);
322 m_router->http(*m_doc_root, std::move(gen), derived().shared_from_this());
327 on_write(
bool close, boost::beast::error_code ec, std::size_t bytes_transferred)
329 m_logger->trace(
"on_write(): bytes written: {}", bytes_transferred);
333 m_logger->error(
"on_write(): {}", ec.message());
356 m_logger->trace(
"do_close()");
359 derived().do_close();
362 m_logger->info(
"HTTP connection closed gracefully");
Definition: request.hpp:19
Definition: connection.hpp:114
Definition: connection.hpp:45
Definition: connection.hpp:41
void do_write(boost::beast::http::message< isRequest, Body, Fields > &&msg)
Definition: connection.hpp:199
connection(std::shared_ptr< spdlog::logger > logger, boost::beast::flat_buffer buffer, std::shared_ptr< handler > router, std::shared_ptr< const std::filesystem::path > http_doc_root)
Definition: connection.hpp:155
std::shared_ptr< spdlog::logger > logger() const noexcept
Definition: connection.hpp:184
Definition: router.hpp:104
static std::shared_ptr< connection > make(const std::shared_ptr< spdlog::logger > logger, stream &&ws, const std::string &agent_string)
Construct a new connection object.
Definition: connection.hpp:106
Websocket stream. May use TLS.
Definition: stream.hpp:50
boost::beast::basic_stream< boost::asio::ip::tcp, boost::asio::any_io_executor, RatePolicy > stream
Definition: stream.hpp:22
boost::beast::error_code error_code
Error code used to signify errors without throwing. Truthy means it holds an error.
Definition: error.hpp:9
Definition: connection.hpp:137
std::string agent_string
Agent string to use, set by the controller.
Definition: connection.hpp:139