Commit b5eea2ee authored by Dominik Charousset's avatar Dominik Charousset

Add new routing API for HTTP

parent 9af5eddd
...@@ -60,29 +60,25 @@ int caf_main(caf::actor_system& sys, const config& cfg) { ...@@ -60,29 +60,25 @@ int caf_main(caf::actor_system& sys, const config& cfg) {
.accept(port) .accept(port)
// Limit how many clients may be connected at any given time. // Limit how many clients may be connected at any given time.
.max_connections(max_connections) .max_connections(max_connections)
// When started, run our worker actor to handle incoming requests. // Provide the time at '/'.
.start([&sys](auto requests) { .route("/", http::method::get,
// Note: requests is an async::consumer_resource<http::request>. [](http::responder& res) {
sys.spawn([requests](caf::event_based_actor* self) { auto str = caf::deep_to_string(caf::make_timestamp());
// For each incoming HTTP request ... res.respond(http::status::ok, "text/plain", str);
requests })
.observe_on(self) // // Launch the server.
.for_each([](const http::request& req) { .start();
// ... respond with the current time as string.
auto str = caf::deep_to_string(caf::make_timestamp());
req.respond(http::status::ok, "text/plain", str);
// Note: we cannot respond more than once to a request.
});
});
});
// Report any error to the user. // Report any error to the user.
if (!server) { if (!server) {
std::cerr << "*** unable to run at port " << port << ": " std::cerr << "*** unable to run at port " << port << ": "
<< to_string(server.error()) << '\n'; << to_string(server.error()) << '\n';
return EXIT_FAILURE; return EXIT_FAILURE;
} }
// Note: the actor system will keep the application running for as long as the // Note: the actor system will only wait for actors on default. Since we don't
// workers are still alive. // start actors, we need to block on something else.
std::cout << "Server is up and running. Press <enter> to shut down.\n";
getchar();
std::cout << "Terminating.\n";
return EXIT_SUCCESS; return EXIT_SUCCESS;
} }
......
...@@ -9,112 +9,114 @@ configure_file(test/pem.cpp.in test/pem.cpp @ONLY) ...@@ -9,112 +9,114 @@ configure_file(test/pem.cpp.in test/pem.cpp @ONLY)
caf_add_component( caf_add_component(
net net
DEPENDENCIES DEPENDENCIES
PUBLIC PUBLIC
$<$<CXX_COMPILER_ID:MSVC>:ws2_32> $<$<CXX_COMPILER_ID:MSVC>:ws2_32>
CAF::core CAF::core
OpenSSL::Crypto OpenSSL::Crypto
OpenSSL::SSL OpenSSL::SSL
PRIVATE PRIVATE
CAF::internal CAF::internal
ENUM_TYPES ENUM_TYPES
net.http.method net.http.method
net.http.status net.http.status
net.octet_stream.errc net.octet_stream.errc
net.ssl.dtls net.ssl.dtls
net.ssl.errc net.ssl.errc
net.ssl.format net.ssl.format
net.ssl.tls net.ssl.tls
net.web_socket.status net.web_socket.status
HEADERS HEADERS
${CAF_NET_HEADERS} ${CAF_NET_HEADERS}
SOURCES SOURCES
src/detail/convert_ip_endpoint.cpp src/detail/convert_ip_endpoint.cpp
src/detail/pollset_updater.cpp src/detail/pollset_updater.cpp
src/detail/rfc6455.cpp src/detail/rfc6455.cpp
src/net/abstract_actor_shell.cpp src/net/abstract_actor_shell.cpp
src/net/actor_shell.cpp src/net/actor_shell.cpp
src/net/datagram_socket.cpp src/net/datagram_socket.cpp
src/net/dsl/config_base.cpp src/net/dsl/config_base.cpp
src/net/generic_lower_layer.cpp src/net/generic_lower_layer.cpp
src/net/generic_upper_layer.cpp src/net/generic_upper_layer.cpp
src/net/http/lower_layer.cpp src/net/http/lower_layer.cpp
src/net/http/method.cpp src/net/http/method.cpp
src/net/http/request.cpp src/net/http/request.cpp
src/net/http/request_header.cpp src/net/http/request_header.cpp
src/net/http/response.cpp src/net/http/responder.cpp
src/net/http/server.cpp src/net/http/response.cpp
src/net/http/server_factory.cpp src/net/http/router.cpp
src/net/http/status.cpp src/net/http/server.cpp
src/net/http/upper_layer.cpp src/net/http/server_factory.cpp
src/net/http/v1.cpp src/net/http/status.cpp
src/net/ip.cpp src/net/http/upper_layer.cpp
src/net/lp/default_trait.cpp src/net/http/v1.cpp
src/net/lp/frame.cpp src/net/ip.cpp
src/net/lp/framing.cpp src/net/lp/default_trait.cpp
src/net/lp/lower_layer.cpp src/net/lp/frame.cpp
src/net/lp/upper_layer.cpp src/net/lp/framing.cpp
src/net/middleman.cpp src/net/lp/lower_layer.cpp
src/net/multiplexer.cpp src/net/lp/upper_layer.cpp
src/net/network_socket.cpp src/net/middleman.cpp
src/net/octet_stream/lower_layer.cpp src/net/multiplexer.cpp
src/net/octet_stream/policy.cpp src/net/network_socket.cpp
src/net/octet_stream/transport.cpp src/net/octet_stream/lower_layer.cpp
src/net/octet_stream/upper_layer.cpp src/net/octet_stream/policy.cpp
src/net/pipe_socket.cpp src/net/octet_stream/transport.cpp
src/net/prometheus/server.cpp src/net/octet_stream/upper_layer.cpp
src/net/socket.cpp src/net/pipe_socket.cpp
src/net/socket_event_layer.cpp src/net/prometheus.cpp
src/net/socket_manager.cpp src/net/socket.cpp
src/net/ssl/connection.cpp src/net/socket_event_layer.cpp
src/net/ssl/context.cpp src/net/socket_manager.cpp
src/net/ssl/dtls.cpp src/net/ssl/connection.cpp
src/net/ssl/errc.cpp src/net/ssl/context.cpp
src/net/ssl/format.cpp src/net/ssl/dtls.cpp
src/net/ssl/password.cpp src/net/ssl/errc.cpp
src/net/ssl/startup.cpp src/net/ssl/format.cpp
src/net/ssl/tcp_acceptor.cpp src/net/ssl/password.cpp
src/net/ssl/tls.cpp src/net/ssl/startup.cpp
src/net/ssl/transport.cpp src/net/ssl/tcp_acceptor.cpp
src/net/ssl/verify.cpp src/net/ssl/tls.cpp
src/net/stream_socket.cpp src/net/ssl/transport.cpp
src/net/tcp_accept_socket.cpp src/net/ssl/verify.cpp
src/net/tcp_stream_socket.cpp src/net/stream_socket.cpp
src/net/this_host.cpp src/net/tcp_accept_socket.cpp
src/net/udp_datagram_socket.cpp src/net/tcp_stream_socket.cpp
src/net/web_socket/client.cpp src/net/this_host.cpp
src/net/web_socket/default_trait.cpp src/net/udp_datagram_socket.cpp
src/net/web_socket/frame.cpp src/net/web_socket/client.cpp
src/net/web_socket/framing.cpp src/net/web_socket/default_trait.cpp
src/net/web_socket/handshake.cpp src/net/web_socket/frame.cpp
src/net/web_socket/lower_layer.cpp src/net/web_socket/framing.cpp
src/net/web_socket/server.cpp src/net/web_socket/handshake.cpp
src/net/web_socket/upper_layer.cpp src/net/web_socket/lower_layer.cpp
src/net/web_socket/server.cpp
src/net/web_socket/upper_layer.cpp
TEST_SOURCES TEST_SOURCES
${CMAKE_CURRENT_BINARY_DIR}/test/pem.cpp ${CMAKE_CURRENT_BINARY_DIR}/test/pem.cpp
test/net-test.cpp test/net-test.cpp
TEST_SUITES TEST_SUITES
detail.convert_ip_endpoint detail.convert_ip_endpoint
detail.rfc6455 detail.rfc6455
net.accept_socket net.accept_socket
net.actor_shell net.actor_shell
net.datagram_socket net.datagram_socket
net.http.server net.http.router
net.ip net.http.server
net.length_prefix_framing net.ip
net.lp.frame net.length_prefix_framing
net.multiplexer net.lp.frame
net.network_socket net.multiplexer
net.octet_stream.transport net.network_socket
net.pipe_socket net.octet_stream.transport
net.prometheus.server net.pipe_socket
net.socket net.socket
net.socket_guard net.socket_guard
net.ssl.transport net.ssl.transport
net.stream_socket net.stream_socket
net.tcp_socket net.tcp_socket
net.typed_actor_shell net.typed_actor_shell
net.udp_datagram_socket net.udp_datagram_socket
net.web_socket.client net.web_socket.client
net.web_socket.frame net.web_socket.frame
net.web_socket.handshake net.web_socket.handshake
net.web_socket.server) net.web_socket.server)
...@@ -156,7 +156,7 @@ public: ...@@ -156,7 +156,7 @@ public:
using super::super; using super::super;
template <class From, class T, class... Args> template <class T, class From, class... Args>
static auto make(client_config_tag<T>, From&& from, Args&&... args) { static auto make(client_config_tag<T>, From&& from, Args&&... args) {
static_assert(std::is_constructible_v<T, Args...>); static_assert(std::is_constructible_v<T, Args...>);
return make_counted<value>(std::forward<From>(from), return make_counted<value>(std::forward<From>(from),
......
...@@ -87,9 +87,10 @@ public: ...@@ -87,9 +87,10 @@ public:
using super::super; using super::super;
template <class T, class... Args> template <class T, class From, class... Args>
static auto make(server_config_tag<T>, const Base& other, Args&&... args) { static auto make(server_config_tag<T>, const From& from, Args&&... args) {
return make_counted<value>(other, std::in_place_type<T>, static_assert(std::is_constructible_v<T, Args...>);
return make_counted<value>(from, std::in_place_type<T>,
std::forward<Args>(args)...); std::forward<Args>(args)...);
} }
......
...@@ -128,14 +128,12 @@ namespace caf::net::http { ...@@ -128,14 +128,12 @@ namespace caf::net::http {
class lower_layer; class lower_layer;
class request; class request;
class request_header; class request_header;
class responder;
class router;
class server; class server;
class upper_layer; class upper_layer;
using header_fields_map
= unordered_flat_map<std::string_view, std::string_view>;
enum class method : uint8_t; enum class method : uint8_t;
enum class status : uint16_t; enum class status : uint16_t;
} // namespace caf::net::http } // namespace caf::net::http
......
...@@ -6,6 +6,7 @@ ...@@ -6,6 +6,7 @@
#include "caf/detail/net_export.hpp" #include "caf/detail/net_export.hpp"
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/net/fwd.hpp"
namespace caf::net { namespace caf::net {
...@@ -15,6 +16,9 @@ class CAF_NET_EXPORT generic_lower_layer { ...@@ -15,6 +16,9 @@ class CAF_NET_EXPORT generic_lower_layer {
public: public:
virtual ~generic_lower_layer(); virtual ~generic_lower_layer();
/// Returns the @ref multiplexer instance that executes this protocol stack.
virtual multiplexer& mpx() noexcept = 0;
/// Queries whether the output device can accept more data straight away. /// Queries whether the output device can accept more data straight away.
[[nodiscard]] virtual bool can_send_more() const noexcept = 0; [[nodiscard]] virtual bool can_send_more() const noexcept = 0;
......
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/detail/parse.hpp"
#include <optional>
namespace caf::net::http {
/// Customization point for adding custom types to the `<arg>` parsing of the
/// @ref router.
template <class>
struct arg_parser;
template <>
struct arg_parser<std::string> {
std::optional<std::string> parse(std::string_view str) {
return std::string{str};
}
};
template <class T>
struct builtin_arg_parser {
std::optional<T> parse(std::string_view str) {
auto tmp = T{};
if (auto err = detail::parse(str, tmp); !err)
return tmp;
else
return {};
}
};
template <class T, bool IsArithmetic = std::is_arithmetic_v<T>>
struct arg_parser_oracle {
using type = arg_parser<T>;
};
template <class T>
struct arg_parser_oracle<T, true> {
using type = builtin_arg_parser<T>;
};
template <class T>
using arg_parser_t = typename arg_parser_oracle<T>::type;
} // namespace caf::net::http
...@@ -20,7 +20,8 @@ ...@@ -20,7 +20,8 @@
namespace caf::net::http { namespace caf::net::http {
/// Implicitly shared Handle type that represents an HTTP client request. /// Implicitly shared handle type that represents an HTTP client request with a
/// promise for the HTTP response.
class CAF_NET_EXPORT request { class CAF_NET_EXPORT request {
public: public:
struct impl { struct impl {
......
...@@ -6,7 +6,7 @@ ...@@ -6,7 +6,7 @@
#include "caf/config_value.hpp" #include "caf/config_value.hpp"
#include "caf/detail/net_export.hpp" #include "caf/detail/net_export.hpp"
#include "caf/net/http/header_fields_map.hpp" #include "caf/net/fwd.hpp"
#include "caf/net/http/method.hpp" #include "caf/net/http/method.hpp"
#include "caf/net/http/status.hpp" #include "caf/net/http/status.hpp"
#include "caf/uri.hpp" #include "caf/uri.hpp"
...@@ -16,45 +16,74 @@ ...@@ -16,45 +16,74 @@
namespace caf::net::http { namespace caf::net::http {
/// Encapsulates meta data for HTTP requests. /// Encapsulates meta data for HTTP requests. This class represents an HTTP
/// request header, providing methods for accessing the HTTP method, path,
/// query, fragment, version, and fields.
class CAF_NET_EXPORT request_header { class CAF_NET_EXPORT request_header {
public: public:
friend class http::server;
friend class web_socket::server;
/// Default constructor.
request_header() = default; request_header() = default;
/// Move constructor.
request_header(request_header&&) = default; request_header(request_header&&) = default;
/// Move assignment operator.
request_header& operator=(request_header&&) = default; request_header& operator=(request_header&&) = default;
/// Copy constructor.
request_header(const request_header&); request_header(const request_header&);
/// Copy assignment operator.
request_header& operator=(const request_header&); request_header& operator=(const request_header&);
/// Assigns the content of another request_header.
void assign(const request_header&); void assign(const request_header&);
/// Returns the HTTP method of the request.
http::method method() const noexcept { http::method method() const noexcept {
return method_; return method_;
} }
/// Returns the path part of the request URI.
std::string_view path() const noexcept { std::string_view path() const noexcept {
return uri_.path(); return uri_.path();
} }
/// Returns the query part of the request URI as a map.
const uri::query_map& query() const noexcept { const uri::query_map& query() const noexcept {
return uri_.query(); return uri_.query();
} }
/// Returns the fragment part of the request URI.
std::string_view fragment() const noexcept { std::string_view fragment() const noexcept {
return uri_.fragment(); return uri_.fragment();
} }
/// Returns the HTTP version of the request.
std::string_view version() const noexcept { std::string_view version() const noexcept {
return version_; return version_;
} }
const header_fields_map& fields() const noexcept { /// Returns the number of fields in the request header.
return fields_; size_t num_fields() const noexcept {
return fields_.size();
}
/// Returns the field at the specified index as a key-value pair.
std::pair<std::string_view, std::string_view> field_at(size_t index) {
return fields_.container().at(index);
}
/// Checks if the request header has a field with the specified key.
bool has_field(std::string_view key) const noexcept {
return fields_.find(key) != fields_.end();
} }
/// Returns the value of the field with the specified key, or an empty view if
/// the field is not found.
std::string_view field(std::string_view key) const noexcept { std::string_view field(std::string_view key) const noexcept {
if (auto i = fields_.find(key); i != fields_.end()) if (auto i = fields_.find(key); i != fields_.end())
return i->second; return i->second;
...@@ -62,35 +91,59 @@ public: ...@@ -62,35 +91,59 @@ public:
return {}; return {};
} }
/// Returns the value of the field with the specified key as the requested
/// type T, or std::nullopt if the field is not found or cannot be converted.
template <class T> template <class T>
std::optional<T> field_as(std::string_view key) const noexcept { std::optional<T> field_as(std::string_view key) const noexcept {
if (auto i = fields_.find(key); i != fields_.end()) { if (auto i = fields_.find(key); i != fields_.end()) {
caf::config_value val{std::string{i->second}}; caf::config_value val{std::string{i->second}};
if (auto res = caf::get_as<T>(val)) if (auto res = caf::get_as<T>(val))
return std::move(*res); return std::move(*res);
else
return {};
} else {
return {};
} }
return {};
}
/// Executes the provided callable `f` for each field in the request header.
/// @param f A function object taking two `std::string_view` parameters: key
/// and value.
template <class F>
void for_each_field(F&& f) const {
for (auto& [key, val] : fields_)
f(key, val);
} }
/// Checks if the request header is valid (non-empty).
bool valid() const noexcept { bool valid() const noexcept {
return !raw_.empty(); return !raw_.empty();
} }
std::pair<status, std::string_view> parse(std::string_view raw); /// Checks whether the client has defined `Transfer-Encoding` as `chunked`.
bool chunked_transfer_encoding() const; bool chunked_transfer_encoding() const;
/// Convenience function for `field_as<size_t>("Content-Length")`.
std::optional<size_t> content_length() const; std::optional<size_t> content_length() const;
/// Parses a raw request header string and returns a pair containing the
/// status and a description for the status.
/// @returns `status::bad_request` on error with a human-readable description
/// of the error, `status::ok` otherwise.
std::pair<status, std::string_view> parse(std::string_view raw);
private: private:
/// Stores the raw HTTP input.
std::vector<char> raw_; std::vector<char> raw_;
/// Stores the HTTP method that we've parsed from the raw input.
http::method method_; http::method method_;
/// Stores the HTTP request URI that we've parsed from the raw input.
uri uri_; uri uri_;
/// Stores the Version of the parsed HTTP input.
std::string_view version_; std::string_view version_;
header_fields_map fields_;
/// A shallow map for looking up individual header fields.
unordered_flat_map<std::string_view, std::string_view> fields_;
}; };
} // namespace caf::net::http } // namespace caf::net::http
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/byte_span.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/http/lower_layer.hpp"
#include "caf/net/http/request_header.hpp"
#include <string_view>
namespace caf::net::http {
/// Responds to an HTTP request at the server. Provides functions for accessing
/// the HTTP client request and for writing the HTTP response.
///
/// This class has a similar API to @ref request, but is used at the @ref server
/// directly. While a @ref request is meant to handled outside of the server by
/// eventually fulfilling the response promise, a `responder` must generate the
/// response immediately.
class CAF_NET_EXPORT responder {
public:
responder(const request_header* hdr, const_byte_span body,
http::router* router)
: hdr_(hdr), body_(body), router_(router) {
// nop
}
responder(const responder&) noexcept = default;
responder& operator=(const responder&) noexcept = default;
/// Returns the HTTP header for the responder.
/// @pre `valid()`
const request_header& header() const noexcept {
return *hdr_;
}
/// Returns the HTTP body (payload) for the responder.
/// @pre `valid()`
const_byte_span body() const noexcept {
return body_;
}
/// @copydoc body
const_byte_span payload() const noexcept {
return body_;
}
/// Returns the router that has created this responder.
http::router* router() const noexcept {
return router_;
}
/// Sends an HTTP response message to the client. Automatically sets the
/// `Content-Type` and `Content-Length` header fields.
/// @pre `valid()`
void respond(status code, std::string_view content_type,
const_byte_span content) {
down()->send_response(code, content_type, content);
}
/// Sends an HTTP response message to the client. Automatically sets the
/// `Content-Type` and `Content-Length` header fields.
/// @pre `valid()`
void respond(status code, std::string_view content_type,
std::string_view content) {
down()->send_response(code, content_type, content);
}
/// Starts writing an HTTP header.
void begin_header(status code) {
down()->begin_header(code);
}
/// Adds a header field. Users may only call this function between
/// `begin_header` and `end_header`.
void add_header_field(std::string_view key, std::string_view val) {
down()->add_header_field(key, val);
}
/// Seals the header and transports it to the client.
bool end_header() {
return down()->end_header();
}
/// Sends the payload after the header.
bool send_payload(const_byte_span bytes) {
return down()->send_payload(bytes);
}
/// Sends a chunk of data if the full payload is unknown when starting to
/// send.
bool send_chunk(const_byte_span bytes) {
return down()->send_chunk(bytes);
}
/// Sends the last chunk, completing a chunked payload.
bool send_end_of_chunks() {
return down()->send_end_of_chunks();
}
/// Converts a responder to a @ref request for processing the HTTP request
/// asynchronously.
request to_request() &&;
private:
lower_layer* down();
const request_header* hdr_;
const_byte_span body_;
http::router* router_;
};
} // namespace caf::net::http
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/detail/print.hpp"
#include "caf/detail/type_list.hpp"
#include "caf/detail/type_traits.hpp"
#include "caf/expected.hpp"
#include "caf/intrusive_ptr.hpp"
#include "caf/net/http/arg_parser.hpp"
#include "caf/net/http/lower_layer.hpp"
#include "caf/net/http/responder.hpp"
#include "caf/net/http/upper_layer.hpp"
#include "caf/ref_counted.hpp"
#include <algorithm>
#include <cassert>
#include <string_view>
#include <unordered_map>
#include <utility>
namespace caf::net::http {
/// Sits on top of a @ref server and dispatches incoming requests to
/// user-defined handlers.
class CAF_NET_EXPORT router : public upper_layer {
public:
// -- member types -----------------------------------------------------------
class route : public ref_counted {
public:
virtual ~route();
/// Returns `true` if the route accepted the request, `false` otherwise.
virtual bool
exec(const request_header& hdr, const_byte_span body, router* parent)
= 0;
/// Counts how many `<arg>` entries are in `path`.
static size_t args_in_path(std::string_view path);
/// Splits `str` in the first component of a path and its remainder.
static std::pair<std::string_view, std::string_view>
next_component(std::string_view str);
/// Matches a two paths by splitting both inputs at '/' and then checking
/// that `predicate` holds for each resulting pair.
template <class F>
bool match_path(std::string_view lhs, std::string_view rhs, F&& predicate) {
std::string_view head1;
std::string_view tail1;
std::string_view head2;
std::string_view tail2;
std::tie(head1, tail1) = next_component(lhs);
std::tie(head2, tail2) = next_component(rhs);
if (!predicate(head1, head2))
return false;
while (!tail1.empty()) {
if (tail2.empty())
return false;
std::tie(head1, tail1) = next_component(tail1);
std::tie(head2, tail2) = next_component(tail2);
if (!predicate(head1, head2))
return false;
}
return tail2.empty();
}
};
using route_ptr = intrusive_ptr<route>;
// -- constructors and destructors -------------------------------------------
router() = default;
explicit router(std::vector<route_ptr> routes) : routes_(std::move(routes)) {
// nop
}
~router() override;
// -- factories --------------------------------------------------------------
static std::unique_ptr<router> make(std::vector<route_ptr> routes);
/// Tries to create a new HTTP route.
/// @param path The path on this server for the new route.
/// @param f The function object for handling requests on the new route.
/// @return the @ref route object on success, an @ref error otherwise.
template <class F>
static expected<route_ptr> make_route(std::string path, F f) {
return make_route_dis(path, std::nullopt, f);
}
/// Tries to create a new HTTP route.
/// @param path The path on this server for the new route.
/// @param method The allowed HTTP method on the new route.
/// @param f The function object for handling requests on the new route.
/// @return the @ref route object on success, an @ref error otherwise.
template <class F>
static expected<route_ptr>
make_route(std::string path, http::method method, F f) {
return make_route_dis(path, method, f);
}
/// Create a new HTTP default "catch all" route.
/// @param f The function object for handling the requests.
/// @return the @ref route object.
template <class F>
static route_ptr make_route(F f) {
return make_counted<default_route_impl<F>>(std::move(f));
}
// -- properties -------------------------------------------------------------
lower_layer* down() {
return down_;
}
// -- API for the responders -------------------------------------------------
/// Lifts a @ref responder to an @ref request object that allows asynchronous
/// processing of the HTTP request.
request lift(responder&& res);
void shutdown(const error& err);
// -- http::upper_layer implementation ---------------------------------------
error start(lower_layer* down) override;
ptrdiff_t consume(const request_header& hdr,
const_byte_span payload) override;
void prepare_send() override;
bool done_sending() override;
void abort(const error& reason) override;
private:
template <class F, class... Ts>
class route_impl : public route {
public:
explicit route_impl(std::string&& path, std::optional<http::method> method,
F&& f)
: path_(std::move(path)), method_(method), f_(std::move(f)) {
// nop
}
bool exec(const request_header& hdr, const_byte_span body,
router* parent) override {
if (method_ && *method_ != hdr.method())
return false;
// Try to match the path to the expected path and extract args.
std::string_view args[sizeof...(Ts)];
auto ok = match_path(path_, hdr.path(),
[pos = args](std::string_view lhs,
std::string_view rhs) mutable {
if (lhs == "<arg>") {
*pos++ = rhs;
return true;
} else {
return lhs == rhs;
}
});
if (!ok)
return false;
// Try to parse the arguments.
using iseq = std::make_index_sequence<sizeof...(Ts)>;
return exec_dis(hdr, body, parent, iseq{}, args);
}
template <size_t... Is>
bool exec_dis(const request_header& hdr, const_byte_span body,
router* parent, std::index_sequence<Is...>,
std::string_view* arr) {
return exec_impl(hdr, body, parent,
std::get<Is>(parsers_).parse(arr[Is])...);
}
template <class... Is>
bool exec_impl(const request_header& hdr, const_byte_span body,
router* parent, std::optional<Ts>&&... args) {
if ((args.has_value() && ...)) {
responder rp{&hdr, body, parent};
f_(rp, std::move(*args)...);
return true;
}
return false;
}
private:
std::string path_;
std::optional<http::method> method_;
F f_;
std::tuple<arg_parser_t<Ts>...> parsers_;
};
template <class F>
class trivial_route_impl : public route {
public:
explicit trivial_route_impl(std::string&& path,
std::optional<http::method> method, F&& f)
: path_(std::move(path)), method_(method), f_(std::move(f)) {
// nop
}
bool exec(const request_header& hdr, const_byte_span body,
router* parent) override {
if (method_ && *method_ != hdr.method())
return false;
if (hdr.path() == path_) {
responder rp{&hdr, body, parent};
f_(rp);
return true;
}
return false;
}
private:
std::string path_;
std::optional<http::method> method_;
F f_;
};
template <class F>
class default_route_impl : public route {
public:
explicit default_route_impl(F&& f) : f_(std::move(f)) {
// nop
}
bool exec(const request_header& hdr, const_byte_span body,
router* parent) override {
responder rp{&hdr, body, parent};
f_(rp);
return true;
}
private:
F f_;
};
// Dispatches to make_route_impl after sanity checking.
template <class F>
static expected<route_ptr>
make_route_dis(std::string& path, std::optional<http::method> method, F& f) {
// F must have signature void (responder&, ...).
using f_trait = detail::get_callable_trait_t<F>;
using f_args = typename f_trait::arg_types;
static_assert(f_trait::num_args > 0, "F must take at least one argument");
using arg_0 = detail::tl_at_t<f_args, 0>;
static_assert(std::is_same_v<arg_0, responder&>,
"F must take 'responder&' as first argument");
// The path must be absolute.
if (path.empty() || path.front() != '/') {
return make_error(sec::invalid_argument,
"expected an absolute path, got: " + path);
}
// The path must has as many <arg> entries as F takes extra arguments.
auto num_args = route::args_in_path(path);
if (num_args != f_trait::num_args - 1) {
auto msg = path;
msg += " defines ";
detail::print(msg, num_args);
msg += " arguments, but F accepts ";
detail::print(msg, f_trait::num_args - 1);
return make_error(sec::invalid_argument, std::move(msg));
}
// Dispatch to the actual factory.
return make_route_impl(path, method, f, f_args{});
}
template <class F, class... Args>
static expected<route_ptr>
make_route_impl(std::string& path, std::optional<http::method> method, F& f,
detail::type_list<responder&, Args...>) {
if constexpr (sizeof...(Args) == 0) {
return make_counted<trivial_route_impl<F>>(std::move(path), method,
std::move(f));
} else {
return make_counted<route_impl<F, Args...>>(std::move(path), method,
std::move(f));
}
}
lower_layer* down_ = nullptr;
std::vector<route_ptr> routes_;
size_t request_id_ = 0;
std::unordered_map<size_t, disposable> pending_;
};
} // namespace caf::net::http
...@@ -11,7 +11,6 @@ ...@@ -11,7 +11,6 @@
#include "caf/error.hpp" #include "caf/error.hpp"
#include "caf/logger.hpp" #include "caf/logger.hpp"
#include "caf/net/fwd.hpp" #include "caf/net/fwd.hpp"
#include "caf/net/http/header_fields_map.hpp"
#include "caf/net/http/lower_layer.hpp" #include "caf/net/http/lower_layer.hpp"
#include "caf/net/http/request_header.hpp" #include "caf/net/http/request_header.hpp"
#include "caf/net/http/status.hpp" #include "caf/net/http/status.hpp"
...@@ -78,6 +77,8 @@ public: ...@@ -78,6 +77,8 @@ public:
// -- http::lower_layer implementation --------------------------------------- // -- http::lower_layer implementation ---------------------------------------
multiplexer& mpx() noexcept override;
bool can_send_more() const noexcept override; bool can_send_more() const noexcept override;
bool is_reading() const noexcept override; bool is_reading() const noexcept override;
......
...@@ -5,6 +5,7 @@ ...@@ -5,6 +5,7 @@
#pragma once #pragma once
#include "caf/async/blocking_producer.hpp" #include "caf/async/blocking_producer.hpp"
#include "caf/async/execution_context.hpp"
#include "caf/defaults.hpp" #include "caf/defaults.hpp"
#include "caf/detail/accept_handler.hpp" #include "caf/detail/accept_handler.hpp"
#include "caf/detail/connection_factory.hpp" #include "caf/detail/connection_factory.hpp"
...@@ -13,12 +14,13 @@ ...@@ -13,12 +14,13 @@
#include "caf/net/checked_socket.hpp" #include "caf/net/checked_socket.hpp"
#include "caf/net/dsl/server_factory_base.hpp" #include "caf/net/dsl/server_factory_base.hpp"
#include "caf/net/http/request.hpp" #include "caf/net/http/request.hpp"
#include "caf/net/http/router.hpp"
#include "caf/net/http/server.hpp" #include "caf/net/http/server.hpp"
#include "caf/net/http/upper_layer.hpp"
#include "caf/net/multiplexer.hpp" #include "caf/net/multiplexer.hpp"
#include "caf/net/octet_stream/transport.hpp" #include "caf/net/octet_stream/transport.hpp"
#include "caf/net/ssl/transport.hpp" #include "caf/net/ssl/transport.hpp"
#include "caf/net/tcp_accept_socket.hpp" #include "caf/net/tcp_accept_socket.hpp"
#include "caf/net/tcp_stream_socket.hpp"
#include "caf/none.hpp" #include "caf/none.hpp"
#include <cstdint> #include <cstdint>
...@@ -31,12 +33,13 @@ class CAF_NET_EXPORT http_request_producer : public atomic_ref_counted, ...@@ -31,12 +33,13 @@ class CAF_NET_EXPORT http_request_producer : public atomic_ref_counted,
public: public:
using buffer_ptr = async::spsc_buffer_ptr<net::http::request>; using buffer_ptr = async::spsc_buffer_ptr<net::http::request>;
http_request_producer(buffer_ptr buf) : buf_(std::move(buf)) { http_request_producer(async::execution_context_ptr ecp, buffer_ptr buf)
: ecp_(std::move(ecp)), buf_(std::move(buf)) {
// nop // nop
} }
static auto make(buffer_ptr buf) { static auto make(async::execution_context_ptr ecp, buffer_ptr buf) {
auto ptr = make_counted<http_request_producer>(buf); auto ptr = make_counted<http_request_producer>(std::move(ecp), buf);
buf->set_producer(ptr); buf->set_producer(ptr);
return ptr; return ptr;
} }
...@@ -54,58 +57,28 @@ public: ...@@ -54,58 +57,28 @@ public:
bool push(const net::http::request& item); bool push(const net::http::request& item);
private: private:
async::execution_context_ptr ecp_;
buffer_ptr buf_; buffer_ptr buf_;
}; };
using http_request_producer_ptr = intrusive_ptr<http_request_producer>; using http_request_producer_ptr = intrusive_ptr<http_request_producer>;
class CAF_NET_EXPORT http_flow_adapter : public net::http::upper_layer {
public:
explicit http_flow_adapter(async::execution_context_ptr loop,
http_request_producer_ptr ptr)
: loop_(std::move(loop)), producer_(std::move(ptr)) {
// nop
}
void prepare_send() override;
bool done_sending() override;
void abort(const error&) override;
error start(net::http::lower_layer* down) override;
ptrdiff_t consume(const net::http::request_header& hdr,
const_byte_span payload) override;
static auto make(async::execution_context_ptr loop,
http_request_producer_ptr ptr) {
return std::make_unique<http_flow_adapter>(loop, ptr);
}
private:
async::execution_context_ptr loop_;
net::http::lower_layer* down_ = nullptr;
std::vector<disposable> pending_;
http_request_producer_ptr producer_;
};
template <class Transport> template <class Transport>
class http_conn_factory class http_conn_factory
: public connection_factory<typename Transport::connection_handle> { : public connection_factory<typename Transport::connection_handle> {
public: public:
using connection_handle = typename Transport::connection_handle; using connection_handle = typename Transport::connection_handle;
http_conn_factory(http_request_producer_ptr producer, http_conn_factory(std::vector<net::http::router::route_ptr> routes,
size_t max_consecutive_reads) size_t max_consecutive_reads)
: producer_(std::move(producer)), : routes_(std::move(routes)),
max_consecutive_reads_(max_consecutive_reads) { max_consecutive_reads_(max_consecutive_reads) {
// nop // nop
} }
net::socket_manager_ptr make(net::multiplexer* mpx, net::socket_manager_ptr make(net::multiplexer* mpx,
connection_handle conn) override { connection_handle conn) override {
auto app = http_flow_adapter::make(mpx, producer_); auto app = net::http::router::make(routes_);
auto serv = net::http::server::make(std::move(app)); auto serv = net::http::server::make(std::move(app));
auto fd = conn.fd(); auto fd = conn.fd();
auto transport = Transport::make(std::move(conn), std::move(serv)); auto transport = Transport::make(std::move(conn), std::move(serv));
...@@ -117,7 +90,7 @@ public: ...@@ -117,7 +90,7 @@ public:
} }
private: private:
http_request_producer_ptr producer_; std::vector<net::http::router::route_ptr> routes_;
size_t max_consecutive_reads_; size_t max_consecutive_reads_;
}; };
...@@ -125,18 +98,74 @@ private: ...@@ -125,18 +98,74 @@ private:
namespace caf::net::http { namespace caf::net::http {
/// Configuration type for WebSocket clients with a handshake object. The
/// handshake object sets the default endpoint to '/' for convenience.
class server_factory_config : public dsl::config_base {
public:
using super = dsl::config_base;
explicit server_factory_config(multiplexer* mpx) : super(mpx) {
// nop
}
explicit server_factory_config(const super& other) : super(other) {
// nop
}
server_factory_config(const server_factory_config&) = default;
std::vector<router::route_ptr> routes;
};
/// Factory type for the `with(...).accept(...).start(...)` DSL. /// Factory type for the `with(...).accept(...).start(...)` DSL.
class server_factory class server_factory
: public dsl::server_factory_base<dsl::config_base, server_factory> { : public dsl::server_factory_base<server_factory_config, server_factory> {
public: public:
using super = dsl::server_factory_base<dsl::config_base, server_factory>; using super = dsl::server_factory_base<server_factory_config, server_factory>;
using config_type = typename super::config_type; using config_type = typename super::config_type;
using super::super; using super::super;
/// Starts a server that accepts incoming connections with the /// Adds a new route to the HTTP server.
/// length-prefixing protocol. /// @param path The path on this server for the new route.
/// @param f The function object for handling requests on the new route.
/// @return a reference to `*this`.
template <class F>
server_factory& route(std::string path, F f) {
auto& cfg = super::config();
if (cfg.failed())
return *this;
auto new_route = router::make_route(std::move(path), std::move(f));
if (!new_route) {
cfg.fail(std::move(new_route.error()));
} else {
cfg.routes.push_back(std::move(*new_route));
}
return *this;
}
/// Adds a new route to the HTTP server.
/// @param path The path on this server for the new route.
/// @param method The allowed HTTP method on the new route.
/// @param f The function object for handling requests on the new route.
/// @return a reference to `*this`.
template <class F>
server_factory& route(std::string path, http::method method, F f) {
auto& cfg = super::config();
if (cfg.failed())
return *this;
auto new_route = router::make_route(std::move(path), method, std::move(f));
if (!new_route) {
cfg.fail(std::move(new_route.error()));
} else {
cfg.routes.push_back(std::move(*new_route));
}
return *this;
}
/// Starts a server that makes HTTP requests without a fixed route available
/// to an observer.
template <class OnStart> template <class OnStart>
[[nodiscard]] expected<disposable> start(OnStart on_start) { [[nodiscard]] expected<disposable> start(OnStart on_start) {
using consumer_resource = async::consumer_resource<request>; using consumer_resource = async::consumer_resource<request>;
...@@ -148,16 +177,50 @@ public: ...@@ -148,16 +177,50 @@ public:
}); });
} }
/// Starts a server that only serves the fixed routes.
[[nodiscard]] expected<disposable> start() {
unit_t dummy;
return start(dummy);
}
private: private:
template <class Acceptor>
expected<disposable> do_start_impl(config_type& cfg, Acceptor acc, unit_t&) {
if (cfg.routes.empty()) {
return make_error(sec::logic_error,
"cannot start an HTTP server without any routes");
}
using transport_t = typename Acceptor::transport_type;
using factory_t = detail::http_conn_factory<transport_t>;
using impl_t = detail::accept_handler<Acceptor>;
auto factory = std::make_unique<factory_t>(cfg.routes,
cfg.max_consecutive_reads);
auto impl = impl_t::make(std::move(acc), std::move(factory),
cfg.max_connections);
auto impl_ptr = impl.get();
auto ptr = net::socket_manager::make(cfg.mpx, std::move(impl));
impl_ptr->self_ref(ptr->as_disposable());
cfg.mpx->start(ptr);
return expected<disposable>{disposable{std::move(ptr)}};
}
template <class Acceptor, class OnStart> template <class Acceptor, class OnStart>
expected<disposable> expected<disposable>
do_start_impl(config_type& cfg, Acceptor acc, OnStart& on_start) { do_start_impl(config_type& cfg, Acceptor acc, OnStart& on_start) {
using transport_t = typename Acceptor::transport_type; using transport_t = typename Acceptor::transport_type;
using factory_t = detail::http_conn_factory<transport_t>; using factory_t = detail::http_conn_factory<transport_t>;
using impl_t = detail::accept_handler<Acceptor>; using impl_t = detail::accept_handler<Acceptor>;
auto routes = cfg.routes;
auto [pull, push] = async::make_spsc_buffer_resource<request>(); auto [pull, push] = async::make_spsc_buffer_resource<request>();
auto producer = detail::http_request_producer::make(push.try_open()); auto producer = detail::http_request_producer::make(cfg.mpx,
auto factory = std::make_unique<factory_t>(std::move(producer), push.try_open());
routes.push_back(router::make_route([producer](responder& res) {
if (!producer->push(std::move(res).to_request())) {
auto err = make_error(sec::runtime_error, "flow disconnected");
res.router()->shutdown(err);
}
}));
auto factory = std::make_unique<factory_t>(std::move(routes),
cfg.max_consecutive_reads); cfg.max_consecutive_reads);
auto impl = impl_t::make(std::move(acc), std::move(factory), auto impl = impl_t::make(std::move(acc), std::move(factory),
cfg.max_connections); cfg.max_connections);
......
...@@ -6,7 +6,6 @@ ...@@ -6,7 +6,6 @@
#include "caf/byte_span.hpp" #include "caf/byte_span.hpp"
#include "caf/detail/net_export.hpp" #include "caf/detail/net_export.hpp"
#include "caf/net/http/header_fields_map.hpp"
#include "caf/net/http/status.hpp" #include "caf/net/http/status.hpp"
#include <string_view> #include <string_view>
...@@ -14,6 +13,8 @@ ...@@ -14,6 +13,8 @@
namespace caf::net::http::v1 { namespace caf::net::http::v1 {
using string_view_pair = std::pair<std::string_view, std::string_view>;
/// Tries splitting the given byte span into an HTTP header (`first`) and a /// Tries splitting the given byte span into an HTTP header (`first`) and a
/// remainder (`second`). Returns an empty `string_view` as `first` for /// remainder (`second`). Returns an empty `string_view` as `first` for
/// incomplete HTTP headers. /// incomplete HTTP headers.
...@@ -21,7 +22,8 @@ CAF_NET_EXPORT std::pair<std::string_view, byte_span> ...@@ -21,7 +22,8 @@ CAF_NET_EXPORT std::pair<std::string_view, byte_span>
split_header(byte_span bytes); split_header(byte_span bytes);
/// Writes an HTTP header to @p buf. /// Writes an HTTP header to @p buf.
CAF_NET_EXPORT void write_header(status code, const header_fields_map& fields, CAF_NET_EXPORT void write_header(status code,
span<const string_view_pair> fields,
byte_buffer& buf); byte_buffer& buf);
/// Write the status code for an HTTP header to @p buf. /// Write the status code for an HTTP header to @p buf.
...@@ -43,7 +45,7 @@ CAF_NET_EXPORT void write_response(status code, std::string_view content_type, ...@@ -43,7 +45,7 @@ CAF_NET_EXPORT void write_response(status code, std::string_view content_type,
/// and Content-Length header fields followed by the user-defined @p fields. /// and Content-Length header fields followed by the user-defined @p fields.
CAF_NET_EXPORT void write_response(status code, std::string_view content_type, CAF_NET_EXPORT void write_response(status code, std::string_view content_type,
std::string_view content, std::string_view content,
const header_fields_map& fields, span<const string_view_pair> fields,
byte_buffer& buf); byte_buffer& buf);
} // namespace caf::net::http::v1 } // namespace caf::net::http::v1
...@@ -11,6 +11,7 @@ ...@@ -11,6 +11,7 @@
#include "caf/net/dsl/has_connect.hpp" #include "caf/net/dsl/has_connect.hpp"
#include "caf/net/dsl/has_context.hpp" #include "caf/net/dsl/has_context.hpp"
#include "caf/net/http/request.hpp" #include "caf/net/http/request.hpp"
#include "caf/net/http/router.hpp"
#include "caf/net/http/server_factory.hpp" #include "caf/net/http/server_factory.hpp"
#include "caf/net/multiplexer.hpp" #include "caf/net/multiplexer.hpp"
#include "caf/net/ssl/context.hpp" #include "caf/net/ssl/context.hpp"
......
...@@ -66,6 +66,8 @@ public: ...@@ -66,6 +66,8 @@ public:
// -- implementation of lp::lower_layer ---------------------------------- // -- implementation of lp::lower_layer ----------------------------------
multiplexer& mpx() noexcept override;
bool can_send_more() const noexcept override; bool can_send_more() const noexcept override;
void request_messages() override; void request_messages() override;
......
...@@ -63,6 +63,8 @@ public: ...@@ -63,6 +63,8 @@ public:
// -- implementation of octet_stream::lower_layer ---------------------------- // -- implementation of octet_stream::lower_layer ----------------------------
multiplexer& mpx() noexcept override;
bool can_send_more() const noexcept override; bool can_send_more() const noexcept override;
void configure_read(receive_policy policy) override; void configure_read(receive_policy policy) override;
......
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/actor_system.hpp"
#include "caf/fwd.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/http/responder.hpp"
#include "caf/telemetry/collector/prometheus.hpp"
#include "caf/telemetry/importer/process.hpp"
#include <chrono>
#include <memory>
#include <string_view>
namespace caf::net::prometheus {
/// State for scraping Metrics data. May be shared among scrapers as long as
/// they don't access the state concurrently.
class scrape_state {
public:
using clock_type = std::chrono::steady_clock;
using time_point = clock_type::time_point;
using duration = clock_type::duration;
std::string_view scrape();
scrape_state(telemetry::metric_registry* ptr, timespan proc_import_interval)
: registry(ptr),
last_scrape(duration{0}),
proc_import_interval(proc_import_interval),
proc_importer(*ptr) {
// nop
}
telemetry::metric_registry* registry;
std::chrono::steady_clock::time_point last_scrape;
timespan proc_import_interval;
telemetry::importer::process proc_importer;
telemetry::collector::prometheus collector;
};
/// Creates a scraper for the given actor system.
/// @param registry The registry for collecting the metrics from.
/// @param proc_import_interval Minimum time between importing process metrics.
/// @return a function object suitable for passing it to an HTTP route.
inline auto scraper(telemetry::metric_registry* registry,
timespan proc_import_interval = std::chrono::seconds{1}) {
auto state = std::make_shared<scrape_state>(registry, proc_import_interval);
return [state](http::responder& res) {
res.respond(http::status::ok, "text/plain;version=0.0.4", state->scrape());
};
}
/// Creates a scraper for the given actor system.
/// @param sys The actor system for collecting the metrics from.
/// @param proc_import_interval Minimum time between importing process metrics.
/// @returns a function object suitable for passing it to an HTTP route.
inline auto scraper(actor_system& sys,
timespan proc_import_interval = std::chrono::seconds{1}) {
return scraper(&sys.metrics(), proc_import_interval);
}
} // namespace caf::net::prometheus
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/fwd.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/http/upper_layer.hpp"
#include "caf/telemetry/collector/prometheus.hpp"
#include "caf/telemetry/importer/process.hpp"
#include <chrono>
#include <memory>
#include <string_view>
namespace caf::net::prometheus {
/// Makes metrics available to clients via the Prometheus exposition format.
class server : public http::upper_layer {
public:
// -- member types -----------------------------------------------------------
/// State for scraping Metrics data. Shared between all server instances.
class scrape_state {
public:
using clock_type = std::chrono::steady_clock;
using time_point = clock_type::time_point;
using duration = clock_type::duration;
std::string_view scrape();
explicit scrape_state(telemetry::metric_registry* ptr)
: registry(ptr), last_scrape(duration{0}), proc_importer(*ptr) {
// nop
}
static std::shared_ptr<scrape_state> make(telemetry::metric_registry* ptr) {
return std::make_shared<scrape_state>(ptr);
}
telemetry::metric_registry* registry;
std::chrono::steady_clock::time_point last_scrape;
telemetry::importer::process proc_importer;
telemetry::collector::prometheus collector;
};
using scrape_state_ptr = std::shared_ptr<scrape_state>;
// -- factories --------------------------------------------------------------
static std::unique_ptr<server> make(scrape_state_ptr state) {
return std::unique_ptr<server>{new server(std::move(state))};
}
// -- implementation of http::upper_layer ------------------------------------
void prepare_send() override;
bool done_sending() override;
void abort(const error& reason) override;
error start(http::lower_layer* down) override;
ptrdiff_t consume(const http::request_header& hdr,
const_byte_span payload) override;
private:
explicit server(scrape_state_ptr state) : state_(std::move(state)) {
// nop
}
scrape_state_ptr state_;
http::lower_layer* down_ = nullptr;
};
} // namespace caf::net::prometheus
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/actor_system.hpp"
#include "caf/defaults.hpp"
#include "caf/detail/accept_handler.hpp"
#include "caf/detail/connection_factory.hpp"
#include "caf/fwd.hpp"
#include "caf/net/checked_socket.hpp"
#include "caf/net/dsl/server_config.hpp"
#include "caf/net/dsl/server_factory_base.hpp"
#include "caf/net/http/server.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/net/octet_stream/transport.hpp"
#include "caf/net/prometheus/server.hpp"
#include "caf/net/ssl/transport.hpp"
#include "caf/net/tcp_accept_socket.hpp"
#include "caf/none.hpp"
#include <cstdint>
#include <functional>
#include <variant>
namespace caf::detail {
template <class Transport>
class prom_conn_factory
: public connection_factory<typename Transport::connection_handle> {
public:
using connection_handle = typename Transport::connection_handle;
using state_ptr = net::prometheus::server::scrape_state_ptr;
explicit prom_conn_factory(state_ptr ptr) : ptr_(std::move(ptr)) {
// nop
}
net::socket_manager_ptr make(net::multiplexer* mpx,
connection_handle conn) override {
auto prom_serv = net::prometheus::server::make(ptr_);
auto http_serv = net::http::server::make(std::move(prom_serv));
auto transport = Transport::make(std::move(conn), std::move(http_serv));
return net::socket_manager::make(mpx, std::move(transport));
}
private:
state_ptr ptr_;
};
} // namespace caf::detail
namespace caf::net::prometheus {
/// Entry point for the `with(...).accept(...).start()` DSL.
class server_factory
: public dsl::server_factory_base<dsl::config_base, server_factory> {
public:
using super = dsl::server_factory_base<dsl::config_base, server_factory>;
using config_type = typename super::config_type;
using super::super;
/// Starts the Prometheus service in the background.
[[nodiscard]] expected<disposable> start() {
auto& cfg = super::config();
return cfg.visit([this, &cfg](auto& data) {
return do_start(cfg, data).or_else([&cfg](const error& err) { //
cfg.call_on_error(err);
});
});
}
private:
template <class Acceptor>
expected<disposable> do_start_impl(config_type& cfg, Acceptor acc) {
using transport_t = typename Acceptor::transport_type;
using factory_t = detail::prom_conn_factory<transport_t>;
using impl_t = detail::accept_handler<Acceptor>;
auto* mpx = cfg.mpx;
auto* registry = &mpx->system().metrics();
auto state = prometheus::server::scrape_state::make(registry);
auto factory = std::make_unique<factory_t>(std::move(state));
auto impl = impl_t::make(std::forward<Acceptor>(acc), std::move(factory),
cfg.max_connections);
auto mgr = socket_manager::make(mpx, std::move(impl));
mpx->start(mgr);
return expected<disposable>{mgr->as_disposable()};
}
expected<disposable> do_start(config_type& cfg,
dsl::server_config::socket& data) {
return checked_socket(data.take_fd())
.and_then(data.acceptor_with_ctx([this, &cfg](auto& acc) {
return this->do_start_impl(cfg, std::move(acc));
}));
}
expected<disposable> do_start(config_type& cfg,
dsl::server_config::lazy& data) {
return make_tcp_accept_socket(data.port, data.bind_address, data.reuse_addr)
.and_then(data.acceptor_with_ctx([this, &cfg](auto& acc) {
return this->do_start_impl(cfg, std::move(acc));
}));
}
expected<disposable> do_start(config_type&, error& err) {
return expected<disposable>{std::move(err)};
}
};
} // namespace caf::net::prometheus
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/fwd.hpp"
#include "caf/net/dsl/generic_config.hpp"
#include "caf/net/dsl/has_accept.hpp"
#include "caf/net/dsl/has_context.hpp"
#include "caf/net/prometheus/server_factory.hpp"
#include "caf/net/ssl/context.hpp"
#include "caf/net/tcp_accept_socket.hpp"
#include <cstdint>
namespace caf::net::prometheus {
/// Entry point for the `with(...).accept(...).start()` DSL.
class with_t : public extend<dsl::base, with_t>::template //
with<dsl::has_accept, dsl::has_context> {
public:
using config_type = dsl::generic_config_value<dsl::config_base>;
template <class... Ts>
explicit with_t(multiplexer* mpx) : config_(config_type::make(mpx)) {
// nop
}
with_t(const with_t&) noexcept = default;
with_t& operator=(const with_t&) noexcept = default;
/// @private
config_type& config() {
return *config_;
}
/// @private
template <class T, class... Ts>
auto make(dsl::server_config_tag<T> token, Ts&&... xs) {
return server_factory{token, *config_, std::forward<Ts>(xs)...};
}
private:
intrusive_ptr<config_type> config_;
};
inline with_t with(actor_system& sys) {
return with_t{multiplexer::from(sys)};
}
inline with_t with(multiplexer* mpx) {
return with_t{mpx};
}
} // namespace caf::net::prometheus
...@@ -79,6 +79,8 @@ public: ...@@ -79,6 +79,8 @@ public:
using web_socket::lower_layer::shutdown; using web_socket::lower_layer::shutdown;
multiplexer& mpx() noexcept override;
bool can_send_more() const noexcept override; bool can_send_more() const noexcept override;
void suspend_reading() override; void suspend_reading() override;
......
...@@ -4,7 +4,6 @@ ...@@ -4,7 +4,6 @@
#include "caf/net/http/lower_layer.hpp" #include "caf/net/http/lower_layer.hpp"
#include "caf/net/http/header_fields_map.hpp"
#include "caf/net/http/status.hpp" #include "caf/net/http/status.hpp"
#include <string> #include <string>
......
#include "caf/net/http/responder.hpp"
#include "caf/net/http/request.hpp"
#include "caf/net/http/router.hpp"
namespace caf::net::http {
lower_layer* responder::down() {
return router_->down();
}
request responder::to_request() && {
return router_->lift(std::move(*this));
}
} // namespace caf::net::http
#include "caf/net/http/router.hpp"
#include "caf/async/future.hpp"
#include "caf/disposable.hpp"
#include "caf/net/http/request.hpp"
#include "caf/net/http/responder.hpp"
#include "caf/net/multiplexer.hpp"
namespace caf::net::http {
// -- member types -------------------------------------------------------------
router::route::~route() {
// nop
}
std::pair<std::string_view, std::string_view>
router::route::next_component(const std::string_view str) {
if (str.empty() || str.front() != '/') {
return {std::string_view{}, std::string_view{}};
}
size_t start = 1;
size_t end = str.find('/', start);
auto component
= str.substr(start, end == std::string_view::npos ? end : end - start);
auto remaining = end == std::string_view::npos ? std::string_view{}
: str.substr(end);
return {component, remaining};
}
size_t router::route::args_in_path(std::string_view str) {
size_t count = 0;
size_t start = 0;
size_t end = 0;
while (end != std::string_view::npos) {
end = str.find('/', start);
auto component
= str.substr(start, end == std::string_view::npos ? end : end - start);
if (component == "<arg>")
++count;
start = end + 1;
}
return count;
}
// -- constructors and destructors ---------------------------------------------
router::~router() {
for (auto& [id, hdl] : pending_)
hdl.dispose();
}
// -- factories ----------------------------------------------------------------
std::unique_ptr<router> router::make(std::vector<route_ptr> routes) {
return std::make_unique<router>(std::move(routes));
}
// -- API for the responders ---------------------------------------------------
request router::lift(responder&& res) {
auto prom = async::promise<response>();
auto fut = prom.get_future();
auto buf = std::vector<std::byte>{res.payload().begin(), res.payload().end()};
auto impl = request::impl{res.header(), std::move(buf), std::move(prom)};
auto lifted = request{std::make_shared<request::impl>(std::move(impl))};
auto request_id = request_id_++;
auto hdl = fut.bind_to(down_->mpx())
.then(
[this, request_id](const response& res) {
down_->begin_header(res.code());
for (auto& [key, val] : res.header_fields())
down_->add_header_field(key, val);
std::ignore = down_->end_header();
down_->send_payload(res.body());
pending_.erase(request_id);
},
[this, request_id](const error& err) {
auto description = to_string(err);
down_->send_response(status::internal_server_error,
"text/plain", description);
pending_.erase(request_id);
});
pending_.emplace(request_id, std::move(hdl));
return lifted;
}
void router::shutdown(const error& err) {
abort(err);
down_->shutdown(err);
}
// -- http::upper_layer implementation -----------------------------------------
void router::prepare_send() {
// nop
}
bool router::done_sending() {
return true;
}
void router::abort(const error&) {
for (auto& [id, hdl] : pending_)
hdl.dispose();
pending_.clear();
}
error router::start(lower_layer* down) {
down_ = down;
down_->request_messages();
return {};
}
ptrdiff_t router::consume(const request_header& hdr, const_byte_span payload) {
for (auto& ptr : routes_)
if (ptr->exec(hdr, payload, this))
return static_cast<ptrdiff_t>(payload.size());
down_->send_response(http::status::not_found, "text/plain", "Not found.");
return static_cast<ptrdiff_t>(payload.size());
}
} // namespace caf::net::http
...@@ -12,6 +12,10 @@ std::unique_ptr<server> server::make(upper_layer_ptr up) { ...@@ -12,6 +12,10 @@ std::unique_ptr<server> server::make(upper_layer_ptr up) {
// -- http::lower_layer implementation ----------------------------------------- // -- http::lower_layer implementation -----------------------------------------
multiplexer& server::mpx() noexcept {
return down_->mpx();
}
bool server::can_send_more() const noexcept { bool server::can_send_more() const noexcept {
return down_->can_send_more(); return down_->can_send_more();
} }
......
...@@ -30,57 +30,4 @@ bool http_request_producer::push(const net::http::request& item) { ...@@ -30,57 +30,4 @@ bool http_request_producer::push(const net::http::request& item) {
return buf_->push(item); return buf_->push(item);
} }
// -- http_flow_adapter --------------------------------------------------------
void http_flow_adapter::prepare_send() {
// nop
}
bool http_flow_adapter::done_sending() {
return true;
}
void http_flow_adapter::abort(const error&) {
for (auto& pending : pending_)
pending.dispose();
}
error http_flow_adapter::start(net::http::lower_layer* down) {
down_ = down;
down_->request_messages();
return none;
}
ptrdiff_t http_flow_adapter::consume(const net::http::request_header& hdr,
const_byte_span payload) {
using namespace net::http;
if (!pending_.empty()) {
CAF_LOG_WARNING("received multiple requests from the same HTTP client: "
"not implemented yet (drop request)");
return static_cast<ptrdiff_t>(payload.size());
}
auto prom = async::promise<response>();
auto fut = prom.get_future();
auto buf = std::vector<std::byte>{payload.begin(), payload.end()};
auto impl = request::impl{hdr, std::move(buf), std::move(prom)};
producer_->push(request{std::make_shared<request::impl>(std::move(impl))});
auto hdl = fut.bind_to(*loop_).then(
[this](const response& res) {
down_->begin_header(res.code());
for (auto& [key, val] : res.header_fields())
down_->add_header_field(key, val);
std::ignore = down_->end_header();
down_->send_payload(res.body());
down_->shutdown();
},
[this](const error& err) {
auto description = to_string(err);
down_->send_response(status::internal_server_error, "text/plain",
description);
down_->shutdown();
});
pending_.emplace_back(std::move(hdl));
return static_cast<ptrdiff_t>(payload.size());
}
} // namespace caf::detail } // namespace caf::detail
...@@ -51,7 +51,7 @@ std::pair<std::string_view, byte_span> split_header(byte_span bytes) { ...@@ -51,7 +51,7 @@ std::pair<std::string_view, byte_span> split_header(byte_span bytes) {
} }
} }
void write_header(status code, const header_fields_map& fields, void write_header(status code, span<const string_view_pair> fields,
byte_buffer& buf) { byte_buffer& buf) {
writer out{&buf}; writer out{&buf};
out << "HTTP/1.1 "sv << std::to_string(static_cast<int>(code)) << ' ' out << "HTTP/1.1 "sv << std::to_string(static_cast<int>(code)) << ' '
...@@ -81,15 +81,14 @@ bool end_header(byte_buffer& buf) { ...@@ -81,15 +81,14 @@ bool end_header(byte_buffer& buf) {
void write_response(status code, std::string_view content_type, void write_response(status code, std::string_view content_type,
std::string_view content, byte_buffer& buf) { std::string_view content, byte_buffer& buf) {
header_fields_map fields; write_response(code, content_type, content, {}, buf);
write_response(code, content_type, content, fields, buf);
writer out{&buf}; writer out{&buf};
out << content; out << content;
} }
void write_response(status code, std::string_view content_type, void write_response(status code, std::string_view content_type,
std::string_view content, const header_fields_map& fields, std::string_view content,
byte_buffer& buf) { span<const string_view_pair> fields, byte_buffer& buf) {
writer out{&buf}; writer out{&buf};
out << "HTTP/1.1 "sv << std::to_string(static_cast<int>(code)) << ' ' out << "HTTP/1.1 "sv << std::to_string(static_cast<int>(code)) << ' '
<< phrase(code) << "\r\n"sv; << phrase(code) << "\r\n"sv;
......
...@@ -82,6 +82,10 @@ bool framing::done_sending() { ...@@ -82,6 +82,10 @@ bool framing::done_sending() {
// -- implementation of lp::lower_layer ---------------------------------------- // -- implementation of lp::lower_layer ----------------------------------------
multiplexer& framing::mpx() noexcept {
return down_->mpx();
}
bool framing::can_send_more() const noexcept { bool framing::can_send_more() const noexcept {
return down_->can_send_more(); return down_->can_send_more();
} }
......
...@@ -7,7 +7,8 @@ ...@@ -7,7 +7,8 @@
#include "caf/actor_system_config.hpp" #include "caf/actor_system_config.hpp"
#include "caf/detail/set_thread_name.hpp" #include "caf/detail/set_thread_name.hpp"
#include "caf/expected.hpp" #include "caf/expected.hpp"
#include "caf/net/prometheus/with.hpp" #include "caf/net/http/with.hpp"
#include "caf/net/prometheus.hpp"
#include "caf/net/tcp_accept_socket.hpp" #include "caf/net/tcp_accept_socket.hpp"
#include "caf/net/tcp_stream_socket.hpp" #include "caf/net/tcp_stream_socket.hpp"
#include "caf/raise_error.hpp" #include "caf/raise_error.hpp"
...@@ -34,9 +35,10 @@ bool inspect(Inspector& f, prom_config& x) { ...@@ -34,9 +35,10 @@ bool inspect(Inspector& f, prom_config& x) {
} }
void launch_prom_server(actor_system& sys, const prom_config& cfg) { void launch_prom_server(actor_system& sys, const prom_config& cfg) {
auto server = prometheus::with(sys) auto server = http::with(sys)
.accept(cfg.port, cfg.address) .accept(cfg.port, cfg.address)
.reuse_address(cfg.reuse_address) .reuse_address(cfg.reuse_address)
.route("/metrics", prometheus::scraper(sys))
.start(); .start();
if (!server) if (!server)
CAF_LOG_WARNING("failed to start Prometheus server: " << server.error()); CAF_LOG_WARNING("failed to start Prometheus server: " << server.error());
......
...@@ -32,6 +32,10 @@ std::unique_ptr<transport> transport::make(stream_socket fd, ...@@ -32,6 +32,10 @@ std::unique_ptr<transport> transport::make(stream_socket fd,
// -- implementation of octet_stream::lower_layer ------------------------------ // -- implementation of octet_stream::lower_layer ------------------------------
multiplexer& transport::mpx() noexcept {
return parent_->mpx();
}
bool transport::can_send_more() const noexcept { bool transport::can_send_more() const noexcept {
return write_buf_.size() < max_write_buf_size_; return write_buf_.size() < max_write_buf_size_;
} }
......
...@@ -2,16 +2,22 @@ ...@@ -2,16 +2,22 @@
// the main distribution directory for license terms and copyright or visit // the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE. // https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once #include "caf/net/prometheus.hpp"
#include "caf/unordered_flat_map.hpp" #include "caf/net/http/lower_layer.hpp"
#include "caf/net/http/request_header.hpp"
#include <string_view> using namespace std::literals;
namespace caf::net::http { namespace caf::net::prometheus {
/// Convenience type alias for a key-value map storing header fields. std::string_view scrape_state::scrape() {
using header_fields_map if (auto now = std::chrono::steady_clock::now();
= unordered_flat_map<std::string_view, std::string_view>; last_scrape + proc_import_interval <= now) {
last_scrape = now;
proc_importer.update();
}
return collector.collect_from(*registry);
}
} // namespace caf::net::http } // namespace caf::net::prometheus
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#include "caf/net/prometheus/server.hpp"
#include "caf/net/http/lower_layer.hpp"
#include "caf/net/http/request_header.hpp"
using namespace std::literals;
namespace caf::net::prometheus {
// -- server::scrape_state -----------------------------------------------------
std::string_view server::scrape_state::scrape() {
// Scrape system metrics at most once per second. TODO: make configurable.
if (auto now = std::chrono::steady_clock::now(); last_scrape + 1s <= now) {
last_scrape = now;
proc_importer.update();
}
return collector.collect_from(*registry);
}
// -- implementation of http::upper_layer --------------------------------------
void server::prepare_send() {
// nop
}
bool server::done_sending() {
return true;
}
void server::abort(const error&) {
// nop
}
error server::start(http::lower_layer* down) {
down_ = down;
down_->request_messages();
return caf::none;
}
ptrdiff_t server::consume(const http::request_header& hdr,
const_byte_span payload) {
if (hdr.path() != "/metrics") {
down_->send_response(http::status::not_found, "text/plain", "Not found.");
down_->shutdown();
} else if (hdr.method() != http::method::get) {
down_->send_response(http::status::method_not_allowed, "text/plain",
"Method not allowed.");
down_->shutdown();
} else if (!hdr.query().empty() || !hdr.fragment().empty()) {
down_->send_response(http::status::bad_request, "text/plain",
"No fragment or query allowed.");
down_->shutdown();
} else {
auto str = state_->scrape();
down_->send_response(http::status::ok, "text/plain;version=0.0.4", str);
down_->shutdown();
}
return static_cast<ptrdiff_t>(payload.size());
}
} // namespace caf::net::prometheus
...@@ -18,6 +18,10 @@ void framing::start(octet_stream::lower_layer* down) { ...@@ -18,6 +18,10 @@ void framing::start(octet_stream::lower_layer* down) {
// -- web_socket::lower_layer implementation ----------------------------------- // -- web_socket::lower_layer implementation -----------------------------------
multiplexer& framing::mpx() noexcept {
return down_->mpx();
}
bool framing::can_send_more() const noexcept { bool framing::can_send_more() const noexcept {
return down_->can_send_more(); return down_->can_send_more();
} }
......
...@@ -15,6 +15,10 @@ using namespace caf; ...@@ -15,6 +15,10 @@ using namespace caf;
// -- mock_stream_transport ---------------------------------------------------- // -- mock_stream_transport ----------------------------------------------------
net::multiplexer& mock_stream_transport::mpx() noexcept {
return *mpx_;
}
bool mock_stream_transport::can_send_more() const noexcept { bool mock_stream_transport::can_send_more() const noexcept {
return true; return true;
} }
......
...@@ -30,6 +30,8 @@ public: ...@@ -30,6 +30,8 @@ public:
// -- implementation of octet_stream::lower_layer ---------------------------- // -- implementation of octet_stream::lower_layer ----------------------------
caf::net::multiplexer& mpx() noexcept override;
bool can_send_more() const noexcept override; bool can_send_more() const noexcept override;
bool is_reading() const noexcept override; bool is_reading() const noexcept override;
...@@ -48,7 +50,8 @@ public: ...@@ -48,7 +50,8 @@ public:
// -- initialization --------------------------------------------------------- // -- initialization ---------------------------------------------------------
caf::error start() { caf::error start(caf::net::multiplexer* ptr) {
mpx_ = ptr;
return up->start(this); return up->start(this);
} }
...@@ -92,6 +95,8 @@ private: ...@@ -92,6 +95,8 @@ private:
caf::byte_buffer read_buf_; caf::byte_buffer read_buf_;
caf::error abort_reason_; caf::error abort_reason_;
caf::net::multiplexer* mpx_;
}; };
// Drop-in replacement for std::barrier (based on the TS API as of 2020). // Drop-in replacement for std::barrier (based on the TS API as of 2020).
......
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#define CAF_SUITE net.http.router
#include "caf/net/http/router.hpp"
#include "net-test.hpp"
#include "caf/net/multiplexer.hpp"
using namespace caf;
using namespace std::literals;
namespace http = caf::net::http;
using http::responder;
using http::router;
class mock_server : public http::lower_layer {
public:
mock_server(net::multiplexer* mpx) : mpx_(mpx) {
// nop
}
net::multiplexer& mpx() noexcept {
return *mpx_;
}
bool can_send_more() const noexcept {
return false;
}
bool is_reading() const noexcept {
return false;
}
void write_later() {
// nop
}
void shutdown() {
// nop
}
void request_messages() {
// nop
}
void suspend_reading() {
// nop
}
void begin_header(http::status) {
// nop
}
void add_header_field(std::string_view, std::string_view) {
// nop
}
bool end_header() {
return true;
}
bool send_payload(const_byte_span) {
return true;
}
bool send_chunk(const_byte_span) {
return true;
}
bool send_end_of_chunks() {
return true;
}
private:
net::multiplexer* mpx_;
};
struct fixture : test_coordinator_fixture<> {
fixture() : mpx(net::multiplexer::make(nullptr)), serv(mpx.get()) {
mpx->set_thread_id();
std::ignore = rt.start(&serv);
}
net::multiplexer_ptr mpx;
mock_server serv;
std::string req;
http::request_header hdr;
http::router rt;
std::vector<config_value> args;
void set_get_request(std::string path) {
req = "GET " + path + " HTTP/1.1\r\n"
+ "Host: localhost:8090\r\n"
"User-Agent: AwesomeLib/1.0\r\n"
"Accept-Encoding: gzip\r\n\r\n";
auto [status, err_msg] = hdr.parse(req);
REQUIRE(status == http::status::ok);
}
void set_post_request(std::string path) {
req = "POST " + path + " HTTP/1.1\r\n"
+ "Host: localhost:8090\r\n"
"User-Agent: AwesomeLib/1.0\r\n"
"Accept-Encoding: gzip\r\n\r\n";
auto [status, err_msg] = hdr.parse(req);
REQUIRE(status == http::status::ok);
}
template <class... Ts>
auto make_args(Ts... xs) {
std::vector<config_value> result;
(result.emplace_back(xs), ...);
return result;
}
};
BEGIN_FIXTURE_SCOPE(fixture)
SCENARIO("routes must have one <arg> entry per argument") {
GIVEN("a make_route call that has fewer arguments than the callback") {
WHEN("evaluating the factory call") {
THEN("the factory produces an error") {
auto res1 = router::make_route("/", [](responder&, int) {});
CHECK_EQ(res1, sec::invalid_argument);
auto res2 = router::make_route("/<arg>", [](responder&, int, int) {});
CHECK_EQ(res2, sec::invalid_argument);
}
}
}
GIVEN("a make_route call that has more arguments than the callback") {
WHEN("evaluating the factory call") {
THEN("the factory produces an error") {
auto res1 = router::make_route("/<arg>/<arg>", [](responder&) {});
CHECK_EQ(res1, sec::invalid_argument);
auto res2 = router::make_route("/<arg>/<arg>", [](responder&, int) {});
CHECK_EQ(res2, sec::invalid_argument);
}
}
}
GIVEN("a make_route call with the matching number of arguments") {
WHEN("evaluating the factory call") {
THEN("the factory produces a valid callback") {
if (auto res = router::make_route("/", [](responder&) {}); CHECK(res)) {
set_get_request("/");
CHECK((*res)->exec(hdr, {}, &rt));
set_get_request("/foo/bar");
CHECK(!(*res)->exec(hdr, {}, &rt));
}
if (auto res = router::make_route("/foo/bar", http::method::get,
[](responder&) {});
CHECK(res)) {
set_get_request("/");
CHECK(!(*res)->exec(hdr, {}, &rt));
set_get_request("/foo");
CHECK(!(*res)->exec(hdr, {}, &rt));
set_get_request("/foo/bar/baz");
CHECK(!(*res)->exec(hdr, {}, &rt));
set_post_request("/foo/bar");
CHECK(!(*res)->exec(hdr, {}, &rt));
set_get_request("/foo/bar");
CHECK((*res)->exec(hdr, {}, &rt));
}
if (auto res = router::make_route(
"/<arg>", [this](responder&, int x) { args = make_args(x); });
CHECK(res)) {
set_get_request("/");
CHECK(!(*res)->exec(hdr, {}, &rt));
set_get_request("/foo/bar");
CHECK(!(*res)->exec(hdr, {}, &rt));
set_get_request("/42");
if (CHECK((*res)->exec(hdr, {}, &rt)))
CHECK_EQ(args, make_args(42));
}
if (auto res = router::make_route("/foo/<arg>/bar",
[this](responder&, int x) {
args = make_args(x);
});
CHECK(res)) {
set_get_request("/");
CHECK(!(*res)->exec(hdr, {}, &rt));
set_get_request("/foo/bar");
CHECK(!(*res)->exec(hdr, {}, &rt));
set_get_request("/foo/123/bar");
if (CHECK((*res)->exec(hdr, {}, &rt)))
CHECK_EQ(args, make_args(123));
}
if (auto res = router::make_route("/foo/<arg>/bar",
[this](responder&, std::string x) {
args = make_args(x);
});
CHECK(res)) {
set_get_request("/");
CHECK(!(*res)->exec(hdr, {}, &rt));
set_get_request("/foo/bar");
CHECK(!(*res)->exec(hdr, {}, &rt));
set_get_request("/foo/my-arg/bar");
if (CHECK((*res)->exec(hdr, {}, &rt)))
CHECK_EQ(args, make_args("my-arg"s));
}
if (auto res
= router::make_route("/<arg>/<arg>/<arg>",
[this](responder&, int x, bool y, int z) {
args = make_args(x, y, z);
});
CHECK(res)) {
set_get_request("/");
CHECK(!(*res)->exec(hdr, {}, &rt));
set_get_request("/foo/bar");
CHECK(!(*res)->exec(hdr, {}, &rt));
set_get_request("/1/true/3?foo=bar");
if (CHECK((*res)->exec(hdr, {}, &rt)))
CHECK_EQ(args, make_args(1, true, 3));
}
}
}
}
}
END_FIXTURE_SCOPE()
...@@ -25,13 +25,6 @@ public: ...@@ -25,13 +25,6 @@ public:
// -- properties ------------------------------------------------------------- // -- properties -------------------------------------------------------------
std::string_view field(std::string_view key) {
if (auto i = hdr.fields().find(key); i != hdr.fields().end())
return i->second;
else
return {};
}
std::string_view param(std::string_view key) { std::string_view param(std::string_view key) {
auto& qm = hdr.query(); auto& qm = hdr.query();
if (auto i = qm.find(key); i != qm.end()) if (auto i = qm.find(key); i != qm.end())
...@@ -95,7 +88,7 @@ SCENARIO("the server parses HTTP GET requests into header fields") { ...@@ -95,7 +88,7 @@ SCENARIO("the server parses HTTP GET requests into header fields") {
auto app = app_ptr.get(); auto app = app_ptr.get();
auto http_ptr = net::http::server::make(std::move(app_ptr)); auto http_ptr = net::http::server::make(std::move(app_ptr));
auto serv = mock_stream_transport::make(std::move(http_ptr)); auto serv = mock_stream_transport::make(std::move(http_ptr));
CHECK_EQ(serv->start(), error{}); CHECK_EQ(serv->start(nullptr), error{});
serv->push(req); serv->push(req);
THEN("the HTTP layer parses the data and calls the application layer") { THEN("the HTTP layer parses the data and calls the application layer") {
CHECK_EQ(serv->handle_input(), static_cast<ptrdiff_t>(req.size())); CHECK_EQ(serv->handle_input(), static_cast<ptrdiff_t>(req.size()));
...@@ -103,9 +96,9 @@ SCENARIO("the server parses HTTP GET requests into header fields") { ...@@ -103,9 +96,9 @@ SCENARIO("the server parses HTTP GET requests into header fields") {
CHECK_EQ(hdr.method(), net::http::method::get); CHECK_EQ(hdr.method(), net::http::method::get);
CHECK_EQ(hdr.version(), "HTTP/1.1"); CHECK_EQ(hdr.version(), "HTTP/1.1");
CHECK_EQ(hdr.path(), "/foo/bar"); CHECK_EQ(hdr.path(), "/foo/bar");
CHECK_EQ(app->field("Host"), "localhost:8090"); CHECK_EQ(app->hdr.field("Host"), "localhost:8090");
CHECK_EQ(app->field("User-Agent"), "AwesomeLib/1.0"); CHECK_EQ(app->hdr.field("User-Agent"), "AwesomeLib/1.0");
CHECK_EQ(app->field("Accept-Encoding"), "gzip"); CHECK_EQ(app->hdr.field("Accept-Encoding"), "gzip");
} }
AND("the server properly formats a response from the application layer") { AND("the server properly formats a response from the application layer") {
CHECK_EQ(serv->output_as_str(), res); CHECK_EQ(serv->output_as_str(), res);
......
...@@ -162,7 +162,7 @@ SCENARIO("length-prefix framing reads data with 32-bit size headers") { ...@@ -162,7 +162,7 @@ SCENARIO("length-prefix framing reads data with 32-bit size headers") {
auto app = app_t<false>::make(nullptr, buf); auto app = app_t<false>::make(nullptr, buf);
auto framing = net::lp::framing::make(std::move(app)); auto framing = net::lp::framing::make(std::move(app));
auto uut = mock_stream_transport::make(std::move(framing)); auto uut = mock_stream_transport::make(std::move(framing));
CHECK_EQ(uut->start(), error{}); CHECK_EQ(uut->start(nullptr), error{});
THEN("the app receives all strings as individual messages") { THEN("the app receives all strings as individual messages") {
encode(uut->input, "hello"); encode(uut->input, "hello");
encode(uut->input, "world"); encode(uut->input, "world");
......
...@@ -36,7 +36,7 @@ SCENARIO("the Prometheus server responds to requests with scrape results") { ...@@ -36,7 +36,7 @@ SCENARIO("the Prometheus server responds to requests with scrape results") {
auto prom_serv = prometheus::server::make(prom_state); auto prom_serv = prometheus::server::make(prom_state);
auto http_serv = net::http::server::make(std::move(prom_serv)); auto http_serv = net::http::server::make(std::move(prom_serv));
auto serv = mock_stream_transport::make(std::move(http_serv)); auto serv = mock_stream_transport::make(std::move(http_serv));
CHECK_EQ(serv->start(), error{}); CHECK_EQ(serv->start(nullptr), error{});
serv->push(request_str); serv->push(request_str);
CHECK_EQ(serv->handle_input(), CHECK_EQ(serv->handle_input(),
static_cast<ptrdiff_t>(request_str.size())); static_cast<ptrdiff_t>(request_str.size()));
......
...@@ -98,7 +98,7 @@ SCENARIO("the client performs the WebSocket handshake on startup") { ...@@ -98,7 +98,7 @@ SCENARIO("the client performs the WebSocket handshake on startup") {
auto& ws_state = *ws; auto& ws_state = *ws;
auto uut = mock_stream_transport::make(std::move(ws)); auto uut = mock_stream_transport::make(std::move(ws));
THEN("the client sends its HTTP request when initializing it") { THEN("the client sends its HTTP request when initializing it") {
CHECK_EQ(uut->start(), error{}); CHECK_EQ(uut->start(nullptr), error{});
CHECK_EQ(uut->output_as_str(), http_request); CHECK_EQ(uut->output_as_str(), http_request);
} }
AND("the client waits for the server handshake and validates it") { AND("the client waits for the server handshake and validates it") {
......
...@@ -42,10 +42,11 @@ public: ...@@ -42,10 +42,11 @@ public:
put(ws, "query", hdr.query()); put(ws, "query", hdr.query());
put(ws, "fragment", hdr.fragment()); put(ws, "fragment", hdr.fragment());
put(ws, "http-version", hdr.version()); put(ws, "http-version", hdr.version());
if (!hdr.fields().empty()) { if (hdr.num_fields() > 0) {
auto& fields = ws["fields"].as_dictionary(); auto& fields = ws["fields"].as_dictionary();
for (auto& [key, val] : hdr.fields()) hdr.for_each_field([&fields](auto key, auto val) {
put(fields, std::string{key}, std::string{val}); put(fields, std::string{key}, std::string{val});
});
} }
return none; return none;
} }
...@@ -81,7 +82,7 @@ struct fixture { ...@@ -81,7 +82,7 @@ struct fixture {
auto ws_ptr = net::web_socket::server::make(std::move(app_ptr)); auto ws_ptr = net::web_socket::server::make(std::move(app_ptr));
ws = ws_ptr.get(); ws = ws_ptr.get();
transport = mock_stream_transport::make(std::move(ws_ptr)); transport = mock_stream_transport::make(std::move(ws_ptr));
if (auto err = transport->start()) if (auto err = transport->start(nullptr))
CAF_FAIL("failed to initialize mock transport: " << err); CAF_FAIL("failed to initialize mock transport: " << err);
rng.seed(0xD3ADC0D3); rng.seed(0xD3ADC0D3);
} }
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment