Commit d6461dac authored by Dominik Charousset's avatar Dominik Charousset

Implement new DSL for HTTP servers

parent ebcddbb6
......@@ -107,7 +107,6 @@ else()
endfunction()
endif()
add_net_example(http secure-time-server)
add_net_example(http time-server)
add_net_example(length_prefix_framing chat-client)
......
// Simple HTTPS server that tells the time.
#include "caf/actor_system.hpp"
#include "caf/actor_system_config.hpp"
#include "caf/caf_main.hpp"
#include "caf/deep_to_string.hpp"
#include "caf/event_based_actor.hpp"
#include "caf/net/http/serve.hpp"
#include "caf/net/middleman.hpp"
#include "caf/net/ssl/acceptor.hpp"
#include "caf/net/ssl/context.hpp"
#include "caf/net/stream_transport.hpp"
#include "caf/net/tcp_accept_socket.hpp"
#include "caf/net/tcp_stream_socket.hpp"
#include "caf/scheduled_actor/flow.hpp"
#include <iostream>
#include <string>
#include <string_view>
#include <utility>
using namespace std::literals;
static constexpr uint16_t default_port = 8080;
struct config : caf::actor_system_config {
config() {
opt_group{custom_options_, "global"} //
.add<uint16_t>("port,p", "port to listen for incoming connections")
.add<std::string>("cert-file", "PEM server certificate file")
.add<std::string>("key-file", "PEM key file for the certificate");
}
};
int caf_main(caf::actor_system& sys, const config& cfg) {
using namespace caf;
// Sanity checking.
auto cert_file = get_or(cfg, "cert-file", ""sv);
auto key_file = get_or(cfg, "key-file", ""sv);
if (cert_file.empty()) {
std::cerr << "*** mandatory parameter 'cert-file' missing or empty\n";
return EXIT_FAILURE;
}
if (key_file.empty()) {
std::cerr << "*** mandatory parameter 'key-file' missing or empty\n";
return EXIT_FAILURE;
}
// Open up a TCP port for incoming connections.
auto port = caf::get_or(cfg, "port", default_port);
auto acc = net::ssl::acceptor::make_with_cert_file(port, cert_file, key_file);
if (!acc) {
std::cerr << "*** unable to initialize TLS: " << to_string(acc.error())
<< '\n';
return EXIT_FAILURE;
}
std::cout << "*** started listening for incoming connections on port " << port
<< '\n';
// Create buffers to signal events from the WebSocket server to the worker.
auto [worker_pull, server_push] = net::http::make_request_resource();
// Spin up the HTTP server.
net::http::serve(sys, std::move(*acc), std::move(server_push));
// Spin up a worker to handle the HTTP requests.
auto worker = sys.spawn([wres = worker_pull](event_based_actor* self) {
// For each incoming request ...
wres
.observe_on(self) //
.for_each([](const net::http::request& req) {
// ... we simply return the current time as string.
// Note: we cannot respond more than once to a request.
auto str = caf::deep_to_string(make_timestamp());
req.respond(net::http::status::ok, "text/plain", str);
});
});
sys.await_all_actors_done();
return EXIT_SUCCESS;
}
CAF_MAIN(caf::net::middleman)
......@@ -5,7 +5,7 @@
#include "caf/caf_main.hpp"
#include "caf/deep_to_string.hpp"
#include "caf/event_based_actor.hpp"
#include "caf/net/http/serve.hpp"
#include "caf/net/http/with.hpp"
#include "caf/net/middleman.hpp"
#include "caf/net/stream_transport.hpp"
#include "caf/net/tcp_accept_socket.hpp"
......@@ -16,47 +16,76 @@
#include <string>
#include <utility>
// -- constants ----------------------------------------------------------------
static constexpr uint16_t default_port = 8080;
static constexpr size_t default_max_connections = 128;
// -- configuration ------------------------------------------------------------
struct config : caf::actor_system_config {
config() {
opt_group{custom_options_, "global"} //
.add<uint16_t>("port,p", "port to listen for incoming connections");
.add<uint16_t>("port,p", "port to listen for incoming connections")
.add<size_t>("max-connections,m", "limit for concurrent clients");
opt_group{custom_options_, "tls"} //
.add<std::string>("key-file,k", "path to the private key file")
.add<std::string>("cert-file,c", "path to the certificate file");
}
};
// -- main ---------------------------------------------------------------------
int caf_main(caf::actor_system& sys, const config& cfg) {
using namespace caf;
// Open up a TCP port for incoming connections.
namespace http = caf::net::http;
namespace ssl = caf::net::ssl;
// Read the configuration.
auto port = caf::get_or(cfg, "port", default_port);
net::tcp_accept_socket fd;
if (auto maybe_fd = net::make_tcp_accept_socket({ipv4_address{}, port},
true)) {
std::cout << "*** started listening for incoming connections on port "
<< port << '\n';
fd = std::move(*maybe_fd);
} else {
std::cerr << "*** unable to open port " << port << ": "
<< to_string(maybe_fd.error()) << '\n';
auto pem = ssl::format::pem;
auto key_file = caf::get_as<std::string>(cfg, "tls.key-file");
auto cert_file = caf::get_as<std::string>(cfg, "tls.cert-file");
auto max_connections = caf::get_or(cfg, "max-connections",
default_max_connections);
if (!key_file != !cert_file) {
std::cerr << "*** inconsistent TLS config: declare neither file or both\n";
return EXIT_FAILURE;
}
// Open up a TCP port for incoming connections and start the server.
auto server
= caf::net::http::with(sys)
// Optionally enable TLS.
.context(ssl::context::enable(key_file && cert_file)
.and_then(ssl::emplace_server(ssl::tls::v1_2))
.and_then(ssl::use_private_key_file(key_file, pem))
.and_then(ssl::use_certificate_file(cert_file, pem)))
// Bind to the user-defined port.
.accept(port)
// Limit how many clients may be connected at any given time.
.max_connections(max_connections)
// When started, run our worker actor to handle incoming connections.
.start([&sys](auto requests) {
// Note: requests is an async::consumer_resource<http::request>.
sys.spawn([requests](caf::event_based_actor* self) {
// For each incoming HTTP request ...
requests
.observe_on(self) //
.for_each([](const http::request& req) {
// ... 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.
if (!server) {
std::cerr << "*** unable to run at port " << port << ": "
<< to_string(server.error()) << '\n';
return EXIT_FAILURE;
}
// Create buffers to signal events from the WebSocket server to the worker.
auto [worker_pull, server_push] = net::http::make_request_resource();
// Spin up the HTTP server.
net::http::serve(sys, fd, std::move(server_push));
// Spin up a worker to handle the HTTP requests.
auto worker = sys.spawn([wres = worker_pull](event_based_actor* self) {
// For each incoming request ...
wres
.observe_on(self) //
.for_each([](const net::http::request& req) {
// ... we simply return the current time as string.
// Note: we cannot respond more than once to a request.
auto str = caf::deep_to_string(make_timestamp());
req.respond(net::http::status::ok, "text/plain", str);
});
});
sys.await_all_actors_done();
// Note: the actor system will keep the application running for as long as the
// workers are still alive.
return EXIT_SUCCESS;
}
......
......@@ -47,9 +47,8 @@ caf_add_component(
src/net/http/method.cpp
src/net/http/request.cpp
src/net/http/response.cpp
src/net/http/serve.cpp
src/net/http/serve.cpp
src/net/http/server.cpp
src/net/http/server_factory.cpp
src/net/http/status.cpp
src/net/http/upper_layer.cpp
src/net/http/v1.cpp
......
// 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/async/spsc_buffer.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/fwd.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/http/request.hpp"
#include "caf/net/ssl/fwd.hpp"
#include <memory>
#include <type_traits>
namespace caf::net::http {
/// Convenience function for creating async resources for connecting the HTTP
/// server to a worker.
inline auto make_request_resource() {
return async::make_spsc_buffer_resource<request>();
}
/// Listens for incoming HTTP requests.
/// @param sys The host system.
/// @param fd An accept socket in listening mode, already bound to a port.
/// @param out A buffer resource that connects the server to a listener that
/// processes the requests.
/// @param cfg Optional configuration parameters for the HTTP layer.
disposable CAF_NET_EXPORT serve(actor_system& sys, tcp_accept_socket fd,
async::producer_resource<request> out,
const settings& cfg = {});
/// Listens for incoming HTTPS requests.
/// @param sys The host system.
/// @param acc An SSL connection acceptor with a socket that in listening mode.
/// @param out A buffer resource that connects the server to a listener that
/// processes the requests.
/// @param cfg Optional configuration parameters for the HTTP layer.
disposable CAF_NET_EXPORT serve(actor_system& sys, ssl::acceptor acc,
async::producer_resource<request> out,
const settings& cfg = {});
} // 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/fwd.hpp"
#include "caf/net/dsl/base.hpp"
#include "caf/net/dsl/generic_config.hpp"
#include "caf/net/dsl/has_accept.hpp"
#include "caf/net/dsl/has_connect.hpp"
#include "caf/net/dsl/has_context.hpp"
#include "caf/net/http/request.hpp"
#include "caf/net/http/server_factory.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/net/ssl/acceptor.hpp"
#include "caf/net/ssl/context.hpp"
#include "caf/net/tcp_accept_socket.hpp"
#include <cstdint>
namespace caf::net::http {
/// Entry point for the `with(...)` DSL.
class with_t : public extend<dsl::base, with_t>:: //
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(with_t&&) noexcept = default;
with_t(const with_t&) noexcept = default;
with_t& operator=(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::http
#include "caf/net/http/server_factory.hpp"
namespace caf::detail {
// TODO: there is currently no back-pressure from the worker to the server.
// -- http_request_producer ----------------------------------------------------
void http_request_producer::on_consumer_ready() {
// nop
}
void http_request_producer::on_consumer_cancel() {
// nop
}
void http_request_producer::on_consumer_demand(size_t) {
// nop
}
void http_request_producer::ref_producer() const noexcept {
ref();
}
void http_request_producer::deref_producer() const noexcept {
deref();
}
bool http_request_producer::push(const net::http::request& 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::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
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