Refactor HttpMessage into generalized templated types

This adds generalized IncomingHttpMessage and OutgoingHttpMessage templates
that support different types of streams (via a std::variant) and can both
be used for either requests or responses.

The tacked on metadata from the old HttpRequest and server connection from
the old HttpServerConnection have been moved to HttpApi(Request|Response)
classes that derive from the above generalized message types.

Author: jschmidt-icinga

lib/base/tlsstream.hpp

@@ -13,6 +13,7 @@
 #include <atomic>
 #include <memory>
 #include <utility>
+#include <variant>
 #include <boost/asio/buffered_stream.hpp>
 #include <boost/asio/io_context.hpp>
 #include <boost/asio/ip/tcp.hpp>
@@ -122,9 +123,9 @@ class AsioTlsStream : public boost::asio::buffered_stream<UnbufferedAsioTlsStrea
 	}
 };
 
-typedef boost::asio::buffered_stream<boost::asio::ip::tcp::socket> AsioTcpStream;
-typedef std::pair<Shared<AsioTlsStream>::Ptr, Shared<AsioTcpStream>::Ptr> OptionalTlsStream;
-
+using AsioTcpStream = boost::asio::buffered_stream<boost::asio::ip::tcp::socket>;
+using OptionalTlsStream = std::pair<Shared<AsioTlsStream>::Ptr, Shared<AsioTcpStream>::Ptr>;
+using AsioTlsOrTcpStream = std::variant<Shared<AsioTlsStream>::Ptr, Shared<AsioTcpStream>::Ptr>;
 }
 
 #endif /* TLSSTREAM_H */

lib/remote/httpmessage.cpp

@@ -1,7 +1,6 @@
 /* Icinga 2 | (c) 2025 Icinga GmbH | GPLv2+ */
 
 #include "remote/httpmessage.hpp"
-#include "base/io-engine.hpp"
 #include "base/json.hpp"
 #include "remote/httputility.hpp"
 #include "remote/url.hpp"
@@ -27,10 +26,15 @@ constexpr std::size_t l_FlushThreshold = 128UL * 1024UL;
  *
  * @ingroup base
  */
+template<typename Message>
 class HttpResponseJsonWriter : public AsyncJsonWriter
 {
 public:
-	explicit HttpResponseJsonWriter(HttpApiResponse& msg) : m_Message{msg}
+	HttpResponseJsonWriter(const HttpResponseJsonWriter&) = delete;
+	HttpResponseJsonWriter(HttpResponseJsonWriter&&) = delete;
+	HttpResponseJsonWriter& operator=(const HttpResponseJsonWriter&) = delete;
+	HttpResponseJsonWriter& operator=(HttpResponseJsonWriter&&) = delete;
+	explicit HttpResponseJsonWriter(Message& msg) : m_Message{msg}
 	{
 		m_Message.body().Start();
 #if BOOST_VERSION >= 107000
@@ -59,23 +63,37 @@ class HttpResponseJsonWriter : public AsyncJsonWriter
 	}
 
 private:
-	HttpApiResponse& m_Message;
+	Message& m_Message;
 };
 
-HttpApiRequest::HttpApiRequest(Shared<AsioTlsStream>::Ptr stream) : m_Stream(std::move(stream))
+template<bool isRequest, typename Body, typename StreamVariant>
+IncomingHttpMessage<isRequest, Body, StreamVariant>::IncomingHttpMessage(StreamVariant stream)
+	: m_Stream(std::move(stream))
 {
 }
 
-void HttpApiRequest::ParseHeader(boost::beast::flat_buffer& buf, boost::asio::yield_context yc)
+template<bool isRequest, typename Body, typename StreamVariant>
+void IncomingHttpMessage<isRequest, Body, StreamVariant>::ParseHeader(
+	boost::beast::flat_buffer& buf,
+	boost::asio::yield_context yc
+)
 {
-	boost::beast::http::async_read_header(*m_Stream, buf, m_Parser, yc);
-	base() = m_Parser.get().base();
+	std::visit([&](auto& stream) { boost::beast::http::async_read_header(*stream, buf, m_Parser, yc); }, m_Stream);
+	Base::base() = m_Parser.get().base();
 }
 
-void HttpApiRequest::ParseBody(boost::beast::flat_buffer& buf, boost::asio::yield_context yc)
+template<bool isRequest, typename Body, typename StreamVariant>
+void IncomingHttpMessage<isRequest, Body, StreamVariant>::ParseBody(
+	boost::beast::flat_buffer& buf,
+	boost::asio::yield_context yc
+)
+{
+	std::visit([&](auto& stream) { boost::beast::http::async_read(*stream, buf, m_Parser, yc); }, m_Stream);
+	Base::body() = std::move(m_Parser.release().body());
+}
+
+HttpApiRequest::HttpApiRequest(Shared<AsioTlsStream>::Ptr stream) : IncomingHttpMessage(std::move(stream))
 {
-	boost::beast::http::async_read(*m_Stream, buf, m_Parser, yc);
-	body() = std::move(m_Parser.release().body());
 }
 
 ApiUser::Ptr HttpApiRequest::User() const
@@ -111,49 +129,72 @@ void HttpApiRequest::DecodeParams()
 	m_Params = HttpUtility::FetchRequestParameters(m_Url, body());
 }
 
-HttpApiResponse::HttpApiResponse(Shared<AsioTlsStream>::Ptr stream, HttpServerConnection::Ptr server)
-	: m_Server(std::move(server)), m_Stream(std::move(stream))
+template<bool isRequest, typename Body, typename StreamVariant>
+OutgoingHttpMessage<isRequest, Body, StreamVariant>::OutgoingHttpMessage(StreamVariant stream)
+	: m_Stream(std::move(stream))
 {
 }
 
-void HttpApiResponse::Clear()
+template<bool isRequest, typename Body, typename StreamVariant>
+void OutgoingHttpMessage<isRequest, Body, StreamVariant>::Clear()
 {
 	ASSERT(!m_SerializationStarted);
-	boost::beast::http::response<body_type>::operator=({});
+	Base::operator=({});
 }
 
