Commit bcd2ee3e authored by Jakob Otto's avatar Jakob Otto

Make emplace function public

parent 666d1c5c
...@@ -21,6 +21,8 @@ ...@@ -21,6 +21,8 @@
#include <map> #include <map>
#include "caf/detail/net_export.hpp" #include "caf/detail/net_export.hpp"
#include "caf/error.hpp"
#include "caf/expected.hpp"
#include "caf/net/basp/application.hpp" #include "caf/net/basp/application.hpp"
#include "caf/net/fwd.hpp" #include "caf/net/fwd.hpp"
#include "caf/net/make_endpoint_manager.hpp" #include "caf/net/make_endpoint_manager.hpp"
...@@ -49,7 +51,7 @@ public: ...@@ -49,7 +51,7 @@ public:
void stop() override; 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; endpoint_manager_ptr peer(const node_id& id) override;
...@@ -69,6 +71,24 @@ public: ...@@ -69,6 +71,24 @@ public:
return listening_port_; 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: private:
class basp_application_factory { class basp_application_factory {
public: public:
...@@ -89,23 +109,6 @@ private: ...@@ -89,23 +109,6 @@ private:
proxy_registry& proxies_; 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); endpoint_manager_ptr get_peer(const node_id& id);
middleman& mm_; middleman& mm_;
......
...@@ -51,7 +51,7 @@ public: ...@@ -51,7 +51,7 @@ public:
endpoint_manager_ptr peer(const node_id& id) override; 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; void resolve(const uri& locator, const actor& listener) override;
......
...@@ -47,7 +47,7 @@ public: ...@@ -47,7 +47,7 @@ public:
virtual endpoint_manager_ptr peer(const node_id& id) = 0; virtual endpoint_manager_ptr peer(const node_id& id) = 0;
/// Establishes a connection to a remote node. /// 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. /// Resolves a path to a remote actor.
virtual void resolve(const uri& locator, const actor& listener) = 0; virtual void resolve(const uri& locator, const actor& listener) = 0;
......
...@@ -51,7 +51,7 @@ error tcp::init() { ...@@ -51,7 +51,7 @@ error tcp::init() {
auto local_address = std::string("[::]:") + std::to_string(conf_port); auto local_address = std::string("[::]:") + std::to_string(conf_port);
if (auto err = detail::parse(local_address, ep)) if (auto err = detail::parse(local_address, ep))
return err; return err;
auto acceptor = make_tcp_accept_socket(ep); auto acceptor = make_tcp_accept_socket(ep, true);
if (!acceptor) if (!acceptor)
return acceptor.error(); return acceptor.error();
auto acc_guard = make_socket_guard(*acceptor); auto acc_guard = make_socket_guard(*acceptor);
...@@ -81,7 +81,7 @@ void tcp::stop() { ...@@ -81,7 +81,7 @@ void tcp::stop() {
peers_.clear(); peers_.clear();
} }
endpoint_manager_ptr tcp::connect(const uri& locator) { expected<endpoint_manager_ptr> tcp::connect(const uri& locator) {
auto auth = locator.authority(); auto auth = locator.authority();
auto host = auth.host; auto host = auth.host;
if (auto hostname = get_if<std::string>(&host)) { if (auto hostname = get_if<std::string>(&host)) {
...@@ -94,7 +94,7 @@ endpoint_manager_ptr tcp::connect(const uri& locator) { ...@@ -94,7 +94,7 @@ endpoint_manager_ptr tcp::connect(const uri& locator) {
return emplace(make_node_id(*locator.authority_only()), *sock); 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) { endpoint_manager_ptr tcp::peer(const node_id& id) {
...@@ -107,12 +107,13 @@ void tcp::resolve(const uri& locator, const actor& listener) { ...@@ -107,12 +107,13 @@ void tcp::resolve(const uri& locator, const actor& listener) {
auto p = peer(make_node_id(*id)); auto p = peer(make_node_id(*id));
if (p == nullptr) { if (p == nullptr) {
CAF_LOG_INFO("connecting to " << CAF_ARG(locator)); 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);
p->resolve(locator, listener);
else
anon_send(listener, error(sec::cannot_connect_to_node));
} else { } else {
anon_send(listener, error(basp::ec::invalid_locator)); anon_send(listener, error(basp::ec::invalid_locator));
} }
......
...@@ -27,6 +27,7 @@ ...@@ -27,6 +27,7 @@
#include "caf/net/multiplexer.hpp" #include "caf/net/multiplexer.hpp"
#include "caf/net/stream_transport.hpp" #include "caf/net/stream_transport.hpp"
#include "caf/raise_error.hpp" #include "caf/raise_error.hpp"
#include "caf/sec.hpp"
#include "caf/send.hpp" #include "caf/send.hpp"
namespace caf::net::backend { namespace caf::net::backend {
...@@ -54,8 +55,9 @@ endpoint_manager_ptr test::peer(const node_id& id) { ...@@ -54,8 +55,9 @@ endpoint_manager_ptr test::peer(const node_id& id) {
return get_peer(id).second; return get_peer(id).second;
} }
endpoint_manager_ptr test::connect(const uri&) { expected<endpoint_manager_ptr> test::connect(const uri&) {
return nullptr; return make_error(sec::runtime_error,
"function not implemented in test_backend");
} }
void test::resolve(const uri& locator, const actor& listener) { void test::resolve(const uri& locator, const actor& listener) {
......
...@@ -165,20 +165,20 @@ CAF_TEST(publish) { ...@@ -165,20 +165,20 @@ CAF_TEST(publish) {
CAF_CHECK_NOT_EQUAL(earth.sys.registry().get(path), nullptr); 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) { CAF_TEST(remote_actor) {
using std::chrono::milliseconds; using std::chrono::milliseconds;
using std::chrono::seconds; 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 dummy = earth.sys.spawn(dummy_actor);
auto name = "dummy"s; earth.mm.publish(dummy, "dummy"s);
earth.mm.publish(dummy, name); auto locator = unbox(make_uri("tcp://earth/name/dummy"s));
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));
CAF_MESSAGE("resolve " << CAF_ARG(locator)); CAF_MESSAGE("resolve " << CAF_ARG(locator));
bool running = true; bool running = true;
auto f = [&]() { auto f = [&]() {
...@@ -195,6 +195,6 @@ CAF_TEST(remote_actor) { ...@@ -195,6 +195,6 @@ CAF_TEST(remote_actor) {
running = false; running = false;
t.join(); t.join();
set_thread_id(); set_thread_id();
}*/ }
CAF_TEST_FIXTURE_SCOPE_END() 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