Unverified Commit 53c85439 authored by Dominik Charousset's avatar Dominik Charousset Committed by GitHub

Merge pull request #71

Add tcp_backend
parents 14cc8fc2 ea77f761
......@@ -35,7 +35,7 @@ endfunction()
add_library(libcaf_net_obj OBJECT ${CAF_NET_HEADERS}
src/actor_proxy_impl.cpp
src/application.cpp
src/basp/application.cpp
src/basp/connection_state_strings.cpp
src/basp/ec_strings.cpp
src/basp/message_type_strings.cpp
......@@ -51,6 +51,7 @@ add_library(libcaf_net_obj OBJECT ${CAF_NET_HEADERS}
src/message_queue.cpp
src/multiplexer.cpp
src/net/backend/test.cpp
src/net/backend/tcp.cpp
src/net/endpoint_manager_queue.cpp
src/net/middleman.cpp
src/net/middleman_backend.cpp
......@@ -147,4 +148,5 @@ caf_incubator_add_test_suites(caf-net-test
transport_worker_dispatcher
udp_datagram_socket
network_socket
net.backend.tcp
)
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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 <map>
#include <mutex>
#include "caf/detail/net_export.hpp"
#include "caf/error.hpp"
#include "caf/expected.hpp"
#include "caf/net/basp/application.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/make_endpoint_manager.hpp"
#include "caf/net/middleman.hpp"
#include "caf/net/middleman_backend.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/net/stream_transport.hpp"
#include "caf/net/tcp_stream_socket.hpp"
#include "caf/node_id.hpp"
namespace caf::net::backend {
/// Minimal backend for tcp communication.
class CAF_NET_EXPORT tcp : public middleman_backend {
public:
using peer_map = std::map<node_id, endpoint_manager_ptr>;
using emplace_return_type = std::pair<peer_map::iterator, bool>;
// -- constructors, destructors, and assignment operators --------------------
tcp(middleman& mm);
~tcp() override;
// -- interface functions ----------------------------------------------------
error init() override;
void stop() override;
expected<endpoint_manager_ptr> get_or_connect(const uri& locator) override;
endpoint_manager_ptr peer(const node_id& id) override;
void resolve(const uri& locator, const actor& listener) override;
strong_actor_ptr make_proxy(node_id nid, actor_id aid) override;
void set_last_hop(node_id*) override;
// -- properties -------------------------------------------------------------
uint16_t port() const noexcept override;
template <class Handle>
expected<endpoint_manager_ptr>
emplace(const node_id& peer_id, Handle socket_handle) {
using transport_type = stream_transport<basp::application>;
if (auto err = nonblocking(socket_handle, true))
return err;
auto mpx = mm_.mpx();
basp::application app{proxies_};
auto mgr = make_endpoint_manager(
mpx, mm_.system(), transport_type{socket_handle, std::move(app)});
if (auto err = mgr->init()) {
CAF_LOG_ERROR("mgr->init() failed: " << err);
return err;
}
mpx->register_reading(mgr);
emplace_return_type res;
{
const std::lock_guard<std::mutex> lock(lock_);
res = peers_.emplace(peer_id, std::move(mgr));
}
if (res.second)
return res.first->second;
else
return make_error(sec::runtime_error, "peer_id already exists");
}
private:
endpoint_manager_ptr get_peer(const node_id& id);
middleman& mm_;
peer_map peers_;
proxy_registry proxies_;
uint16_t listening_port_;
std::mutex lock_;
};
} // namespace caf::net::backend
......@@ -51,6 +51,8 @@ public:
endpoint_manager_ptr peer(const node_id& id) override;
expected<endpoint_manager_ptr> get_or_connect(const uri& locator) override;
void resolve(const uri& locator, const actor& listener) override;
strong_actor_ptr make_proxy(node_id nid, actor_id aid) override;
......@@ -63,6 +65,8 @@ public:
return get_peer(peer_id).first;
}
uint16_t port() const noexcept override;
peer_entry& emplace(const node_id& peer_id, stream_socket first,
stream_socket second);
......
......@@ -31,6 +31,7 @@
#include "caf/byte.hpp"
#include "caf/callback.hpp"
#include "caf/defaults.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/detail/worker_hub.hpp"
#include "caf/error.hpp"
#include "caf/fwd.hpp"
......@@ -52,7 +53,7 @@
namespace caf::net::basp {
/// An implementation of BASP as an application layer protocol.
class application {
class CAF_NET_EXPORT application {
public:
// -- member types -----------------------------------------------------------
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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 "caf/detail/net_export.hpp"
#include "caf/error.hpp"
#include "caf/net/basp/application.hpp"
#include "caf/proxy_registry.hpp"
namespace caf::net::basp {
/// Factory for basp::application.
/// @relates doorman
class CAF_NET_EXPORT application_factory {
public:
using application_type = basp::application;
application_factory(proxy_registry& proxies) : proxies_(proxies) {
// nop
}
template <class Parent>
error init(Parent&) {
return none;
}
application_type make() const {
return application_type{proxies_};
}
private:
proxy_registry& proxies_;
};
} // namespace caf::net::basp
......@@ -22,6 +22,7 @@
#include <cstdint>
#include "caf/detail/comparable.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/fwd.hpp"
#include "caf/meta/hex_formatted.hpp"
#include "caf/meta/type_name.hpp"
......@@ -33,7 +34,7 @@ namespace caf::net::basp {
/// @addtogroup BASP
/// The header of a Binary Actor System Protocol (BASP) message.
struct header : detail::comparable<header> {
struct CAF_NET_EXPORT header : detail::comparable<header> {
// -- constructors, destructors, and assignment operators --------------------
constexpr header() noexcept
......@@ -74,11 +75,11 @@ struct header : detail::comparable<header> {
/// Serializes a header to a byte representation.
/// @relates header
std::array<byte, header_size> to_bytes(header x);
CAF_NET_EXPORT std::array<byte, header_size> to_bytes(header x);
/// Serializes a header to a byte representation.
/// @relates header
void to_bytes(header x, byte_buffer& buf);
CAF_NET_EXPORT void to_bytes(header x, byte_buffer& buf);
/// @relates header
template <class Inspector>
......
......@@ -24,6 +24,7 @@
#include "caf/byte_buffer.hpp"
#include "caf/config.hpp"
#include "caf/detail/abstract_worker.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/detail/worker_hub.hpp"
#include "caf/fwd.hpp"
#include "caf/net/basp/header.hpp"
......@@ -36,8 +37,8 @@
namespace caf::net::basp {
/// Deserializes payloads for BASP messages asynchronously.
class worker : public detail::abstract_worker,
public remote_message_handler<worker> {
class CAF_NET_EXPORT worker : public detail::abstract_worker,
public remote_message_handler<worker> {
public:
// -- friends ----------------------------------------------------------------
......
......@@ -19,15 +19,21 @@
#pragma once
#include <cstddef>
#include <cstdint>
#include "caf/detail/net_export.hpp"
// -- hard-coded default values for various CAF options ------------------------
namespace caf::defaults::middleman {
/// Maximum number of cached buffers for sending payloads.
extern const size_t max_payload_buffers;
CAF_NET_EXPORT extern const size_t max_payload_buffers;
/// Maximum number of cached buffers for sending headers.
extern const size_t max_header_buffers;
CAF_NET_EXPORT extern const size_t max_header_buffers;
/// Port to listen on for tcp.
CAF_NET_EXPORT extern const uint16_t tcp_port;
} // namespace caf::defaults::middleman
......@@ -21,6 +21,7 @@
#include <string>
#include "caf/actor.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/detail/serialized_size.hpp"
#include "caf/fwd.hpp"
#include "caf/intrusive/drr_queue.hpp"
......@@ -33,7 +34,7 @@
namespace caf::net {
class endpoint_manager_queue {
class CAF_NET_EXPORT endpoint_manager_queue {
public:
enum class element_type { event, message };
......
......@@ -18,6 +18,9 @@
#pragma once
#include <chrono>
#include <set>
#include <string>
#include <thread>
#include "caf/actor_system.hpp"
......@@ -25,6 +28,8 @@
#include "caf/detail/type_list.hpp"
#include "caf/fwd.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/middleman_backend.hpp"
#include "caf/scoped_actor.hpp"
namespace caf::net {
......@@ -72,9 +77,44 @@ public:
// -- remoting ---------------------------------------------------------------
expected<endpoint_manager_ptr> connect(const uri& locator);
// Publishes an actor.
template <class Handle>
void publish(Handle whom, const std::string& path) {
// TODO: Currently, we can't get the interface from the registry. Either we
// change that, or we need to associate the handle with the interface.
system().registry().put(path, whom);
}
/// Resolves a path to a remote actor.
void resolve(const uri& locator, const actor& listener);
template <class Handle = actor, class Duration = std::chrono::seconds>
expected<Handle>
remote_actor(const uri& locator, Duration timeout = std::chrono::seconds(5)) {
scoped_actor self{sys_};
resolve(locator, self);
Handle handle;
error err;
self->receive(
[&handle](strong_actor_ptr& ptr, const std::set<std::string>&) {
// TODO: This cast is not type-safe.
handle = actor_cast<Handle>(std::move(ptr));
},
[&err](const error& e) { err = e; },
after(timeout) >>
[&err] {
err = make_error(sec::runtime_error,
"manager did not respond with a proxy.");
});
if (err)
return err;
if (handle)
return handle;
return make_error(sec::runtime_error, "cast to handle-type failed");
}
// -- properties -------------------------------------------------------------
actor_system& system() {
......@@ -91,6 +131,8 @@ public:
middleman_backend* backend(string_view scheme) const noexcept;
expected<uint16_t> port(string_view scheme) const;
private:
// -- constructors, destructors, and assignment operators --------------------
......
......@@ -46,6 +46,9 @@ public:
/// @returns The endpoint manager for `peer` on success, `nullptr` otherwise.
virtual endpoint_manager_ptr peer(const node_id& id) = 0;
/// Establishes a connection to a remote node.
virtual expected<endpoint_manager_ptr> get_or_connect(const uri& locator) = 0;
/// Resolves a path to a remote actor.
virtual void resolve(const uri& locator, const actor& listener) = 0;
......@@ -57,6 +60,8 @@ public:
return id_;
}
virtual uint16_t port() const = 0;
private:
/// Stores the technology-specific identifier.
std::string id_;
......
......@@ -22,6 +22,7 @@
#include <mutex>
#include <thread>
#include "caf/detail/net_export.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/operation.hpp"
#include "caf/net/pipe_socket.hpp"
......@@ -37,7 +38,8 @@ struct pollfd;
namespace caf::net {
/// Multiplexes any number of ::socket_manager objects with a ::socket.
class multiplexer : public std::enable_shared_from_this<multiplexer> {
class CAF_NET_EXPORT multiplexer
: public std::enable_shared_from_this<multiplexer> {
public:
// -- member types -----------------------------------------------------------
......
......@@ -18,6 +18,7 @@
#pragma once
#include "caf/detail/net_export.hpp"
#include "caf/error.hpp"
#include "caf/fwd.hpp"
#include "caf/net/fwd.hpp"
......@@ -28,7 +29,7 @@
namespace caf::net {
/// Manages the lifetime of a single socket and handles any I/O events on it.
class socket_manager : public ref_counted {
class CAF_NET_EXPORT socket_manager : public ref_counted {
public:
// -- constructors, destructors, and assignment operators --------------------
......
......@@ -29,7 +29,6 @@
#include "caf/detail/network_order.hpp"
#include "caf/detail/parse.hpp"
#include "caf/error.hpp"
#include "caf/expected.hpp"
#include "caf/logger.hpp"
#include "caf/net/basp/constants.hpp"
#include "caf/net/basp/ec.hpp"
......
......@@ -24,4 +24,6 @@ const size_t max_payload_buffers = 100;
const size_t max_header_buffers = 10;
const uint16_t tcp_port = 0;
} // namespace caf::defaults::middleman
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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/backend/tcp.hpp"
#include <mutex>
#include <string>
#include "caf/net/actor_proxy_impl.hpp"
#include "caf/net/basp/application.hpp"
#include "caf/net/basp/application_factory.hpp"
#include "caf/net/basp/ec.hpp"
#include "caf/net/doorman.hpp"
#include "caf/net/ip.hpp"
#include "caf/net/make_endpoint_manager.hpp"
#include "caf/net/middleman.hpp"
#include "caf/net/socket_guard.hpp"
#include "caf/net/stream_transport.hpp"
#include "caf/net/tcp_accept_socket.hpp"
#include "caf/net/tcp_stream_socket.hpp"
#include "caf/send.hpp"
namespace caf::net::backend {
tcp::tcp(middleman& mm)
: middleman_backend("tcp"), mm_(mm), proxies_(mm.system(), *this) {
// nop
}
tcp::~tcp() {
// nop
}
error tcp::init() {
uint16_t conf_port = get_or<uint16_t>(
mm_.system().config(), "middleman.tcp-port", defaults::middleman::tcp_port);
ip_endpoint ep;
auto local_address = std::string("[::]:") + std::to_string(conf_port);
if (auto err = detail::parse(local_address, ep))
return err;
auto acceptor = make_tcp_accept_socket(ep, true);
if (!acceptor)
return acceptor.error();
auto acc_guard = make_socket_guard(*acceptor);
if (auto err = nonblocking(acc_guard.socket(), true))
return err;
auto port = local_port(*acceptor);
if (!port)
return port.error();
listening_port_ = *port;
CAF_LOG_INFO("doorman spawned on " << CAF_ARG(*port));
auto doorman_uri = make_uri("tcp://doorman");
if (!doorman_uri)
return doorman_uri.error();
auto& mpx = mm_.mpx();
auto mgr = make_endpoint_manager(
mpx, mm_.system(),
doorman{acc_guard.release(), basp::application_factory{proxies_}});
if (auto err = mgr->init()) {
CAF_LOG_ERROR("mgr->init() failed: " << err);
return err;
}
return none;
}
void tcp::stop() {
for (const auto& p : peers_)
proxies_.erase(p.first);
peers_.clear();
}
expected<endpoint_manager_ptr> tcp::get_or_connect(const uri& locator) {
if (auto auth = locator.authority_only()) {
auto id = make_node_id(*auth);
if (auto ptr = peer(id))
return ptr;
auto host = locator.authority().host;
if (auto hostname = get_if<std::string>(&host)) {
for (const auto& addr : ip::resolve(*hostname)) {
ip_endpoint ep{addr, locator.authority().port};
auto sock = make_connected_tcp_stream_socket(ep);
if (!sock)
continue;
else
return emplace(id, *sock);
}
}
}
return sec::cannot_connect_to_node;
}
endpoint_manager_ptr tcp::peer(const node_id& id) {
return get_peer(id);
}
void tcp::resolve(const uri& locator, const actor& listener) {
if (auto p = get_or_connect(locator))
(*p)->resolve(locator, listener);
else
anon_send(listener, p.error());
}
strong_actor_ptr tcp::make_proxy(node_id nid, actor_id aid) {
using impl_type = actor_proxy_impl;
using hdl_type = strong_actor_ptr;
actor_config cfg;
return make_actor<impl_type, hdl_type>(aid, nid, &mm_.system(), cfg,
peer(nid));
}
void tcp::set_last_hop(node_id*) {
// nop
}
uint16_t tcp::port() const noexcept {
return listening_port_;
}
endpoint_manager_ptr tcp::get_peer(const node_id& id) {
const std::lock_guard<std::mutex> lock(lock_);
auto i = peers_.find(id);
if (i != peers_.end())
return i->second;
return nullptr;
}
} // namespace caf::net::backend
......@@ -27,6 +27,7 @@
#include "caf/net/multiplexer.hpp"
#include "caf/net/stream_transport.hpp"
#include "caf/raise_error.hpp"
#include "caf/sec.hpp"
#include "caf/send.hpp"
namespace caf::net::backend {
......@@ -54,6 +55,13 @@ endpoint_manager_ptr test::peer(const node_id& id) {
return get_peer(id).second;
}
expected<endpoint_manager_ptr> test::get_or_connect(const uri& locator) {
if (auto ptr = peer(make_node_id(*locator.authority_only())))
return ptr;
return make_error(sec::runtime_error,
"connecting not implemented in test backend");
}
void test::resolve(const uri& locator, const actor& listener) {
auto id = locator.authority_only();
if (id)
......@@ -74,6 +82,10 @@ void test::set_last_hop(node_id*) {
// nop
}
uint16_t test::port() const noexcept {
return 0;
}
test::peer_entry& test::emplace(const node_id& peer_id, stream_socket first,
stream_socket second) {
using transport_type = stream_transport<basp::application>;
......
......@@ -20,11 +20,14 @@
#include "caf/actor_system_config.hpp"
#include "caf/detail/set_thread_name.hpp"
#include "caf/expected.hpp"
#include "caf/init_global_meta_objects.hpp"
#include "caf/net/basp/ec.hpp"
#include "caf/net/endpoint_manager.hpp"
#include "caf/net/middleman_backend.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/raise_error.hpp"
#include "caf/sec.hpp"
#include "caf/send.hpp"
#include "caf/uri.hpp"
......@@ -93,6 +96,13 @@ void* middleman::subtype_ptr() {
return this;
}
expected<endpoint_manager_ptr> middleman::connect(const uri& locator) {
if (auto ptr = backend(locator.scheme()))
return ptr->get_or_connect(locator);
else
return basp::ec::invalid_scheme;
}
void middleman::resolve(const uri& locator, const actor& listener) {
auto ptr = backend(locator.scheme());
if (ptr != nullptr)
......@@ -111,4 +121,11 @@ middleman_backend* middleman::backend(string_view scheme) const noexcept {
return nullptr;
}
expected<uint16_t> middleman::port(string_view scheme) const {
if (auto ptr = backend(scheme))
return ptr->port();
else
return basp::ec::invalid_scheme;
}
} // namespace caf::net
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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. *
******************************************************************************/
#define CAF_SUITE net.backend.tcp
#include "caf/net/backend/tcp.hpp"
#include "caf/net/test/host_fixture.hpp"
#include "caf/test/dsl.hpp"
#include <string>
#include <thread>
#include "caf/actor_system_config.hpp"
#include "caf/ip_endpoint.hpp"
#include "caf/net/middleman.hpp"
#include "caf/net/socket_guard.hpp"
#include "caf/net/tcp_accept_socket.hpp"
#include "caf/uri.hpp"
using namespace caf;
using namespace caf::net;
using namespace std::literals::string_literals;
namespace {
behavior dummy_actor(event_based_actor*) {
return {
// nop
};
}
struct earth_node {
uri operator()() {
return unbox(make_uri("tcp://earth"));
}
};
struct mars_node {
uri operator()() {
return unbox(make_uri("tcp://mars"));
}
};
template <class Node>
struct config : actor_system_config {
config() {
Node this_node;
put(content, "middleman.this-node", this_node());
load<middleman, backend::tcp>();
}
};
class planet_driver {
public:
virtual ~planet_driver() = default;
virtual bool handle_io_event() = 0;
};
template <class Node>
class planet : public test_coordinator_fixture<config<Node>> {
public:
planet(planet_driver& driver)
: mm(this->sys.network_manager()), mpx(mm.mpx()), driver_(driver) {
mpx->set_thread_id();
}
node_id id() const {
return this->sys.node();
}
bool handle_io_event() override {
return driver_.handle_io_event();
}
net::middleman& mm;
multiplexer_ptr mpx;
private:
planet_driver& driver_;
};
struct fixture : host_fixture, planet_driver {
fixture() : earth(*this), mars(*this) {
earth.run();
mars.run();
CAF_REQUIRE_EQUAL(earth.mpx->num_socket_managers(), 2);
CAF_REQUIRE_EQUAL(mars.mpx->num_socket_managers(), 2);
}
bool handle_io_event() override {
return earth.mpx->poll_once(false) || mars.mpx->poll_once(false);
}
void run() {
earth.run();
}
planet<earth_node> earth;
planet<mars_node> mars;
};
} // namespace
CAF_TEST_FIXTURE_SCOPE(tcp_backend_tests, fixture)
CAF_TEST(doorman accept) {
auto backend = earth.mm.backend("tcp");
CAF_CHECK(backend);
uri::authority_type auth;
auth.host = "localhost"s;
auth.port = backend->port();
CAF_MESSAGE("trying to connect to earth at " << CAF_ARG(auth));
auto sock = make_connected_tcp_stream_socket(auth);
CAF_CHECK(sock);
auto guard = make_socket_guard(*sock);
int runs = 0;
while (earth.mpx->num_socket_managers() < 3) {
if (++runs >= 5)
CAF_FAIL("doorman did not create endpoint_manager");
run();
}
CAF_CHECK_EQUAL(earth.mpx->num_socket_managers(), 3);
}
CAF_TEST(connect) {
uri::authority_type auth;
auth.host = "0.0.0.0"s;
auth.port = 0;
auto acceptor = unbox(make_tcp_accept_socket(auth, false));
auto acc_guard = make_socket_guard(acceptor);
auto port = unbox(local_port(acc_guard.socket()));
auto uri_str = std::string("tcp://localhost:") + std::to_string(port);
CAF_MESSAGE("connecting to " << CAF_ARG(uri_str));
CAF_CHECK(earth.mm.connect(*make_uri(uri_str)));
auto sock = unbox(accept(acc_guard.socket()));
auto sock_guard = make_socket_guard(sock);
handle_io_event();
CAF_CHECK_EQUAL(earth.mpx->num_socket_managers(), 3);
}
CAF_TEST(publish) {
auto dummy = earth.sys.spawn(dummy_actor);
auto path = "dummy"s;
CAF_MESSAGE("publishing actor " << CAF_ARG(path));
earth.mm.publish(dummy, path);
CAF_MESSAGE("check registry for " << CAF_ARG(path));
CAF_CHECK_NOT_EQUAL(earth.sys.registry().get(path), nullptr);
}
CAF_TEST(resolve) {
using std::chrono::milliseconds;
using std::chrono::seconds;
auto sockets = unbox(make_stream_socket_pair());
auto earth_be = reinterpret_cast<net::backend::tcp*>(earth.mm.backend("tcp"));
CAF_CHECK(earth_be->emplace(mars.id(), sockets.first));
auto mars_be = reinterpret_cast<net::backend::tcp*>(mars.mm.backend("tcp"));
CAF_CHECK(mars_be->emplace(earth.id(), sockets.second));
handle_io_event();
CAF_CHECK_EQUAL(earth.mpx->num_socket_managers(), 3);
CAF_CHECK_EQUAL(mars.mpx->num_socket_managers(), 3);
auto dummy = earth.sys.spawn(dummy_actor);
earth.mm.publish(dummy, "dummy"s);
auto locator = unbox(make_uri("tcp://earth/name/dummy"s));
CAF_MESSAGE("resolve " << CAF_ARG(locator));
mars.mm.resolve(locator, mars.self);
mars.run();
earth.run();
mars.run();
mars.self->receive(
[](strong_actor_ptr& ptr, const std::set<std::string>&) {
CAF_MESSAGE("resolved actor!");
CAF_CHECK_NOT_EQUAL(ptr, nullptr);
},
[](const error& err) {
CAF_FAIL("got error while resolving: " << CAF_ARG(err));
},
after(std::chrono::seconds(0)) >>
[] { CAF_FAIL("manager did not respond with a proxy."); });
}
CAF_TEST_FIXTURE_SCOPE_END()
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