-void HttpApiResponse::Flush(boost::asio::yield_context yc)
+template<bool isRequest, typename Body, typename StreamVariant>
+void OutgoingHttpMessage<isRequest, Body, StreamVariant>::Flush(boost::asio::yield_context yc, bool finish)
 {
-	if (!chunked() && !has_content_length()) {
+	if (!Base::chunked() && !Base::has_content_length()) {
 		ASSERT(!m_SerializationStarted);
-		prepare_payload();
+		Base::prepare_payload();
 	}
 
-	m_SerializationStarted = true;
-
-	if (!m_Serializer.is_header_done()) {
-		boost::beast::http::write_header(*m_Stream, m_Serializer);
-	}
+	std::visit(
+		[&](auto& stream) {
+			m_SerializationStarted = true;
+
+			if (!m_Serializer.is_header_done()) {
+				boost::beast::http::write_header(*stream, m_Serializer);
+			}
+
+			if (finish) {
+				Base::body().Finish();
+			}
+
+			boost::system::error_code ec;
+			boost::beast::http::async_write(*stream, m_Serializer, yc[ec]);
+			if (ec && ec != boost::beast::http::error::need_buffer) {
+				if (yc.ec_) {
+					*yc.ec_ = ec;
+					return;
+				}
+				BOOST_THROW_EXCEPTION(boost::system::system_error{ec});
+			}
+			stream->async_flush(yc);
+
+			ASSERT(m_Serializer.is_done() || !Base::body().Finished());
+		},
+		m_Stream
+	);
+}
 
-	boost::system::error_code ec;
-	boost::beast::http::async_write(*m_Stream, m_Serializer, yc[ec]);
-	if (ec && ec != boost::beast::http::error::need_buffer) {
-		if (yc.ec_) {
-			*yc.ec_ = ec;
-			return;
-		}
-		BOOST_THROW_EXCEPTION(boost::system::system_error{ec});
-	}
-	m_Stream->async_flush(yc);
+template<bool isRequest, typename Body, typename StreamVariant>
+void OutgoingHttpMessage<isRequest, Body, StreamVariant>::StartStreaming()
+{
+	ASSERT(Base::body().Size() == 0 && !m_SerializationStarted);
+	Base::body().Start();
+	Base::chunked(true);
+}
 
-	ASSERT(m_Serializer.is_done() || !body().Finished());
+HttpApiResponse::HttpApiResponse(Shared<AsioTlsStream>::Ptr stream, HttpServerConnection::Ptr server)
+	: OutgoingHttpMessage(std::move(stream)), m_Server(std::move(server))
+{
 }
 
 void HttpApiResponse::StartStreaming(bool checkForDisconnect)
 {
-	ASSERT(body().Size() == 0 && !m_SerializationStarted);
-	body().Start();
-	chunked(true);
+	OutgoingHttpMessage::StartStreaming();
 
 	if (checkForDisconnect) {
 		ASSERT(m_Server);
@@ -167,30 +208,56 @@ bool HttpApiResponse::IsClientDisconnected() const
 	return m_Server->Disconnected();
 }
 
-void HttpApiResponse::SendFile(const String& path, const boost::asio::yield_context& yc)
+template<bool isRequest, typename Body, typename StreamVariant>
+void OutgoingHttpMessage<isRequest, Body, StreamVariant>::SendFile(
+	const String& path,
+	const boost::asio::yield_context& yc
+)
 {
 	std::ifstream fp(path.CStr(), std::ifstream::in | std::ifstream::binary | std::ifstream::ate);
 	fp.exceptions(std::ifstream::badbit | std::ifstream::eofbit);
 
 	std::uint64_t remaining = fp.tellg();
 	fp.seekg(0);
 
-	content_length(remaining);
-	body().Start();
+	Base::content_length(remaining);
+	Base::body().Start();
 
 	while (remaining) {
 		auto maxTransfer = std::min(remaining, static_cast<std::uint64_t>(l_FlushThreshold));
 
-		auto buf = *body().Buffer().prepare(maxTransfer).begin();
+		using BodyBuffer = std::decay_t<decltype(std::declval<typename Body::value_type>().Buffer())>;
+		using BufferOrSequence = typename BodyBuffer::mutable_buffers_type;
+		
+		boost::asio::mutable_buffer buf;
+
+		if constexpr (!std::is_same_v<BufferOrSequence, boost::asio::mutable_buffer>) {
+			buf = *Base::body().Buffer().prepare(maxTransfer).begin();
+		} else {
+			buf = Base::body().Buffer().prepare(maxTransfer);
+		}
 		fp.read(static_cast<char*>(buf.data()), buf.size());
-		body().Buffer().commit(buf.size());
+		Base::body().Buffer().commit(buf.size());
 
 		remaining -= buf.size();
 		Flush(yc);
 	}
 }
 
-JsonEncoder HttpApiResponse::GetJsonEncoder(bool pretty)
+template<bool isRequest, typename Body, typename StreamVariant>
+JsonEncoder OutgoingHttpMessage<isRequest, Body, StreamVariant>::GetJsonEncoder(bool pretty)
 {
-	return JsonEncoder{std::make_shared<HttpResponseJsonWriter>(*this), pretty};
+	return JsonEncoder{
+		std::make_shared<HttpResponseJsonWriter<OutgoingHttpMessage<isRequest, Body, StreamVariant>>>(*this), pretty
+	};
 }
+
+// More general instantiations
+template class icinga::OutgoingHttpMessage<true, SerializableFlatBufferBody, AsioTlsOrTcpStream>;
+template class icinga::OutgoingHttpMessage<false, SerializableFlatBufferBody, AsioTlsOrTcpStream>;
+template class icinga::IncomingHttpMessage<true, boost::beast::http::string_body, AsioTlsOrTcpStream>;
+template class icinga::IncomingHttpMessage<false, boost::beast::http::string_body, AsioTlsOrTcpStream>;
+
+// Instantiations specifically for HttpApi(Request|Response)
+template class icinga::IncomingHttpMessage<true, boost::beast::http::string_body, std::variant<Shared<AsioTlsStream>::Ptr>>;
+template class icinga::OutgoingHttpMessage<false, SerializableMultiBufferBody, std::variant<Shared<AsioTlsStream>::Ptr>>;