Commit c6c05de4 authored by Dominik Charousset's avatar Dominik Charousset

Add utilities for REST APIs plus an example

parent 7994c1a5
......@@ -89,6 +89,7 @@ else()
endfunction()
endif()
add_net_example(http rest)
add_net_example(http time-server)
add_net_example(length_prefix_framing chat-client)
......
// This example an HTTP server that implements a REST API by forwarding requests
// to an actor. The actor in this example is a simple key-value store. The actor
// is not aware of HTTP and the HTTP server is sending regular request messages
// to actor.
#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/with.hpp"
#include "caf/net/middleman.hpp"
#include "caf/scheduled_actor/flow.hpp"
#include <algorithm>
#include <cctype>
#include <iostream>
#include <string>
#include <utility>
namespace http = caf::net::http;
// -- 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<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");
}
};
// -- our key-value store actor ------------------------------------------------
struct kvs_actor_state {
explicit kvs_actor_state(caf::event_based_actor* self_ptr) : self(self_ptr) {
// nop
}
caf::behavior make_behavior() {
return {
[this](caf::get_atom, const std::string& key) -> std::string {
if (auto i = data.find(key); i != data.end())
return i->second;
else
return {};
},
[this](caf::put_atom, const std::string& key, std::string& value) {
data.insert_or_assign(key, std::move(value));
},
[this](caf::delete_atom, const std::string& key) { data.erase(key); },
};
}
std::map<std::string, std::string> data;
caf::event_based_actor* self;
};
using kvs_actor_impl = caf::stateful_actor<kvs_actor_state>;
// -- utility functions --------------------------------------------------------
bool is_ascii(caf::span<const std::byte> buffer) {
auto pred = [](auto x) { return isascii(static_cast<unsigned char>(x)); };
return std::all_of(buffer.begin(), buffer.end(), pred);
}
std::string to_ascii(caf::span<const std::byte> buffer) {
return std::string{reinterpret_cast<const char*>(buffer.data()),
buffer.size()};
}
// -- main ---------------------------------------------------------------------
int caf_main(caf::actor_system& sys, const config& cfg) {
using namespace std::literals;
namespace ssl = caf::net::ssl;
// Read the configuration.
auto port = caf::get_or(cfg, "port", default_port);
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;
}
// Spin up our key-value store actor.
auto kvs = sys.spawn<kvs_actor_impl>();
// Open up a TCP port for incoming connections and start the server.
auto server
= 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)
// Stop the server if our key-value store actor terminates.
.monitor(kvs)
// Forward incoming requests to the kvs actor.
.route("/api/<arg>", http::method::get,
[kvs](http::responder& res, std::string key) {
auto* self = res.self();
auto prom = std::move(res).to_promise();
self->request(kvs, 2s, caf::get_atom_v, std::move(key))
.then(
[prom](const std::string& value) mutable {
prom.respond(http::status::ok, "text/plain", value);
},
[prom](const caf::error& what) mutable {
prom.respond(http::status::internal_server_error, what);
});
})
.route("/api/<arg>", http::method::post,
[kvs](http::responder& res, std::string key) {
auto value = res.payload();
if (!is_ascii(value)) {
res.respond(http::status::bad_request, "text/plain",
"Expected an ASCII payload.");
return;
}
auto* self = res.self();
auto prom = std::move(res).to_promise();
self
->request(kvs, 2s, caf::put_atom_v, std::move(key),
to_ascii(value))
.then(
[prom]() mutable {
prom.respond(http::status::no_content);
},
[prom](const caf::error& what) mutable {
prom.respond(http::status::internal_server_error, what);
});
})
.route("/api/<arg>", http::method::del,
[kvs](http::responder& res, std::string key) {
auto* self = res.self();
auto prom = std::move(res).to_promise();
self->request(kvs, 2s, caf::delete_atom_v, std::move(key))
.then(
[prom]() mutable {
prom.respond(http::status::no_content);
},
[prom](const caf::error& what) mutable {
prom.respond(http::status::internal_server_error, what);
});
})
// Launch the server.
.start();
// Report any error to the user.
if (!server) {
std::cerr << "*** unable to run at port " << port << ": "
<< to_string(server.error()) << '\n';
return EXIT_FAILURE;
}
// Note: the actor system will keep the application running for as long as the
// kvs actor stays alive.
return EXIT_SUCCESS;
}
CAF_MAIN(caf::net::middleman)
......@@ -4,7 +4,9 @@
#pragma once
#include "caf/async/execution_context.hpp"
#include "caf/detail/connection_factory.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/net/socket_event_layer.hpp"
#include "caf/net/socket_manager.hpp"
#include "caf/settings.hpp"
......@@ -30,10 +32,12 @@ public:
// -- constructors, destructors, and assignment operators --------------------
accept_handler(Acceptor acc, factory_ptr fptr, size_t max_connections)
accept_handler(Acceptor acc, factory_ptr fptr, size_t max_connections,
std::vector<strong_actor_ptr> monitored_actors = {})
: acc_(std::move(acc)),
factory_(std::move(fptr)),
max_connections_(max_connections) {
max_connections_(max_connections),
monitored_actors_(std::move(monitored_actors)) {
CAF_ASSERT(max_connections_ > 0);
}
......@@ -41,14 +45,18 @@ public:
on_conn_close_.dispose();
if (valid(acc_))
close(acc_);
if (monitor_callback_)
monitor_callback_.dispose();
}
// -- factories --------------------------------------------------------------
static std::unique_ptr<accept_handler> make(Acceptor acc, factory_ptr fptr,
size_t max_connections) {
static std::unique_ptr<accept_handler>
make(Acceptor acc, factory_ptr fptr, size_t max_connections,
std::vector<strong_actor_ptr> monitored_actors = {}) {
return std::make_unique<accept_handler>(std::move(acc), std::move(fptr),
max_connections);
max_connections,
std::move(monitored_actors));
}
// -- implementation of socket_event_layer -----------------------------------
......@@ -60,6 +68,17 @@ public:
CAF_LOG_DEBUG("factory_->start failed:" << err);
return err;
}
if (!monitored_actors_.empty()) {
monitor_callback_ = make_action([this] { owner_->shutdown(); });
auto ctx = async::execution_context_ptr{owner_->mpx_ptr()};
for (auto& hdl : monitored_actors_) {
CAF_ASSERT(hdl);
hdl->get()->attach_functor([ctx, cb = monitor_callback_] {
if (!cb.disposed())
ctx->schedule(cb);
});
}
}
on_conn_close_ = make_action([this] { connection_closed(); });
owner->register_reading();
return none;
......@@ -135,6 +154,12 @@ private:
/// to keep the acceptor alive while the manager is not registered for writing
/// or reading.
disposable self_ref_;
/// An action for stopping this handler if an observed actor terminates.
action monitor_callback_;
/// List of actors that we add monitors to in `start`.
std::vector<strong_actor_ptr> monitored_actors_;
};
} // namespace caf::detail
......@@ -45,6 +45,10 @@ public:
/// Sends the last chunk, completing a chunked payload.
virtual bool send_end_of_chunks() = 0;
/// Sends a response that only consists of a header with a status code such as
/// `status::no_content`.
bool send_response(status code);
/// Convenience function for sending header and payload. Automatically sets
/// the header fields `Content-Type` and `Content-Length`.
bool send_response(status code, std::string_view content_type,
......@@ -54,6 +58,9 @@ public:
bool send_response(status code, std::string_view content_type,
std::string_view content);
/// @copydoc send_response
bool send_response(status code, const error& err);
/// Asks the stream to swap the HTTP layer with `next` after returning from
/// `consume`.
/// @note may only be called from the upper layer in `consume`.
......
......@@ -23,6 +23,87 @@ namespace caf::net::http {
/// response immediately.
class CAF_NET_EXPORT responder {
public:
// -- member types -----------------------------------------------------------
/// Implementation detail for `promise`.
class CAF_NET_EXPORT promise_state : public ref_counted {
public:
explicit promise_state(lower_layer* down) : down_(down) {
// nop
}
promise_state(const promise_state&) = delete;
promise_state& operator-(const promise_state&) = delete;
~promise_state() override;
/// Returns a pointer to the HTTP layer.
lower_layer* down() {
return down_;
}
/// Marks the promise as fulfilled
void set_completed() {
completed_ = true;
}
private:
lower_layer* down_;
bool completed_ = false;
};
/// Allows users to respond to an incoming HTTP request at some later time.
class CAF_NET_EXPORT promise {
public:
explicit promise(responder& parent);
promise(const promise&) noexcept = default;
promise& operator=(const promise&) noexcept = default;
/// Sends an HTTP response that only consists of a header with a status code
/// such as `status::no_content`.
void respond(status code) {
impl_->down()->send_response(code);
impl_->set_completed();
}
/// Sends an HTTP response message to the client. Automatically sets the
/// `Content-Type` and `Content-Length` header fields.
void respond(status code, std::string_view content_type,
const_byte_span content) {
impl_->down()->send_response(code, content_type, content);
impl_->set_completed();
}
/// Sends an HTTP response message to the client. Automatically sets the
/// `Content-Type` and `Content-Length` header fields.
void respond(status code, std::string_view content_type,
std::string_view content) {
impl_->down()->send_response(code, content_type, content);
impl_->set_completed();
}
/// Sends an HTTP response message with an error to the client.
/// Converts @p what to a string representation and then transfers this
/// description to the client.
void respond(status code, const error& what) {
impl_->down()->send_response(code, what);
impl_->set_completed();
}
/// Returns a pointer to the HTTP layer.
lower_layer* down() {
return impl_->down();
}
private:
intrusive_ptr<promise_state> impl_;
};
// -- constructors, destructors, and assignment operators --------------------
responder(const request_header* hdr, const_byte_span body,
http::router* router)
: hdr_(hdr), body_(body), router_(router) {
......@@ -33,6 +114,8 @@ public:
responder& operator=(const responder&) noexcept = default;
// -- properties -------------------------------------------------------------
/// Returns the HTTP header for the responder.
/// @pre `valid()`
const request_header& header() const noexcept {
......@@ -55,6 +138,21 @@ public:
return router_;
}
/// Returns the @ref actor_shell object from the @ref router for interacting
/// with actors in the system.
actor_shell* self();
/// Returns a pointer to the HTTP layer.
lower_layer* down();
// -- responding -------------------------------------------------------------
/// Sends an HTTP response that only consists of a header with a status code
/// such as `status::no_content`.
void respond(status code) {
down()->send_response(code);
}
/// Sends an HTTP response message to the client. Automatically sets the
/// `Content-Type` and `Content-Length` header fields.
/// @pre `valid()`
......@@ -71,6 +169,13 @@ public:
down()->send_response(code, content_type, content);
}
/// Sends an HTTP response message with an error to the client.
/// Converts @p what to a string representation and then transfers this
/// description to the client.
void respond(status code, const error& what) {
down()->send_response(code, what);
}
/// Starts writing an HTTP header.
void begin_header(status code) {
down()->begin_header(code);
......@@ -103,12 +208,15 @@ public:
return down()->send_end_of_chunks();
}
// -- conversions ------------------------------------------------------------
/// Converts a responder to a @ref request for processing the HTTP request
/// asynchronously.
request to_request() &&;
/// Returns a pointer to the HTTP layer.
lower_layer* down();
/// Converts the responder to a promise object for responding to the HTTP
/// request at some later time but from the same @ref socket_manager.
promise to_promise() &&;
private:
const request_header* hdr_;
......
......@@ -9,6 +9,7 @@
#include "caf/detail/type_traits.hpp"
#include "caf/expected.hpp"
#include "caf/intrusive_ptr.hpp"
#include "caf/net/actor_shell.hpp"
#include "caf/net/http/arg_parser.hpp"
#include "caf/net/http/lower_layer.hpp"
#include "caf/net/http/responder.hpp"
......@@ -44,10 +45,15 @@ public:
// -- properties -------------------------------------------------------------
/// Returns a pointer to the underlying HTTP layer.
lower_layer* down() {
return down_;
}
/// Returns an @ref actor_shell for this router that enables routes to
/// interact with actors.
actor_shell* self();
// -- API for the responders -------------------------------------------------
/// Lifts a @ref responder to an @ref request object that allows asynchronous
......@@ -70,10 +76,20 @@ public:
void abort(const error& reason) override;
private:
/// Handle to the underlying HTTP layer.
lower_layer* down_ = nullptr;
/// List of user-defined routes.
std::vector<route_ptr> routes_;
/// Generates ascending IDs for `pending_`.
size_t request_id_ = 0;
/// Keeps track of pending HTTP requests when lifting @ref responder objects.
std::unordered_map<size_t, disposable> pending_;
/// Lazily initialized for allowing a @ref route to interact with actors.
actor_shell_ptr shell_;
};
} // namespace caf::net::http
......@@ -92,6 +92,7 @@ public:
private:
std::vector<net::http::route_ptr> routes_;
size_t max_consecutive_reads_;
action monitor_;
};
} // namespace caf::detail
......@@ -115,6 +116,7 @@ public:
server_factory_config(const server_factory_config&) = default;
std::vector<route_ptr> routes;
std::vector<strong_actor_ptr> monitored_actors;
};
/// Factory type for the `with(...).accept(...).start(...)` DSL.
......@@ -127,6 +129,22 @@ public:
using super::super;
/// Monitors the actor handle @p hdl and stops the server if the monitored
/// actor terminates.
template <class ActorHandle>
server_factory& monitor(const ActorHandle& hdl) {
auto& cfg = super::config();
auto ptr = actor_cast<strong_actor_ptr>(hdl);
if (!ptr) {
auto err = make_error(sec::logic_error,
"cannot monitor an invalid actor handle");
cfg.fail(std::move(err));
return *this;
}
cfg.monitored_actors.push_back(std::move(ptr));
return *this;
}
/// Adds a new route to the HTTP server.
/// @param path The path on this server for the new route.
/// @param f The function object for handling requests on the new route.
......@@ -198,7 +216,7 @@ private:
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);
cfg.max_connections, cfg.monitored_actors);
auto impl_ptr = impl.get();
auto ptr = net::socket_manager::make(cfg.mpx, std::move(impl));
impl_ptr->self_ref(ptr->as_disposable());
......@@ -225,7 +243,7 @@ private:
auto factory = std::make_unique<factory_t>(std::move(routes),
cfg.max_consecutive_reads);
auto impl = impl_t::make(std::move(acc), std::move(factory),
cfg.max_connections);
cfg.max_connections, cfg.monitored_actors);
auto impl_ptr = impl.get();
auto ptr = net::socket_manager::make(cfg.mpx, std::move(impl));
impl_ptr->self_ref(ptr->as_disposable());
......
......@@ -6,7 +6,7 @@
namespace caf::net {
/// Creates a new @ref actor_shell and registers it at the actor system.
template <class Handle>
template <class Handle> // Note: default is caf::actor; see fwd.hpp
actor_shell_ptr_t<Handle>
make_actor_shell(actor_system& sys, async::execution_context_ptr loop) {
auto f = [](abstract_actor_shell* self, message& msg) -> result<message> {
......
......@@ -17,6 +17,11 @@ lower_layer::~lower_layer() {
// nop
}
bool lower_layer::send_response(status code) {
begin_header(code);
return end_header() && send_payload(const_byte_span{});
}
bool lower_layer::send_response(status code, std::string_view content_type,
const_byte_span content) {
auto content_size = std::to_string(content.size());
......@@ -31,4 +36,9 @@ bool lower_layer::send_response(status code, std::string_view content_type,
return send_response(code, content_type, as_bytes(make_span(content)));
}
bool lower_layer::send_response(status code, const error& err) {
auto msg = to_string(err);
return send_response(code, "text/plain", msg);
}
} // namespace caf::net::http
......@@ -5,6 +5,22 @@
namespace caf::net::http {
responder::promise_state::~promise_state() {
if (!completed_) {
down_->send_response(status::internal_server_error, "text/plain",
"Internal server error: broken responder promise.");
}
}
responder::promise::promise(responder& parent)
: impl_(make_counted<promise_state>(parent.down())) {
// nop
}
actor_shell* responder::self() {
return router_->self();
}
lower_layer* responder::down() {
return router_->down();
}
......@@ -13,4 +29,8 @@ request responder::to_request() && {
return router_->lift(std::move(*this));
}
responder::promise responder::to_promise() && {
return promise{*this};
}
} // namespace caf::net::http
......@@ -4,6 +4,7 @@
#include "caf/disposable.hpp"
#include "caf/net/http/request.hpp"
#include "caf/net/http/responder.hpp"
#include "caf/net/make_actor_shell.hpp"
#include "caf/net/multiplexer.hpp"
namespace caf::net::http {
......@@ -21,6 +22,16 @@ std::unique_ptr<router> router::make(std::vector<route_ptr> routes) {
return std::make_unique<router>(std::move(routes));
}
// -- properties ---------------------------------------------------------------
actor_shell* router::self() {
if (shell_)
return shell_.get();
auto& mpx = down_->mpx();
shell_ = make_actor_shell(mpx.system(), &mpx);
return shell_.get();
}
// -- API for the responders ---------------------------------------------------
request router::lift(responder&& res) {
......
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