Commit 8454845b authored by Dominik Charousset's avatar Dominik Charousset

Streamline socket manager and processing layer API

parent 2da2b91a
......@@ -20,7 +20,7 @@ namespace caf::net::binary {
/// Translates between a message-oriented transport and data flows.
template <class Trait>
class flow_bridge : public web_socket::upper_layer {
class flow_bridge : public upper_layer {
public:
using input_type = typename Trait::input_type;
......@@ -56,7 +56,7 @@ public:
// -- implementation of web_socket::lower_layer ------------------------------
error init(net::socket_manager* mgr, web_socket::lower_layer* down,
error start(net::socket_manager* mgr, web_socket::lower_layer* down,
const settings& cfg) override {
CAF_ASSERT(mgr != nullptr);
down_ = down;
......
......@@ -21,7 +21,7 @@ public:
/// @param down A pointer to the lower layer that remains valid for the
/// lifetime of the upper layer.
/// @param config Protocol-dependent configuration parameters.
virtual error init(lower_layer* down, const settings& config) = 0;
virtual error start(lower_layer* down, const settings& config) = 0;
/// Consumes bytes from the lower layer.
/// @param payload Payload of the received message.
......
......@@ -40,16 +40,20 @@ public:
// -- implementation of socket_event_layer -----------------------------------
error init(socket_manager* owner, const settings& cfg) override {
error start(socket_manager* owner, const settings& cfg) override {
CAF_LOG_TRACE("");
owner_ = owner;
cfg_ = cfg;
if (auto err = factory_.init(owner, cfg))
if (auto err = factory_.start(owner, cfg))
return err;
owner->register_reading();
return none;
}
socket handle() const override {
return fd_;
}
void handle_read_event() override {
CAF_LOG_TRACE("");
CAF_ASSERT(owner_ != nullptr);
......@@ -60,7 +64,7 @@ public:
owner_->deregister();
return;
}
if (auto err = child->init(cfg_)) {
if (auto err = child->start(cfg_)) {
CAF_LOG_ERROR("failed to initialize new child:" << err);
owner_->deregister();
return;
......
......@@ -56,9 +56,9 @@ public:
// -- public member functions ------------------------------------------------
error init(endpoint_manager& manager) override {
error start(endpoint_manager& manager) override {
CAF_LOG_TRACE("");
if (auto err = super::init(manager))
if (auto err = super::start(manager))
return err;
prepare_next_read();
return none;
......
......@@ -64,7 +64,7 @@ public:
void abort(const error& reason) override;
error init(net::http::lower_layer* down, const settings& config) 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;
......@@ -89,7 +89,7 @@ public:
// nop
}
error init(net::socket_manager*, const settings&) {
error start(net::socket_manager*, const settings&) {
return none;
}
......@@ -98,7 +98,7 @@ public:
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, fd, std::move(transport));
auto res = net::socket_manager::make(mpx, std::move(transport));
mpx->watch(res->as_disposable());
return res;
}
......@@ -139,8 +139,8 @@ disposable serve(actor_system& sys, Socket fd,
auto producer = detail::http_request_producer::make(std::move(buf));
auto factory = factory_t{std::move(producer)};
auto impl = impl_t::make(fd, limit, std::move(factory));
auto ptr = socket_manager::make(&mpx, fd, std::move(impl));
mpx.init(ptr);
auto ptr = socket_manager::make(&mpx, std::move(impl));
mpx.start(ptr);
return ptr->as_disposable();
} else {
return disposable{};
......
......@@ -103,7 +103,7 @@ public:
// -- stream_oriented::upper_layer implementation ----------------------------
error init(stream_oriented::lower_layer* down,
error start(stream_oriented::lower_layer* down,
const settings& config) override;
void abort(const error& reason) override;
......
......@@ -20,7 +20,7 @@ public:
/// Initializes the upper layer.
/// @param down A pointer to the lower layer that remains valid for the
/// lifetime of the upper layer.
virtual error init(lower_layer* down, const settings& config) = 0;
virtual error start(lower_layer* down, const settings& config) = 0;
/// Consumes an HTTP message.
/// @param hdr The header fields for the received message.
......
......@@ -52,7 +52,7 @@ public:
// -- implementation of stream_oriented::upper_layer -------------------------
error init(stream_oriented::lower_layer* down,
error start(stream_oriented::lower_layer* down,
const settings& config) override;
void abort(const error& reason) override;
......
......@@ -64,7 +64,7 @@ public:
// nop
}
error init(net::socket_manager* mgr, binary::lower_layer* down,
error start(net::socket_manager* mgr, binary::lower_layer* down,
const settings&) {
down_ = down;
if (auto in = make_consumer_adapter(in_res_, mgr->mpx_ptr(),
......
......@@ -109,7 +109,7 @@ public:
/// Registers `mgr` for initialization in the multiplexer's thread.
/// @thread-safe
void init(socket_manager_ptr mgr);
void start(socket_manager_ptr mgr);
/// Signals the multiplexer to initiate shutdown.
/// @thread-safe
......@@ -227,7 +227,7 @@ private:
void do_register_reading(const socket_manager_ptr& mgr);
void do_init(const socket_manager_ptr& mgr);
void do_start(const socket_manager_ptr& mgr);
};
} // namespace caf::net
......@@ -25,7 +25,7 @@ public:
using msg_buf = std::array<std::byte, sizeof(intptr_t) + 1>;
enum class code : uint8_t {
init_manager,
start_manager,
shutdown_reading,
shutdown_writing,
run_action,
......@@ -42,7 +42,9 @@ public:
// -- implementation of socket_event_layer -----------------------------------
error init(socket_manager* owner, const settings& cfg) override;
error start(socket_manager* owner, const settings& cfg) override;
socket handle() const override;
void handle_read_event() override;
......
......@@ -21,7 +21,7 @@ public:
// nop
}
error init(net::socket_manager*, const settings&) {
error start(net::socket_manager*, const settings&) {
return none;
}
......@@ -30,7 +30,7 @@ public:
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));
return net::socket_manager::make(mpx, fd, std::move(transport));
return net::socket_manager::make(mpx, std::move(transport));
}
void abort(const error&) {
......@@ -58,8 +58,8 @@ disposable serve(actor_system& sys, Socket fd) {
auto state = prometheus::server::scrape_state::make(registry);
auto factory = factory_t{std::move(state)};
auto impl = impl_t::make(fd, 0, std::move(factory));
auto mgr = socket_manager::make(mpx, fd, std::move(impl));
mpx->init(mgr);
auto mgr = socket_manager::make(mpx, std::move(impl));
mpx->start(mgr);
return mgr->as_disposable();
}
......
......@@ -61,7 +61,7 @@ public:
void abort(const error& reason) override;
error init(http::lower_layer* down, const settings& config) override;
error start(http::lower_layer* down, const settings& config) override;
ptrdiff_t consume(const http::header& hdr, const_byte_span payload) override;
......
......@@ -16,8 +16,11 @@ class CAF_NET_EXPORT socket_event_layer {
public:
virtual ~socket_event_layer();
/// Initializes the layer.
virtual error init(socket_manager* owner, const settings& cfg) = 0;
/// Starts processing on this layer.
virtual error start(socket_manager* owner, const settings& cfg) = 0;
/// Returns the handle for the managed socket.
virtual socket handle() const = 0;
/// Handles a read event on the managed socket.
virtual void handle_read_event() = 0;
......
......@@ -37,7 +37,7 @@ public:
/// @pre `handle != invalid_socket`
/// @pre `mpx!= nullptr`
socket_manager(multiplexer* mpx, socket handle, event_handler_ptr handler);
socket_manager(multiplexer* mpx, event_handler_ptr handler);
~socket_manager() override;
......@@ -47,8 +47,7 @@ public:
// -- factories --------------------------------------------------------------
static socket_manager_ptr make(multiplexer* mpx, socket handle,
event_handler_ptr handler);
static socket_manager_ptr make(multiplexer* mpx, event_handler_ptr handler);
// -- properties -------------------------------------------------------------
......@@ -127,8 +126,8 @@ public:
// -- callbacks for the multiplexer ------------------------------------------
/// Initializes the manager and its all of its sub-components.
error init(const settings& cfg);
/// Starts the manager and its all of its processing layers.
error start(const settings& cfg);
/// Called whenever the socket received new data.
void handle_read_event();
......
......@@ -23,7 +23,7 @@ public:
/// Initializes the upper layer.
/// @param down A pointer to the lower layer that remains valid for the
/// lifetime of the upper layer.
virtual error init(lower_layer* down, const settings& config) = 0;
virtual error start(lower_layer* down, const settings& config) = 0;
/// Consumes bytes from the lower layer.
/// @param buffer Available bytes to read.
......
......@@ -155,7 +155,9 @@ public:
// -- implementation of socket_event_layer -----------------------------------
error init(socket_manager* owner, const settings& config) override;
error start(socket_manager* owner, const settings& config) override;
socket handle() const override;
void handle_read_event() override;
......
......@@ -27,7 +27,7 @@ public:
// nop
}
error init(net::socket_manager*, const settings&) {
error start(net::socket_manager*, const settings&) {
return none;
}
......@@ -36,7 +36,7 @@ public:
auto app = net::web_socket::flow_bridge<Trait>::make(mpx, connector_);
auto ws = net::web_socket::server::make(std::move(app));
auto transport = Transport::make(fd, std::move(ws));
return net::socket_manager::make(mpx, fd, std::move(transport));
return net::socket_manager::make(mpx, std::move(transport));
}
void abort(const error&) {
......@@ -97,8 +97,8 @@ void accept(actor_system& sys, Socket fd, acceptor_resource_t<Ts...> out,
auto conn = std::make_shared<conn_t>(std::move(on_request), std::move(buf));
auto factory = factory_t{std::move(conn)};
auto impl = impl_t::make(fd, limit, std::move(factory));
auto ptr = socket_manager::make(&mpx, fd, std::move(impl));
mpx.init(ptr);
auto ptr = socket_manager::make(&mpx, std::move(impl));
mpx.start(ptr);
}
// TODO: accept() should return a disposable to stop the WebSocket server.
}
......
......@@ -71,7 +71,7 @@ public:
// -- implementation of stream_oriented::upper_layer -------------------------
error init(stream_oriented::lower_layer* down,
error start(stream_oriented::lower_layer* down,
const settings& config) override;
void abort(const error& reason) override;
......
......@@ -36,8 +36,8 @@ void connect(actor_system& sys, Socket fd, handshake hs, Init init) {
auto bridge = flow_bridge<trait_t>::make(mpx, std::move(conn));
auto impl = client::make(std::move(hs), std::move(bridge));
auto transport = Transport::make(fd, std::move(impl));
auto ptr = socket_manager::make(mpx, fd, std::move(transport));
mpx->init(ptr);
auto ptr = socket_manager::make(mpx, std::move(transport));
mpx->start(ptr);
init(connect_event_t{app_pull, app_push});
}
......
......@@ -65,7 +65,7 @@ public:
// -- implementation of web_socket::lower_layer ------------------------------
error init(web_socket::lower_layer* down, const settings& cfg) override {
error start(web_socket::lower_layer* down, const settings& cfg) override {
down_ = down;
auto [err, pull, push] = conn_->on_request(cfg);
if (!err) {
......
......@@ -49,7 +49,7 @@ public:
// -- initialization ---------------------------------------------------------
void init(stream_oriented::lower_layer* down);
void start(stream_oriented::lower_layer* down);
// -- properties -------------------------------------------------------------
......
......@@ -74,7 +74,7 @@ public:
// -- stream_oriented::upper_layer implementation ----------------------------
error init(stream_oriented::lower_layer* down,
error start(stream_oriented::lower_layer* down,
const settings& config) override;
void abort(const error& reason) override;
......
......@@ -16,7 +16,7 @@ class CAF_NET_EXPORT upper_layer : public generic_upper_layer {
public:
virtual ~upper_layer();
virtual error init(lower_layer* down, const settings& cfg) = 0;
virtual error start(lower_layer* down, const settings& cfg) = 0;
virtual ptrdiff_t consume_binary(byte_span buf) = 0;
virtual ptrdiff_t consume_text(std::string_view buf) = 0;
};
......
......@@ -48,7 +48,7 @@ void http_flow_adapter::abort(const error&) {
pending.dispose();
}
error http_flow_adapter::init(net::http::lower_layer* down, const settings&) {
error http_flow_adapter::start(net::http::lower_layer* down, const settings&) {
down_ = down;
down_->request_messages();
return none;
......
......@@ -80,11 +80,11 @@ bool server::send_end_of_chunks() {
// -- stream_oriented::upper_layer implementation ------------------------------
error server::init(stream_oriented::lower_layer* down, const settings& cfg) {
error server::start(stream_oriented::lower_layer* down, const settings& cfg) {
down_ = down;
if (auto max_size = get_as<uint32_t>(cfg, "http.max-request-size"))
max_request_size_ = *max_size;
if (auto err = up_->init(this, cfg))
if (auto err = up_->start(this, cfg))
return err;
else
return none;
......
......@@ -24,10 +24,10 @@ length_prefix_framing::make(upper_layer_ptr up) {
// -- implementation of stream_oriented::upper_layer ---------------------------
error length_prefix_framing::init(stream_oriented::lower_layer* down,
error length_prefix_framing::start(stream_oriented::lower_layer* down,
const settings& cfg) {
down_ = down;
return up_->init(this, cfg);
return up_->start(this, cfg);
}
void length_prefix_framing::abort(const error& reason) {
......
......@@ -110,9 +110,8 @@ error multiplexer::init() {
if (!pipe_handles)
return std::move(pipe_handles.error());
auto updater = pollset_updater::make(pipe_handles->first);
auto mgr = socket_manager::make(this, pipe_handles->first,
std::move(updater));
if (auto err = mgr->init(settings{}))
auto mgr = socket_manager::make(this, std::move(updater));
if (auto err = mgr->start(settings{}))
return err;
write_handle_ = pipe_handles->second;
pollset_.emplace_back(pollfd{pipe_handles->first.id, input_mask, 0});
......@@ -182,12 +181,12 @@ void multiplexer::watch(disposable what) {
// -- thread-safe signaling ----------------------------------------------------
void multiplexer::init(socket_manager_ptr mgr) {
void multiplexer::start(socket_manager_ptr mgr) {
CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle().id));
if (std::this_thread::get_id() == tid_) {
do_init(mgr);
do_start(mgr);
} else {
write_to_pipe(pollset_updater::code::init_manager, mgr.release());
write_to_pipe(pollset_updater::code::start_manager, mgr.release());
}
}
......@@ -455,14 +454,14 @@ void multiplexer::do_shutdown() {
apply_updates();
}
void multiplexer::do_init(const socket_manager_ptr& mgr) {
void multiplexer::do_start(const socket_manager_ptr& mgr) {
CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle().id));
if (!shutting_down_) {
error err;
if (owner_)
err = mgr->init(content(system().config()));
err = mgr->start(content(system().config()));
else
err = mgr->init(settings{});
err = mgr->start(settings{});
if (err) {
CAF_LOG_DEBUG("mgr->init failed: " << err);
// The socket manager should not register itself for any events if
......
......@@ -29,13 +29,17 @@ std::unique_ptr<pollset_updater> pollset_updater::make(pipe_socket fd) {
// -- interface functions ------------------------------------------------------
error pollset_updater::init(socket_manager* owner, const settings&) {
error pollset_updater::start(socket_manager* owner, const settings&) {
CAF_LOG_TRACE("");
owner_ = owner;
mpx_ = owner->mpx_ptr();
return nonblocking(fd_, true);
}
socket pollset_updater::handle() const {
return fd_;
}
void pollset_updater::handle_read_event() {
CAF_LOG_TRACE("");
auto as_mgr = [](intptr_t ptr) {
......@@ -57,8 +61,8 @@ void pollset_updater::handle_read_event() {
intptr_t ptr;
memcpy(&ptr, buf_.data() + 1, sizeof(intptr_t));
switch (static_cast<code>(opcode)) {
case code::init_manager:
mpx_->do_init(as_mgr(ptr));
case code::start_manager:
mpx_->do_start(as_mgr(ptr));
break;
case code::run_action:
run_action(ptr);
......
......@@ -36,7 +36,7 @@ void server::abort(const error&) {
// nop
}
error server::init(http::lower_layer* down, const settings&) {
error server::start(http::lower_layer* down, const settings&) {
down_ = down;
down_->request_messages();
return caf::none;
......
......@@ -13,9 +13,11 @@ namespace caf::net {
// -- constructors, destructors, and assignment operators ----------------------
socket_manager::socket_manager(multiplexer* mpx, socket fd,
event_handler_ptr handler)
: fd_(fd), mpx_(mpx), handler_(std::move(handler)), disposed_(false) {
socket_manager::socket_manager(multiplexer* mpx, event_handler_ptr handler)
: fd_(handler->handle()),
mpx_(mpx),
handler_(std::move(handler)),
disposed_(false) {
CAF_ASSERT(fd_ != invalid_socket);
CAF_ASSERT(mpx_ != nullptr);
CAF_ASSERT(handler_ != nullptr);
......@@ -28,11 +30,10 @@ socket_manager::~socket_manager() {
// -- factories ----------------------------------------------------------------
socket_manager_ptr socket_manager::make(multiplexer* mpx, socket handle,
socket_manager_ptr socket_manager::make(multiplexer* mpx,
event_handler_ptr handler) {
CAF_ASSERT(mpx != nullptr);
return make_counted<socket_manager>(std::move(mpx), handle,
std::move(handler));
return make_counted<socket_manager>(std::move(mpx), std::move(handler));
}
// -- properties ---------------------------------------------------------------
......@@ -108,13 +109,14 @@ void socket_manager::shutdown() {
// -- callbacks for the multiplexer --------------------------------------------
error socket_manager::init(const settings& cfg) {
error socket_manager::start(const settings& cfg) {
CAF_LOG_TRACE(CAF_ARG(cfg));
if (auto err = nonblocking(fd_, true)) {
CAF_LOG_ERROR("failed to set nonblocking flag in socket:" << err);
return err;
} else {
return handler_->start(this, cfg);
}
return handler_->init(this, cfg);
}
void socket_manager::handle_read_event() {
......
......@@ -35,7 +35,7 @@ public:
// -- interface functions ----------------------------------------------------
error init(socket_manager* owner, const settings& cfg) override {
error start(socket_manager* owner, const settings& cfg) override {
owner_ = owner;
cfg_ = cfg;
owner->register_writing();
......@@ -102,7 +102,7 @@ public:
bool do_handover(std::unique_ptr<socket_event_layer>& next) override {
next = transport::make(std::move(policy_.conn), std::move(up_));
if (auto err = next->init(owner_, cfg_))
if (auto err = next->start(owner_, cfg_))
return false;
else
return true;
......
......@@ -127,7 +127,7 @@ void stream_transport::shutdown() {
// -- implementation of transport ----------------------------------------------
error stream_transport::init(socket_manager* owner, const settings& config) {
error stream_transport::start(socket_manager* owner, const settings& config) {
parent_ = owner;
namespace mm = defaults::middleman;
auto default_max_reads = static_cast<uint32_t>(mm::max_consecutive_reads);
......@@ -145,7 +145,11 @@ error stream_transport::init(socket_manager* owner, const settings& config) {
CAF_LOG_ERROR("send_buffer_size: " << socket_buf_size.error());
return std::move(socket_buf_size.error());
}
return up_->init(this, config);
return up_->start(this, config);
}
socket stream_transport::handle() const {
return fd_;
}
void stream_transport::handle_read_event() {
......
......@@ -38,9 +38,9 @@ std::unique_ptr<client> client::make(handshake_ptr hs, upper_layer_ptr up) {
// -- implementation of stream_oriented::upper_layer ---------------------------
error client::init(stream_oriented::lower_layer* down, const settings& cfg) {
error client::start(stream_oriented::lower_layer* down, const settings& cfg) {
CAF_ASSERT(hs_ != nullptr);
framing_.init(down);
framing_.start(down);
if (!hs_->has_mandatory_fields())
return make_error(sec::runtime_error,
"handshake data lacks mandatory fields");
......@@ -110,7 +110,7 @@ bool client::handle_header(std::string_view http) {
auto http_ok = hs_->is_valid_http_1_response(http);
hs_.reset();
if (http_ok) {
if (auto err = upper_layer().init(&framing_, cfg_)) {
if (auto err = upper_layer().start(&framing_, cfg_)) {
CAF_LOG_DEBUG("failed to initialize WebSocket framing layer");
return false;
} else {
......
......@@ -10,7 +10,7 @@ namespace caf::net::web_socket {
// -- initialization ---------------------------------------------------------
void framing::init(stream_oriented::lower_layer* down) {
void framing::start(stream_oriented::lower_layer* down) {
std::random_device rd;
rng_.seed(rd());
down_ = down;
......
......@@ -22,8 +22,8 @@ std::unique_ptr<server> server::make(upper_layer_ptr up) {
// -- stream_oriented::upper_layer implementation ------------------------------
error server::init(stream_oriented::lower_layer* down, const settings& cfg) {
framing_.init(down);
error server::start(stream_oriented::lower_layer* down, const settings& cfg) {
framing_.start(down);
cfg_ = cfg;
lower_layer().configure_read(receive_policy::up_to(handshake::max_http_size));
return none;
......@@ -126,7 +126,7 @@ bool server::handle_header(std::string_view http) {
put(fields, std::string{key}, std::string{val});
}
// Try to initialize the upper layer.
if (auto err = upper_layer().init(&framing_, cfg_)) {
if (auto err = upper_layer().start(&framing_, cfg_)) {
auto descr = to_string(err);
CAF_LOG_DEBUG("upper layer rejected a WebSocket connection:" << descr);
write_response(http::status::bad_request, descr);
......
......@@ -48,13 +48,13 @@ public:
// -- initialization ---------------------------------------------------------
caf::error init(const caf::settings& cfg) {
return up->init(this, cfg);
caf::error start(const caf::settings& cfg) {
return up->start(this, cfg);
}
caf::error init() {
caf::error start() {
caf::settings cfg;
return init(cfg);
return start(cfg);
}
// -- buffer management ------------------------------------------------------
......
......@@ -38,7 +38,7 @@ public:
return std::make_unique<app_t>(sys, std::move(loop), std::move(hdl));
}
error init(net::stream_oriented::lower_layer* down,
error start(net::stream_oriented::lower_layer* down,
const settings&) override {
this->down = down;
self->set_behavior([this](std::string& line) {
......@@ -196,8 +196,8 @@ CAF_TEST(actor shells expose their mailbox to their owners) {
auto app_uptr = app_t::make(sys, mpx);
auto app = app_uptr.get();
auto transport = net::stream_transport::make(fd, std::move(app_uptr));
auto mgr = net::socket_manager::make(mpx.get(), fd, std::move(transport));
if (auto err = mgr->init(content(cfg)))
auto mgr = net::socket_manager::make(mpx.get(), std::move(transport));
if (auto err = mgr->start(content(cfg)))
CAF_FAIL("mgr->init() failed: " << err);
auto hdl = app->self.as_actor();
anon_send(hdl, "line 1");
......@@ -219,9 +219,9 @@ CAF_TEST(actor shells can send requests and receive responses) {
auto app_uptr = app_t::make(sys, mpx, worker);
auto app = app_uptr.get();
auto transport = net::stream_transport::make(fd, std::move(app_uptr));
auto mgr = net::socket_manager::make(mpx.get(), fd, std::move(transport));
if (auto err = mgr->init(content(cfg)))
CAF_FAIL("mgr->init() failed: " << err);
auto mgr = net::socket_manager::make(mpx.get(), std::move(transport));
if (auto err = mgr->start(content(cfg)))
CAF_FAIL("mgr->start() failed: " << err);
send(input);
run_while([&] { return app->consumed_bytes != input.size(); });
expect((int32_t), to(worker).with(123));
......
......@@ -48,7 +48,7 @@ public:
// -- implementation of http::upper_layer ------------------------------------
error init(net::http::lower_layer* down_ptr, const settings&) override {
error start(net::http::lower_layer* down_ptr, const settings&) override {
down = down_ptr;
down->request_messages();
return none;
......@@ -95,7 +95,7 @@ SCENARIO("the server parses HTTP GET requests into header fields") {
auto app = app_ptr.get();
auto http_ptr = net::http::server::make(std::move(app_ptr));
auto serv = mock_stream_transport::make(std::move(http_ptr));
CHECK_EQ(serv->init(settings{}), error{});
CHECK_EQ(serv->start(settings{}), error{});
serv->push(req);
THEN("the HTTP layer parses the data and calls the application layer") {
CHECK_EQ(serv->handle_input(), static_cast<ptrdiff_t>(req.size()));
......
......@@ -48,7 +48,7 @@ public:
return std::make_unique<app_t>(std::move(loop), std::move(inputs));
}
caf::error init(net::binary::lower_layer* down_ptr,
caf::error start(net::binary::lower_layer* down_ptr,
const settings&) override {
// Start reading immediately.
down = down_ptr;
......@@ -144,7 +144,7 @@ SCENARIO("length-prefix framing reads data with 32-bit size headers") {
auto app = app_t<false>::make(nullptr, buf);
auto framing = net::length_prefix_framing::make(std::move(app));
auto uut = mock_stream_transport::make(std::move(framing));
CHECK_EQ(uut->init(), error{});
CHECK_EQ(uut->start(), error{});
THEN("the app receives all strings as individual messages") {
encode(uut->input, "hello");
encode(uut->input, "world");
......@@ -191,8 +191,8 @@ SCENARIO("calling suspend_reading temporarily halts receiving of messages") {
auto app_ptr = app.get();
auto framing = net::length_prefix_framing::make(std::move(app));
auto transport = net::stream_transport::make(fd2, std::move(framing));
auto mgr = net::socket_manager::make(mpx.get(), fd2, std::move(transport));
CHECK_EQ(mgr->init(settings{}), none);
auto mgr = net::socket_manager::make(mpx.get(), std::move(transport));
CHECK_EQ(mgr->start(settings{}), none);
mpx->apply_updates();
REQUIRE_EQ(mpx->num_socket_managers(), 2u);
CHECK_EQ(mpx->mask_of(mgr), net::operation::read);
......
......@@ -63,11 +63,15 @@ public:
// -- implementation of socket_event_layer -----------------------------------
error init(net::socket_manager* mgr, const settings&) override {
error start(net::socket_manager* mgr, const settings&) override {
mgr_ = mgr;
return none;
}
net::socket handle() const override {
return fd_;
}
void handle_read_event() override {
if (read_capacity() < 1024)
rd_buf_.resize(rd_buf_.size() + 2048);
......@@ -151,8 +155,8 @@ struct fixture {
make_manager(net::stream_socket fd, std::string name) {
auto mock = mock_event_layer::make(fd, std::move(name), manager_count);
auto mock_ptr = mock.get();
auto mgr = net::socket_manager::make(mpx.get(), fd, std::move(mock));
std::ignore = mgr->init(settings{});
auto mgr = net::socket_manager::make(mpx.get(), std::move(mock));
std::ignore = mgr->start(settings{});
return {mock_ptr, std::move(mgr)};
}
......
......@@ -36,7 +36,7 @@ SCENARIO("the Prometheus server responds to requests with scrape results") {
auto prom_serv = prometheus::server::make(prom_state);
auto http_serv = net::http::server::make(std::move(prom_serv));
auto serv = mock_stream_transport::make(std::move(http_serv));
CHECK_EQ(serv->init(settings{}), error{});
CHECK_EQ(serv->start(settings{}), error{});
serv->push(request_str);
CHECK_EQ(serv->handle_input(),
static_cast<ptrdiff_t>(request_str.size()));
......
......@@ -59,7 +59,7 @@ public:
*done_ = true;
}
error init(net::stream_oriented::lower_layer* down,
error start(net::stream_oriented::lower_layer* down,
const settings&) override {
MESSAGE("initialize dummy app");
down_ = down;
......@@ -197,9 +197,8 @@ SCENARIO("ssl::transport::make_client performs the client handshake") {
auto mock = mock_application::make(done, buf);
auto transport = ssl::transport::make_client(std::move(conn),
std::move(mock));
auto mgr = net::socket_manager::make(mpx.get(), client_fd,
std::move(transport));
mpx->init(mgr);
auto mgr = net::socket_manager::make(mpx.get(), std::move(transport));
mpx->start(mgr);
mpx->apply_updates();
while (!*done)
mpx->poll_once(true);
......@@ -239,9 +238,8 @@ SCENARIO("ssl::transport::make_server performs the server handshake") {
auto mock = mock_application::make(done, buf);
auto transport = ssl::transport::make_server(std::move(conn),
std::move(mock));
auto mgr = net::socket_manager::make(mpx.get(), server_fd,
std::move(transport));
mpx->init(mgr);
auto mgr = net::socket_manager::make(mpx.get(), std::move(transport));
mpx->start(mgr);
mpx->apply_updates();
while (!*done)
mpx->poll_once(true);
......
......@@ -75,7 +75,7 @@ public:
net::stream_oriented::lower_layer* down;
error init(net::stream_oriented::lower_layer* down_ptr,
error start(net::stream_oriented::lower_layer* down_ptr,
const settings&) override {
down = down_ptr;
down->configure_read(net::receive_policy::exactly(hello_manager.size()));
......@@ -116,11 +116,10 @@ BEGIN_FIXTURE_SCOPE(fixture)
CAF_TEST(receive) {
auto mock = mock_application::make(shared_recv_buf, shared_send_buf);
auto transport = net::stream_transport::make(recv_socket_guard.get(),
auto transport = net::stream_transport::make(recv_socket_guard.release(),
std::move(mock));
auto mgr = net::socket_manager::make(mpx.get(), recv_socket_guard.release(),
std::move(transport));
CHECK_EQ(mgr->init(config), none);
auto mgr = net::socket_manager::make(mpx.get(), std::move(transport));
CHECK_EQ(mgr->start(config), none);
mpx->apply_updates();
CHECK_EQ(mpx->num_socket_managers(), 2u);
CHECK_EQ(static_cast<size_t>(write(send_socket_guard.socket(),
......@@ -135,11 +134,10 @@ CAF_TEST(receive) {
CAF_TEST(send) {
auto mock = mock_application::make(shared_recv_buf, shared_send_buf);
auto transport = net::stream_transport::make(recv_socket_guard.get(),
auto transport = net::stream_transport::make(recv_socket_guard.release(),
std::move(mock));
auto mgr = net::socket_manager::make(mpx.get(), recv_socket_guard.release(),
std::move(transport));
CHECK_EQ(mgr->init(config), none);
auto mgr = net::socket_manager::make(mpx.get(), std::move(transport));
CHECK_EQ(mgr->start(config), none);
mpx->apply_updates();
CHECK_EQ(mpx->num_socket_managers(), 2u);
mgr->register_writing();
......
......@@ -41,7 +41,7 @@ public:
return std::make_unique<app_t>(sys, std::move(loop), std::move(hdl));
}
error init(net::stream_oriented::lower_layer* down,
error start(net::stream_oriented::lower_layer* down,
const settings&) override {
this->down = down;
self->set_behavior([this](std::string& line) {
......@@ -196,9 +196,9 @@ CAF_TEST(actor shells expose their mailbox to their owners) {
auto app_uptr = app_t::make(sys, mpx);
auto app = app_uptr.get();
auto transport = net::stream_transport::make(fd, std::move(app_uptr));
auto mgr = net::socket_manager::make(mpx.get(), fd, std::move(transport));
if (auto err = mgr->init(content(cfg)))
CAF_FAIL("mgr->init() failed: " << err);
auto mgr = net::socket_manager::make(mpx.get(), std::move(transport));
if (auto err = mgr->start(content(cfg)))
CAF_FAIL("mgr->start() failed: " << err);
auto hdl = app->self.as_actor();
anon_send(hdl, "line 1");
anon_send(hdl, "line 2");
......@@ -219,9 +219,9 @@ CAF_TEST(actor shells can send requests and receive responses) {
auto app_uptr = app_t::make(sys, mpx, worker);
auto app = app_uptr.get();
auto transport = net::stream_transport::make(fd, std::move(app_uptr));
auto mgr = net::socket_manager::make(mpx.get(), fd, std::move(transport));
if (auto err = mgr->init(content(cfg)))
CAF_FAIL("mgr->init() failed: " << err);
auto mgr = net::socket_manager::make(mpx.get(), std::move(transport));
if (auto err = mgr->start(content(cfg)))
CAF_FAIL("mgr->start() failed: " << err);
send(input);
run_while([&] { return app->consumed_bytes != input.size(); });
expect((int32_t), to(worker).with(123));
......
......@@ -26,8 +26,8 @@ public:
settings cfg;
error init(net::web_socket::lower_layer*, const settings& init_cfg) override {
cfg = init_cfg;
error start(net::web_socket::lower_layer*, const settings& args) override {
cfg = args;
return none;
}
......@@ -101,7 +101,7 @@ SCENARIO("the client performs the WebSocket handshake on startup") {
auto& ws_state = *ws;
auto uut = mock_stream_transport::make(std::move(ws));
THEN("the client sends its HTTP request when initializing it") {
CHECK_EQ(uut->init(), error{});
CHECK_EQ(uut->start(), error{});
CHECK_EQ(uut->output_as_str(), http_request);
}
AND("the client waits for the server handshake and validates it") {
......
......@@ -33,8 +33,8 @@ public:
return std::make_unique<app_t>();
}
error init(net::web_socket::lower_layer*, const settings& init_cfg) override {
cfg = init_cfg;
error start(net::web_socket::lower_layer*, const settings& args) override {
cfg = args;
return none;
}
......@@ -69,7 +69,7 @@ struct fixture {
auto ws_ptr = net::web_socket::server::make(std::move(app_ptr));
ws = ws_ptr.get();
transport = mock_stream_transport::make(std::move(ws_ptr));
if (auto err = transport->init())
if (auto err = transport->start())
CAF_FAIL("failed to initialize mock transport: " << err);
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