Commit dacd73df authored by Dominik Charousset's avatar Dominik Charousset

Implement http::serve overload for HTTPS

parent a22c0b2a
......@@ -139,6 +139,7 @@ else()
endfunction()
endif()
add_net_example(http secure-time-server)
add_net_example(http time-server)
add_net_example(web_socket echo)
......
// 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 <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);
net::tcp_accept_socket fd;
if (auto maybe_fd = net::make_tcp_accept_socket({ipv4_address{}, port})) {
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';
return EXIT_FAILURE;
}
// Create the OpenSSL context and set key and certificate.
auto ctx = net::ssl::context::make_server(net::ssl::tls::any);
if (!ctx) {
std::cerr << "*** unable to create SSL context: " << to_string(ctx.error())
<< '\n';
return EXIT_FAILURE;
}
if (!ctx->use_certificate_from_file(cert_file.c_str(),
net::ssl::format::pem)) {
std::cerr << "*** unable to load certificate file: "
<< ctx->last_error_string() << '\n';
return EXIT_FAILURE;
}
if (!ctx->use_private_key_from_file(key_file.c_str(),
net::ssl::format::pem)) {
std::cerr << "*** unable to load private key file: "
<< ctx->last_error_string() << '\n';
return EXIT_FAILURE;
}
// Tie context and socket up into an acceptor for the http::serve API.
auto acc = net::ssl::acceptor{fd, std::move(*ctx)};
// 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)
// Simple WebSocket server that sends everything it receives back to the sender.
// Simple HTTP server that tells the time.
#include "caf/actor_system.hpp"
#include "caf/actor_system_config.hpp"
......
......@@ -42,6 +42,7 @@ caf_add_component(
src/net/generic_upper_layer.cpp
src/net/http/header.cpp
src/net/http/lower_layer.cpp
src/net/http/serve.cpp
src/net/http/method.cpp
src/net/http/request.cpp
src/net/http/response.cpp
......@@ -61,6 +62,7 @@ caf_add_component(
src/net/socket.cpp
src/net/socket_event_layer.cpp
src/net/socket_manager.cpp
src/net/ssl/acceptor.cpp
src/net/ssl/connection.cpp
src/net/ssl/context.cpp
src/net/ssl/dtls.cpp
......
......@@ -4,54 +4,54 @@
#pragma once
#include "caf/net/connection_factory.hpp"
#include "caf/detail/connection_factory.hpp"
#include "caf/net/socket_event_layer.hpp"
#include "caf/net/socket_manager.hpp"
#include "caf/settings.hpp"
namespace caf::net {
namespace caf::detail {
/// A connection_acceptor accepts connections from an accept socket and creates
/// socket managers to handle them via its factory.
template <class Socket>
class connection_acceptor : public socket_event_layer {
/// Accepts incoming clients with an Acceptor and handles them via a connection
/// factory.
template <class Acceptor, class ConnectionHandle>
class accept_handler : public net::socket_event_layer {
public:
// -- member types -----------------------------------------------------------
using socket_type = Socket;
using socket_type = net::socket;
using connected_socket_type = typename socket_type::connected_socket_type;
using connection_handle = ConnectionHandle;
using factory_type = connection_factory<connected_socket_type>;
using factory_type = connection_factory<connection_handle>;
// -- constructors, destructors, and assignment operators --------------------
template <class FactoryPtr, class... Ts>
connection_acceptor(Socket fd, FactoryPtr fptr, size_t max_connections)
: fd_(fd),
factory_(factory_type::decorate(std::move(fptr))),
accept_handler(Acceptor acc, FactoryPtr fptr, size_t max_connections)
: acc_(std::move(acc)),
factory_(std::move(fptr)),
max_connections_(max_connections) {
CAF_ASSERT(max_connections_ > 0);
}
~connection_acceptor() {
~accept_handler() {
on_conn_close_.dispose();
if (fd_.id != invalid_socket_id)
close(fd_);
if (valid(acc_))
close(acc_);
}
// -- factories --------------------------------------------------------------
template <class FactoryPtr, class... Ts>
static std::unique_ptr<connection_acceptor>
make(Socket fd, FactoryPtr fptr, size_t max_connections) {
return std::make_unique<connection_acceptor>(fd, std::move(fptr),
max_connections);
static std::unique_ptr<accept_handler>
make(Acceptor acc, FactoryPtr fptr, size_t max_connections) {
return std::make_unique<accept_handler>(std::move(acc), std::move(fptr),
max_connections);
}
// -- implementation of socket_event_layer -----------------------------------
error start(socket_manager* owner, const settings& cfg) override {
error start(net::socket_manager* owner, const settings& cfg) override {
CAF_LOG_TRACE("");
owner_ = owner;
cfg_ = cfg;
......@@ -64,8 +64,8 @@ public:
return none;
}
socket handle() const override {
return fd_;
net::socket handle() const override {
return acc_.fd();
}
void handle_read_event() override {
......@@ -73,8 +73,8 @@ public:
CAF_ASSERT(owner_ != nullptr);
if (open_connections_ == max_connections_) {
owner_->deregister_reading();
} else if (auto x = accept(fd_)) {
socket_manager_ptr child = factory_->make(owner_->mpx_ptr(), *x);
} else if (auto conn = accept(acc_)) {
auto child = factory_->make(owner_->mpx_ptr(), std::move(*conn));
if (!child) {
CAF_LOG_ERROR("factory failed to create a new child");
on_conn_close_.dispose();
......@@ -85,12 +85,12 @@ public:
owner_->deregister_reading();
child->add_cleanup_listener(on_conn_close_);
std::ignore = child->start(cfg_);
} else if (x.error() == sec::unavailable_or_would_block) {
} else if (conn.error() == sec::unavailable_or_would_block) {
// Encountered a "soft" error: simply try again later.
CAF_LOG_DEBUG("accept failed:" << x.error());
CAF_LOG_DEBUG("accept failed:" << conn.error());
} else {
// Encountered a "hard" error: stop.
abort(x.error());
abort(conn.error());
owner_->deregister_reading();
}
}
......@@ -118,15 +118,15 @@ private:
--open_connections_;
}
Socket fd_;
Acceptor acc_;
connection_factory_ptr<connected_socket_type> factory_;
detail::connection_factory_ptr<connection_handle> factory_;
size_t max_connections_;
size_t open_connections_ = 0;
socket_manager* owner_ = nullptr;
net::socket_manager* owner_ = nullptr;
action on_conn_close_;
......@@ -138,4 +138,4 @@ private:
disposable self_ref_;
};
} // namespace caf::net
} // namespace caf::detail
// 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 <memory>
namespace caf::detail {
/// Creates new socket managers for an @ref accept_handler.
template <class ConnectionHandle>
class connection_factory {
public:
using connection_handle_type = ConnectionHandle;
virtual ~connection_factory() {
// nop
}
virtual error start(net::socket_manager*, const settings&) {
return none;
}
virtual void abort(const error&) {
// nop
}
virtual net::socket_manager_ptr make(net::multiplexer*, ConnectionHandle) = 0;
};
template <class ConnectionHandle>
using connection_factory_ptr
= std::unique_ptr<connection_factory<ConnectionHandle>>;
} // namespace caf::detail
// 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 <memory>
namespace caf::net {
/// Creates new socket managers for a @ref socket_acceptor.
template <class Socket>
class connection_factory {
public:
using socket_type = Socket;
virtual ~connection_factory() {
// nop
}
virtual error start(net::socket_manager*, const settings&) {
return none;
}
virtual void abort(const error&) {
// nop
}
virtual socket_manager_ptr make(multiplexer*, Socket) = 0;
template <class Factory>
static std::unique_ptr<connection_factory>
decorate(std::unique_ptr<Factory> ptr);
};
template <class Socket>
using connection_factory_ptr = std::unique_ptr<connection_factory<Socket>>;
template <class Socket, class DecoratedSocket>
class connection_factory_decorator : public connection_factory<Socket> {
public:
using decorated_ptr = connection_factory_ptr<DecoratedSocket>;
explicit connection_factory_decorator(decorated_ptr decorated)
: decorated_(std::move(decorated)) {
// nop
}
error start(net::socket_manager* mgr, const settings& cfg) override {
return decorated_->start(mgr, cfg);
}
void abort(const error& reason) override {
decorated_->abort(reason);
}
socket_manager_ptr make(multiplexer* mpx, Socket fd) override {
return decorated_->make(mpx, fd);
}
private:
decorated_ptr decorated_;
};
/// Lifts a factory from a subtype of `Socket` to a factory for `Socket`.
template <class Socket>
template <class Factory>
std::unique_ptr<connection_factory<Socket>>
connection_factory<Socket>::decorate(std::unique_ptr<Factory> ptr) {
using socket_sub_type = typename Factory::socket_type;
if constexpr (std::is_same_v<Socket, socket_sub_type>) {
return ptr;
} else {
static_assert(std::is_assignable_v<socket_sub_type, Socket>);
using decorator_t = connection_factory_decorator<Socket, socket_sub_type>;
return std::make_unique<decorator_t>(std::move(ptr));
}
}
} // namespace caf::net
......@@ -4,110 +4,16 @@
#pragma once
#include "caf/actor_system.hpp"
#include "caf/async/blocking_producer.hpp"
#include "caf/defaults.hpp"
#include "caf/detail/atomic_ref_counted.hpp"
#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/http/server.hpp"
#include "caf/net/middleman.hpp"
#include "caf/net/ssl/fwd.hpp"
#include <memory>
#include <type_traits>
namespace caf::detail {
class CAF_NET_EXPORT http_request_producer : public atomic_ref_counted,
public async::producer {
public:
using buffer_ptr = async::spsc_buffer_ptr<net::http::request>;
http_request_producer(buffer_ptr buf) : buf_(std::move(buf)) {
// nop
}
static auto make(buffer_ptr buf) {
auto ptr = make_counted<http_request_producer>(buf);
buf->set_producer(ptr);
return ptr;
}
void on_consumer_ready() override;
void on_consumer_cancel() override;
void on_consumer_demand(size_t) override;
void ref_producer() const noexcept override;
void deref_producer() const noexcept override;
bool push(const net::http::request& item);
private:
buffer_ptr buf_;
};
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& reason) override;
error start(net::http::lower_layer* down, const settings& config) override;
ptrdiff_t consume(const net::http::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>
class http_acceptor_factory
: public net::connection_factory<typename Transport::socket_type> {
public:
using socket_type = typename Transport::socket_type;
explicit http_acceptor_factory(http_request_producer_ptr producer)
: producer_(std::move(producer)) {
// nop
}
net::socket_manager_ptr make(net::multiplexer* mpx, socket_type fd) override {
auto app = http_flow_adapter::make(mpx, producer_);
auto serv = net::http::server::make(std::move(app));
auto transport = Transport::make(fd, std::move(serv));
auto res = net::socket_manager::make(mpx, std::move(transport));
mpx->watch(res->as_disposable());
return res;
}
private:
http_request_producer_ptr producer_;
};
} // namespace caf::detail
namespace caf::net::http {
/// Convenience function for creating async resources for connecting the HTTP
......@@ -118,29 +24,22 @@ inline auto make_request_resource() {
/// Listens for incoming HTTP requests on @p fd.
/// @param sys The host system.
/// @param fd An accept socket in listening mode. For a TCP socket, this socket
/// must already listen to a port.
/// @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 Configuration parameters for the acceptor.
template <class Transport = stream_transport, class Socket>
disposable serve(actor_system& sys, Socket fd,
async::producer_resource<request> out,
const settings& cfg = {}) {
using factory_t = detail::http_acceptor_factory<Transport>;
using impl_t = connection_acceptor<Socket>;
auto max_connections = get_or(cfg, defaults::net::max_connections);
if (auto buf = out.try_open()) {
auto& mpx = sys.network_manager().mpx();
auto producer = detail::http_request_producer::make(std::move(buf));
auto factory = std::make_unique<factory_t>(std::move(producer));
auto impl = impl_t::make(fd, std::move(factory), max_connections);
auto ptr = socket_manager::make(&mpx, std::move(impl));
mpx.start(ptr);
return ptr->as_disposable();
} else {
return disposable{};
}
}
/// @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 HTTP requests on @p fd.
/// @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
......@@ -9,7 +9,6 @@
#include "caf/detail/net_export.hpp"
#include "caf/error.hpp"
#include "caf/logger.hpp"
#include "caf/net/connection_acceptor.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/http/header.hpp"
#include "caf/net/http/header_fields_map.hpp"
......
......@@ -13,7 +13,6 @@
#include "caf/detail/net_export.hpp"
#include "caf/detail/type_list.hpp"
#include "caf/fwd.hpp"
#include "caf/net/connection_acceptor.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/net/socket_manager.hpp"
......
......@@ -5,6 +5,8 @@
#pragma once
#include "caf/actor_system.hpp"
#include "caf/detail/accept_handler.hpp"
#include "caf/detail/connection_factory.hpp"
#include "caf/fwd.hpp"
#include "caf/net/http/server.hpp"
#include "caf/net/prometheus/server.hpp"
......@@ -13,21 +15,22 @@
namespace caf::detail {
template <class Transport>
class prometheus_acceptor_factory
: public net::connection_factory<typename Transport::socket_type> {
class prometheus_conn_factory
: public connection_factory<typename Transport::connection_handle> {
public:
using state_ptr = net::prometheus::server::scrape_state_ptr;
using socket_type = typename Transport::socket_type;
using connection_handle = typename Transport::connection_handle;
explicit prometheus_acceptor_factory(state_ptr ptr) : ptr_(std::move(ptr)) {
explicit prometheus_conn_factory(state_ptr ptr) : ptr_(std::move(ptr)) {
// nop
}
net::socket_manager_ptr make(net::multiplexer* mpx, socket_type fd) override {
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(fd, std::move(http_serv));
auto transport = Transport::make(std::move(conn), std::move(http_serv));
return net::socket_manager::make(mpx, std::move(transport));
}
......@@ -45,8 +48,9 @@ namespace caf::net::prometheus {
/// must already listen to a port.
template <class Transport = stream_transport, class Socket>
disposable serve(actor_system& sys, Socket fd, const settings& cfg = {}) {
using factory_t = detail::prometheus_acceptor_factory<Transport>;
using impl_t = connection_acceptor<Socket>;
using factory_t = detail::prometheus_conn_factory<Transport>;
using conn_t = typename Transport::connection_handle;
using impl_t = detail::accept_handler<Socket, conn_t>;
auto mpx = &sys.network_manager().mpx();
auto registry = &sys.metrics();
auto state = prometheus::server::scrape_state::make(registry);
......
......@@ -45,6 +45,10 @@ struct CAF_NET_EXPORT socket : detail::comparable<socket> {
constexpr bool operator!() const noexcept {
return id == invalid_socket_id;
}
constexpr socket fd() const noexcept {
return *this;
}
};
/// @relates socket
......@@ -62,6 +66,11 @@ To socket_cast(From x) {
return To{x.id};
}
/// Checks whether `x` contains a valid ID.
constexpr bool valid(socket x) noexcept {
return x.id != invalid_socket_id;
}
/// Closes socket `x`.
/// @relates socket
void CAF_NET_EXPORT close(socket x);
......
// 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/net_export.hpp"
#include "caf/net/ssl/context.hpp"
#include "caf/net/ssl/fwd.hpp"
#include "caf/net/tcp_accept_socket.hpp"
namespace caf::net::ssl {
/// Wraps an accept socket and an SSL context.
class CAF_NET_EXPORT acceptor {
public:
// -- constructors, destructors, and assignment operators --------------------
acceptor() = delete;
acceptor(const acceptor&) = delete;
acceptor& operator=(const acceptor&) = delete;
acceptor(acceptor&& other);
acceptor& operator=(acceptor&& other);
acceptor(tcp_accept_socket fd, context ctx) : fd_(fd), ctx_(std::move(ctx)) {
// nop
}
// -- properties -------------------------------------------------------------
tcp_accept_socket fd() const noexcept {
return fd_;
}
context& ctx() noexcept {
return ctx_;
}
const context& ctx() const noexcept {
return ctx_;
}
private:
tcp_accept_socket fd_;
context ctx_;
};
/// Checks whether `acc` has a valid socket descriptor.
bool CAF_NET_EXPORT valid(const acceptor& acc);
/// Closes the socket of `obj`.
void CAF_NET_EXPORT close(acceptor& acc);
/// Tries to accept a new connection on `acc`. On success, wraps the new socket
/// into an SSL @ref connection and returns it.
expected<connection> CAF_NET_EXPORT accept(acceptor& acc);
} // namespace caf::net::ssl
......@@ -85,6 +85,14 @@ public:
/// Returns the file descriptor for this connection.
stream_socket fd() const noexcept;
bool valid() const noexcept {
return pimpl_ != nullptr;
}
explicit operator bool() const noexcept {
return valid();
}
private:
constexpr explicit connection(impl* ptr) : pimpl_(ptr) {
// nop
......@@ -93,4 +101,8 @@ private:
impl* pimpl_;
};
inline bool valid(const connection& conn) {
return conn.valid();
}
} // namespace caf::net::ssl
......@@ -11,6 +11,8 @@
#include "caf/net/ssl/tls.hpp"
#include "caf/net/stream_socket.hpp"
#include <string>
namespace caf::net::ssl {
struct close_on_shutdown_t {};
......@@ -53,6 +55,16 @@ public:
static expected<context> make_client(dtls min_version,
dtls max_version = dtls::any);
// -- properties -------------------------------------------------------------
explicit operator bool() const noexcept {
return pimpl_ != nullptr;
}
bool operator!() const noexcept {
return pimpl_ == nullptr;
}
// -- native handles ---------------------------------------------------------
/// Reinterprets `native_handle` as the native implementation type and takes
......
......@@ -6,6 +6,7 @@
namespace caf::net::ssl {
class acceptor;
class connection;
class context;
class transport;
......
......@@ -18,6 +18,8 @@ public:
using super = stream_transport;
using connection_handle = connection;
using worker_ptr = std::unique_ptr<socket_event_layer>;
class policy_impl : public stream_transport::policy {
......
......@@ -23,6 +23,10 @@ struct CAF_NET_EXPORT stream_socket : network_socket {
using super = network_socket;
using super::super;
constexpr stream_socket fd() const noexcept {
return *this;
}
};
/// Creates two connected sockets to mimic network communication (usually for
......
......@@ -24,6 +24,8 @@ public:
using socket_type = stream_socket;
using connection_handle = stream_socket;
/// An owning smart pointer type for storing an upper layer object.
using upper_layer_ptr = std::unique_ptr<stream_oriented::upper_layer>;
......@@ -153,6 +155,10 @@ public:
return *up_;
}
policy& active_policy() {
return *policy_;
}
// -- implementation of socket_event_layer -----------------------------------
error start(socket_manager* owner, const settings& config) override;
......
......@@ -18,8 +18,6 @@ struct CAF_NET_EXPORT tcp_accept_socket : network_socket {
using super = network_socket;
using super::super;
using connected_socket_type = tcp_stream_socket;
};
/// Creates a new TCP socket to accept connections on a given port.
......@@ -29,7 +27,7 @@ struct CAF_NET_EXPORT tcp_accept_socket : network_socket {
/// @relates tcp_accept_socket
expected<tcp_accept_socket>
CAF_NET_EXPORT make_tcp_accept_socket(ip_endpoint node,
bool reuse_addr = false);
bool reuse_addr = true);
/// Creates a new TCP socket to accept connections on a given port.
/// @param node The endpoint to listen on and the filter for incoming addresses.
......@@ -39,7 +37,7 @@ expected<tcp_accept_socket>
/// @relates tcp_accept_socket
expected<tcp_accept_socket>
CAF_NET_EXPORT make_tcp_accept_socket(const uri::authority_type& node,
bool reuse_addr = false);
bool reuse_addr = true);
/// Creates a new TCP socket to accept connections on a given port.
/// @param port The port for listening to incoming connection. Passing 0 lets
......@@ -49,7 +47,7 @@ expected<tcp_accept_socket>
/// @param reuse_addr Optionally sets the SO_REUSEADDR option on the socket.
/// @relates tcp_accept_socket
expected<tcp_accept_socket> CAF_NET_EXPORT make_tcp_accept_socket(
uint16_t port, std::string addr = "0.0.0.0", bool reuse_addr = false);
uint16_t port, std::string addr = "0.0.0.0", bool reuse_addr = true);
/// Accepts a connection on `x`.
/// @param x Listening endpoint.
......
......@@ -6,6 +6,8 @@
#include "caf/actor_system.hpp"
#include "caf/cow_tuple.hpp"
#include "caf/detail/accept_handler.hpp"
#include "caf/detail/connection_factory.hpp"
#include "caf/net/flow_connector.hpp"
#include "caf/net/web_socket/default_trait.hpp"
#include "caf/net/web_socket/flow_bridge.hpp"
......@@ -19,14 +21,14 @@
namespace caf::detail {
template <class Transport, class Trait>
class ws_acceptor_factory
: public net::connection_factory<typename Transport::socket_type> {
class ws_conn_factory
: public connection_factory<typename Transport::connection_handle> {
public:
using socket_type = typename Transport::socket_type;
using connector_pointer = net::flow_connector_ptr<Trait>;
explicit ws_acceptor_factory(connector_pointer connector)
explicit ws_conn_factory(connector_pointer connector)
: connector_(std::move(connector)) {
// nop
}
......@@ -86,13 +88,15 @@ disposable accept(actor_system& sys, Socket fd, acceptor_resource_t<Ts...> out,
using request_t = request<default_trait, Ts...>;
static_assert(std::is_invocable_v<OnRequest, const settings&, request_t&>,
"invalid signature found for on_request");
using factory_t = detail::ws_acceptor_factory<Transport, trait_t>;
using impl_t = connection_acceptor<Socket>;
using conn_t = flow_connector_request_impl<OnRequest, trait_t, Ts...>;
using factory_t = detail::ws_conn_factory<Transport, trait_t>;
using conn_t = typename Transport::connection_handle;
using impl_t = detail::accept_handler<Socket, conn_t>;
using connector_t = flow_connector_request_impl<OnRequest, trait_t, Ts...>;
auto max_connections = get_or(cfg, defaults::net::max_connections);
if (auto buf = out.try_open()) {
auto& mpx = sys.network_manager().mpx();
auto conn = std::make_shared<conn_t>(std::move(on_request), std::move(buf));
auto conn = std::make_shared<connector_t>(std::move(on_request),
std::move(buf));
auto factory = std::make_unique<factory_t>(std::move(conn));
auto impl = impl_t::make(fd, std::move(factory), max_connections);
auto impl_ptr = impl.get();
......
......@@ -8,7 +8,6 @@
#include "caf/detail/net_export.hpp"
#include "caf/error.hpp"
#include "caf/logger.hpp"
#include "caf/net/connection_acceptor.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/http/header.hpp"
#include "caf/net/http/method.hpp"
......
......@@ -4,86 +4,209 @@
#include "caf/net/http/serve.hpp"
#include "caf/detail/accept_handler.hpp"
#include "caf/detail/connection_factory.hpp"
#include "caf/logger.hpp"
#include "caf/net/http/lower_layer.hpp"
#include "caf/net/http/server.hpp"
#include "caf/net/http/upper_layer.hpp"
#include "caf/net/middleman.hpp"
#include "caf/net/ssl/acceptor.hpp"
#include "caf/net/ssl/transport.hpp"
#include "caf/net/stream_transport.hpp"
#include "caf/net/tcp_accept_socket.hpp"
#include "caf/net/tcp_stream_socket.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
}
class http_request_producer : public atomic_ref_counted,
public async::producer {
public:
using buffer_ptr = async::spsc_buffer_ptr<net::http::request>;
void http_request_producer::on_consumer_cancel() {
}
http_request_producer(buffer_ptr buf) : buf_(std::move(buf)) {
// nop
}
void http_request_producer::on_consumer_demand(size_t) {
// nop
}
static auto make(buffer_ptr buf) {
auto ptr = make_counted<http_request_producer>(buf);
buf->set_producer(ptr);
return ptr;
}
void http_request_producer::ref_producer() const noexcept {
ref();
}
void on_consumer_ready() override {
// nop
}
void http_request_producer::deref_producer() const noexcept {
deref();
}
void on_consumer_cancel() override {
// nop
}
bool http_request_producer::push(const net::http::request& item) {
return buf_->push(item);
}
void on_consumer_demand(size_t) override {
// nop
}
void ref_producer() const noexcept override {
ref();
}
void deref_producer() const noexcept override {
deref();
}
bool push(const net::http::request& item) {
return buf_->push(item);
}
private:
buffer_ptr buf_;
};
using http_request_producer_ptr = intrusive_ptr<http_request_producer>;
// -- http_flow_adapter --------------------------------------------------------
void http_flow_adapter::prepare_send() {
// nop
}
class 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
}
bool http_flow_adapter::done_sending() {
return true;
}
void prepare_send() override {
// nop
}
void http_flow_adapter::abort(const error&) {
for (auto& pending : pending_)
pending.dispose();
}
bool done_sending() override {
return true;
}
error http_flow_adapter::start(net::http::lower_layer* down, const settings&) {
down_ = down;
down_->request_messages();
return none;
}
void abort(const error&) override {
for (auto& pending : pending_)
pending.dispose();
}
error start(net::http::lower_layer* down, const settings&) override {
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)");
ptrdiff_t consume(const net::http::header& hdr,
const_byte_span payload) override {
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());
}
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());
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_;
};
// -- http_conn_factory --------------------------------------------------------
template <class Transport>
class http_conn_factory
: public connection_factory<typename Transport::connection_handle> {
public:
using connection_handle = typename Transport::connection_handle;
explicit http_conn_factory(http_request_producer_ptr producer)
: producer_(std::move(producer)) {
// nop
}
net::socket_manager_ptr make(net::multiplexer* mpx,
connection_handle conn) override {
auto app = http_flow_adapter::make(mpx, producer_);
auto serv = net::http::server::make(std::move(app));
auto fd = conn.fd();
auto transport = Transport::make(std::move(conn), std::move(serv));
transport->active_policy().accept(fd);
auto res = net::socket_manager::make(mpx, std::move(transport));
mpx->watch(res->as_disposable());
return res;
}
private:
http_request_producer_ptr producer_;
};
// -- http_serve_impl ----------------------------------------------------------
template <class Transport, class Acceptor>
disposable http_serve_impl(actor_system& sys, Acceptor acc,
async::producer_resource<net::http::request> out,
const settings& cfg = {}) {
using factory_t = http_conn_factory<Transport>;
using conn_t = typename Transport::connection_handle;
using impl_t = accept_handler<Acceptor, conn_t>;
auto max_conn = get_or(cfg, defaults::net::max_connections);
if (auto buf = out.try_open()) {
auto& mpx = sys.network_manager().mpx();
auto producer = http_request_producer::make(std::move(buf));
auto factory = std::make_unique<factory_t>(std::move(producer));
auto impl = impl_t::make(std::move(acc), std::move(factory), max_conn);
auto ptr = net::socket_manager::make(&mpx, std::move(impl));
mpx.start(ptr);
return ptr->as_disposable();
} else {
return disposable{};
}
}
} // namespace caf::detail
namespace caf::net::http {
disposable serve(actor_system& sys, tcp_accept_socket fd,
async::producer_resource<request> out, const settings& cfg) {
return detail::http_serve_impl<stream_transport>(sys, fd, std::move(out),
cfg);
}
disposable serve(actor_system& sys, ssl::acceptor acc,
async::producer_resource<request> out, const settings& cfg) {
return detail::http_serve_impl<ssl::transport>(sys, std::move(acc),
std::move(out), 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.
#include "caf/net/ssl/acceptor.hpp"
#include "caf/expected.hpp"
#include "caf/net/ssl/connection.hpp"
#include "caf/net/tcp_stream_socket.hpp"
namespace caf::net::ssl {
// -- constructors, destructors, and assignment operators ----------------------
acceptor::acceptor(acceptor&& other)
: fd_(other.fd_), ctx_(std::move(other.ctx_)) {
other.fd_.id = invalid_socket_id;
}
acceptor& acceptor::operator=(acceptor&& other) {
fd_ = other.fd_;
ctx_ = std::move(other.ctx_);
other.fd_.id = invalid_socket_id;
return *this;
}
bool valid(const acceptor& acc) {
return valid(acc.fd());
}
void close(acceptor& acc) {
close(acc.fd());
}
expected<connection> accept(acceptor& acc) {
if (auto fd = accept(acc.fd()); fd) {
return acc.ctx().new_connection(*fd);
} else {
return expected<connection>{std::move(fd.error())};
}
}
} // namespace caf::net::ssl
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