Commit 98d522e2 authored by Dominik Charousset's avatar Dominik Charousset

Add new middleman class

parent 39ca8e49
...@@ -19,6 +19,7 @@ set(LIBCAF_NET_SRCS ...@@ -19,6 +19,7 @@ set(LIBCAF_NET_SRCS
src/ip.cpp src/ip.cpp
src/message_type.cpp src/message_type.cpp
src/multiplexer.cpp src/multiplexer.cpp
src/net/middleman.cpp
src/network_socket.cpp src/network_socket.cpp
src/pipe_socket.cpp src/pipe_socket.cpp
src/pollset_updater.cpp src/pollset_updater.cpp
...@@ -27,7 +28,6 @@ set(LIBCAF_NET_SRCS ...@@ -27,7 +28,6 @@ set(LIBCAF_NET_SRCS
src/stream_socket.cpp src/stream_socket.cpp
src/tcp_accept_socket.cpp src/tcp_accept_socket.cpp
src/tcp_stream_socket.cpp src/tcp_stream_socket.cpp
src/tcp_stream_socket.cpp
src/udp_datagram_socket.cpp src/udp_datagram_socket.cpp
) )
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#pragma once
#include <thread>
#include "caf/actor_system.hpp"
#include "caf/net/fwd.hpp"
namespace caf {
namespace net {
class middleman : public actor_system::module {
public:
// -- constructors, destructors, and assignment operators --------------------
~middleman() override;
// -- interface functions ----------------------------------------------------
void start() override;
void stop() override;
void init(actor_system_config&) override;
id_t id() const override;
void* subtype_ptr() override;
// -- factory functions ------------------------------------------------------
static actor_system::module* make(actor_system&, detail::type_list<>);
// -- properties -------------------------------------------------------------
const actor_system_config& config() const noexcept {
return sys_.config();
}
const multiplexer_ptr& mpx() const noexcept {
return mpx_;
}
private:
// -- constructors, destructors, and assignment operators --------------------
explicit middleman(actor_system& sys);
// -- member variables -------------------------------------------------------
/// Points to the parent system.
actor_system& sys_;
/// Stores the global socket I/O multiplexer.
multiplexer_ptr mpx_;
/// Runs the multiplexer's event loop
std::thread mpx_thread_;
};
} // namespace net
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/net/middleman.hpp"
#include "caf/actor_system_config.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/raise_error.hpp"
#include "caf/uri.hpp"
namespace caf {
namespace net {
middleman::middleman(actor_system& sys) : sys_(sys) {
mpx_ = std::make_shared<multiplexer>();
}
middleman::~middleman() {
// nop
}
void middleman::start() {
if (!get_or(config(), "middleman.manual-multiplexing", false)) {
auto mpx = mpx_;
mpx_thread_ = std::thread{[mpx] { mpx->run(); }};
}
}
void middleman::stop() {
mpx_->close_pipe();
if (mpx_thread_.joinable())
mpx_thread_.join();
}
void middleman::init(actor_system_config&cfg) {
if (auto err = mpx_->init())
CAF_RAISE_ERROR("mpx->init failed");
if (auto node_uri = get_if<uri>(&cfg, "middleman.this-node")) {
auto this_node = make_node_id(std::move(*node_uri));
sys_.node_.swap(this_node);
} else {
CAF_RAISE_ERROR("no valid entry for middleman.this-node found");
}
}
actor_system::module::id_t middleman::id() const {
return module::network_manager;
}
void* middleman::subtype_ptr() {
return this;
}
actor_system::module* middleman::make(actor_system& sys, detail::type_list<>) {
return new middleman(sys);
}
} // namespace net
} // namespace caf
...@@ -33,6 +33,7 @@ ...@@ -33,6 +33,7 @@
#include "caf/net/basp/constants.hpp" #include "caf/net/basp/constants.hpp"
#include "caf/net/basp/ec.hpp" #include "caf/net/basp/ec.hpp"
#include "caf/net/make_endpoint_manager.hpp" #include "caf/net/make_endpoint_manager.hpp"
#include "caf/net/middleman.hpp"
#include "caf/net/multiplexer.hpp" #include "caf/net/multiplexer.hpp"
#include "caf/net/stream_socket.hpp" #include "caf/net/stream_socket.hpp"
#include "caf/net/stream_transport.hpp" #include "caf/net/stream_transport.hpp"
...@@ -58,26 +59,31 @@ size_t fetch_size(variant<size_t, sec> x) { ...@@ -58,26 +59,31 @@ size_t fetch_size(variant<size_t, sec> x) {
return get<size_t>(x); return get<size_t>(x);
} }
struct fixture : test_coordinator_fixture<>, struct config : actor_system_config {
config() {
put(content, "middleman.this-node", unbox(make_uri("test:earth")));
load<middleman>();
}
};
struct fixture : test_coordinator_fixture<config>,
host_fixture, host_fixture,
proxy_registry::backend { proxy_registry::backend {
fixture() { fixture() {
uri mars_uri; uri mars_uri;
REQUIRE_OK(parse("tcp://mars", mars_uri)); REQUIRE_OK(parse("test:mars", mars_uri));
mars = make_node_id(mars_uri); mars = make_node_id(mars_uri);
mpx = std::make_shared<multiplexer>();
if (auto err = mpx->init())
CAF_FAIL("mpx->init failed: " << sys.render(err));
auto proxies = std::make_shared<proxy_registry>(sys, *this); auto proxies = std::make_shared<proxy_registry>(sys, *this);
auto sockets = unbox(make_stream_socket_pair()); auto sockets = unbox(make_stream_socket_pair());
sock = sockets.first; sock = sockets.first;
nonblocking(sockets.first, true); nonblocking(sockets.first, true);
nonblocking(sockets.second, true); nonblocking(sockets.second, true);
auto mpx = sys.network_manager().mpx();
mgr = make_endpoint_manager(mpx, sys, mgr = make_endpoint_manager(mpx, sys,
transport_type{sockets.second, transport_type{sockets.second,
basp::application{proxies}}); basp::application{proxies}});
REQUIRE_OK(mgr->init()); REQUIRE_OK(mgr->init());
mpx->handle_updates(); handle_io_event();
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 2u); CAF_CHECK_EQUAL(mpx->num_socket_managers(), 2u);
auto& dref = dynamic_cast<endpoint_manager_impl<transport_type>&>(*mgr); auto& dref = dynamic_cast<endpoint_manager_impl<transport_type>&>(*mgr);
app = &dref.application(); app = &dref.application();
...@@ -88,6 +94,7 @@ struct fixture : test_coordinator_fixture<>, ...@@ -88,6 +94,7 @@ struct fixture : test_coordinator_fixture<>,
} }
bool handle_io_event() override { bool handle_io_event() override {
auto mpx = sys.network_manager().mpx();
mpx->handle_updates(); mpx->handle_updates();
return mpx->poll_once(false); return mpx->poll_once(false);
} }
...@@ -157,8 +164,6 @@ struct fixture : test_coordinator_fixture<>, ...@@ -157,8 +164,6 @@ struct fixture : test_coordinator_fixture<>,
node_id mars; node_id mars;
multiplexer_ptr mpx;
endpoint_manager_ptr mgr; endpoint_manager_ptr mgr;
stream_socket sock; stream_socket sock;
......
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