Commit a608f8c0 authored by Jakob Otto's avatar Jakob Otto

Make emplace function public

parent c9b52f01
......@@ -21,6 +21,8 @@
#include <map>
#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"
......@@ -49,7 +51,7 @@ public:
void stop() override;
endpoint_manager_ptr connect(const uri& locator) override;
expected<endpoint_manager_ptr> connect(const uri& locator) override;
endpoint_manager_ptr peer(const node_id& id) override;
......@@ -69,6 +71,24 @@ public:
return listening_port_;
}
template <class Handle>
expected<endpoint_manager_ptr>
emplace(const node_id& peer_id, Handle socket_handle) {
using transport_type = stream_transport<basp::application>;
nonblocking(socket_handle, true);
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);
peers_.emplace(peer_id, std::move(mgr));
return peers_[peer_id];
}
private:
class basp_application_factory {
public:
......@@ -89,23 +109,6 @@ private:
proxy_registry& proxies_;
};
template <class Handle>
endpoint_manager_ptr& emplace(const node_id& peer_id, Handle socket_handle) {
using transport_type = stream_transport<basp::application>;
nonblocking(socket_handle, true);
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);
CAF_RAISE_ERROR("mgr->init() failed");
}
mpx->register_reading(mgr);
peers_.emplace(peer_id, std::move(mgr));
return peers_[peer_id];
}
endpoint_manager_ptr get_peer(const node_id& id);
middleman& mm_;
......
......@@ -51,7 +51,7 @@ public:
endpoint_manager_ptr peer(const node_id& id) override;
endpoint_manager_ptr connect(const uri& locator) override;
expected<endpoint_manager_ptr> connect(const uri& locator) override;
void resolve(const uri& locator, const actor& listener) override;
......
......@@ -47,7 +47,7 @@ public:
virtual endpoint_manager_ptr peer(const node_id& id) = 0;
/// Establishes a connection to a remote node.
virtual endpoint_manager_ptr connect(const uri& locator) = 0;
virtual expected<endpoint_manager_ptr> connect(const uri& locator) = 0;
/// Resolves a path to a remote actor.
virtual void resolve(const uri& locator, const actor& listener) = 0;
......
......@@ -51,7 +51,7 @@ error tcp::init() {
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);
auto acceptor = make_tcp_accept_socket(ep, true);
if (!acceptor)
return acceptor.error();
auto acc_guard = make_socket_guard(*acceptor);
......@@ -81,7 +81,7 @@ void tcp::stop() {
peers_.clear();
}
endpoint_manager_ptr tcp::connect(const uri& locator) {
expected<endpoint_manager_ptr> tcp::connect(const uri& locator) {
auto auth = locator.authority();
auto host = auth.host;
if (auto hostname = get_if<std::string>(&host)) {
......@@ -94,7 +94,7 @@ endpoint_manager_ptr tcp::connect(const uri& locator) {
return emplace(make_node_id(*locator.authority_only()), *sock);
}
}
return nullptr;
return sec::cannot_connect_to_node;
}
endpoint_manager_ptr tcp::peer(const node_id& id) {
......@@ -107,12 +107,13 @@ void tcp::resolve(const uri& locator, const actor& listener) {
auto p = peer(make_node_id(*id));
if (p == nullptr) {
CAF_LOG_INFO("connecting to " << CAF_ARG(locator));
p = connect(locator);
auto res = connect(locator);
if (!res)
anon_send(listener, error(sec::cannot_connect_to_node));
else
p = *res;
}
if (p != nullptr)
p->resolve(locator, listener);
else
anon_send(listener, error(sec::cannot_connect_to_node));
p->resolve(locator, listener);
} else {
anon_send(listener, error(basp::ec::invalid_locator));
}
......
......@@ -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,8 +55,9 @@ endpoint_manager_ptr test::peer(const node_id& id) {
return get_peer(id).second;
}
endpoint_manager_ptr test::connect(const uri&) {
return nullptr;
expected<endpoint_manager_ptr> test::connect(const uri&) {
return make_error(sec::runtime_error,
"function not implemented in test_backend");
}
void test::resolve(const uri& locator, const actor& listener) {
......
......@@ -165,20 +165,20 @@ CAF_TEST(publish) {
CAF_CHECK_NOT_EQUAL(earth.sys.registry().get(path), nullptr);
}
/*
// TODO: `middleman.this-node` cannot be set according to randomly bound port of
the node. Thus, the test segfaults when creating the new actor with
`nullptr` for the `endpoint_manager`...
CAF_TEST(remote_actor) {
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"));
earth_be->emplace(mars.id(), sockets.first);
auto mars_be = reinterpret_cast<net::backend::tcp*>(mars.mm.backend("tcp"));
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);
auto name = "dummy"s;
earth.mm.publish(dummy, name);
auto port = unbox(earth.mm.port("tcp"));
auto ep_str = "tcp://localhost:"s + std::to_string(port);
auto locator = unbox(make_uri(ep_str + "/name/"s + name));
earth.mm.publish(dummy, "dummy"s);
auto locator = unbox(make_uri("tcp://earth/name/dummy"s));
CAF_MESSAGE("resolve " << CAF_ARG(locator));
bool running = true;
auto f = [&]() {
......@@ -195,6 +195,6 @@ CAF_TEST(remote_actor) {
running = false;
t.join();
set_thread_id();
}*/
}
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