Commit 83ce0ab8 authored by Jakob Otto's avatar Jakob Otto

Cleanup connect function

parent 5eb4c6a9
...@@ -35,7 +35,7 @@ endfunction() ...@@ -35,7 +35,7 @@ endfunction()
add_library(libcaf_net_obj OBJECT ${CAF_NET_HEADERS} add_library(libcaf_net_obj OBJECT ${CAF_NET_HEADERS}
src/actor_proxy_impl.cpp src/actor_proxy_impl.cpp
src/application.cpp src/basp/application.cpp
src/basp/connection_state_strings.cpp src/basp/connection_state_strings.cpp
src/basp/ec_strings.cpp src/basp/ec_strings.cpp
src/basp/message_type_strings.cpp src/basp/message_type_strings.cpp
......
...@@ -92,11 +92,11 @@ public: ...@@ -92,11 +92,11 @@ public:
// TODO: Use function view? // TODO: Use function view?
scoped_actor self{sys_}; scoped_actor self{sys_};
resolve(locator, self); resolve(locator, self);
Handle actor_ptr; Handle handle;
error err = none; error err;
self->receive( self->receive(
[&actor_ptr](strong_actor_ptr& ptr, const std::set<std::string>&) { [&handle](strong_actor_ptr& ptr, const std::set<std::string>&) {
actor_ptr = actor_cast<Handle>(std::move(ptr)); handle = actor_cast<Handle>(std::move(ptr));
}, },
[&err](const error& e) { err = e; }, [&err](const error& e) { err = e; },
// TODO: how long should the middleman wait before erroring out? // TODO: how long should the middleman wait before erroring out?
...@@ -107,7 +107,9 @@ public: ...@@ -107,7 +107,9 @@ public:
}); });
if (err) if (err)
return err; return err;
return actor_ptr; if (handle)
return handle;
return make_error(sec::runtime_error, "cast to handle-type failed");
} }
// -- properties ------------------------------------------------------------- // -- properties -------------------------------------------------------------
......
...@@ -103,10 +103,19 @@ endpoint_manager_ptr tcp::peer(const node_id& id) { ...@@ -103,10 +103,19 @@ endpoint_manager_ptr tcp::peer(const node_id& id) {
void tcp::resolve(const uri& locator, const actor& listener) { void tcp::resolve(const uri& locator, const actor& listener) {
auto id = locator.authority_only(); auto id = locator.authority_only();
if (id) if (id) {
peer(make_node_id(*id))->resolve(locator, listener); auto p = peer(make_node_id(*id));
else if (p == nullptr) {
CAF_LOG_INFO("connecting to " << CAF_ARG(locator));
p = connect(locator);
}
if (p != nullptr)
p->resolve(locator, listener);
else
anon_send(listener, error(sec::cannot_connect_to_node));
} else {
anon_send(listener, error(basp::ec::invalid_locator)); anon_send(listener, error(basp::ec::invalid_locator));
}
} }
strong_actor_ptr tcp::make_proxy(node_id nid, actor_id aid) { strong_actor_ptr tcp::make_proxy(node_id nid, actor_id aid) {
......
...@@ -101,7 +101,7 @@ expected<endpoint_manager_ptr> middleman::connect(const uri& locator) { ...@@ -101,7 +101,7 @@ expected<endpoint_manager_ptr> middleman::connect(const uri& locator) {
if (auto ret = ptr->connect(locator)) if (auto ret = ptr->connect(locator))
return ret; return ret;
else else
return sec::cannot_connect_to_node; return basp::ec::invalid_scheme;
} }
void middleman::resolve(const uri& locator, const actor& listener) { void middleman::resolve(const uri& locator, const actor& listener) {
...@@ -127,7 +127,7 @@ expected<uint16_t> middleman::port(string_view scheme) const { ...@@ -127,7 +127,7 @@ expected<uint16_t> middleman::port(string_view scheme) const {
if (ptr != nullptr) if (ptr != nullptr)
return ptr->port(); return ptr->port();
else else
return sec::invalid_protocol_family; return basp::ec::invalid_scheme;
} }
} // namespace caf::net } // namespace caf::net
...@@ -81,11 +81,7 @@ public: ...@@ -81,11 +81,7 @@ public:
// nop // nop
} }
virtual bool consume_message() = 0;
virtual bool handle_io_event() = 0; virtual bool handle_io_event() = 0;
virtual bool trigger_timeout() = 0;
}; };
template <class Node> template <class Node>
...@@ -100,31 +96,10 @@ public: ...@@ -100,31 +96,10 @@ public:
return this->sys.node(); return this->sys.node();
} }
bool consume_message() override {
return driver_.consume_message();
}
bool handle_io_event() override { bool handle_io_event() override {
return driver_.handle_io_event(); return driver_.handle_io_event();
} }
bool trigger_timeout() override {
return driver_.trigger_timeout();
}
actor resolve(string_view locator) {
auto hdl = actor_cast<actor>(this->self);
this->sys.network_manager().resolve(unbox(make_uri(locator)), hdl);
this->run();
actor result;
this->self->receive(
[&](strong_actor_ptr& ptr, const std::set<std::string>&) {
CAF_MESSAGE("resolved " << locator);
result = actor_cast<actor>(std::move(ptr));
});
return result;
}
net::middleman& mm; net::middleman& mm;
multiplexer_ptr mpx; multiplexer_ptr mpx;
...@@ -139,18 +114,10 @@ struct fixture : host_fixture, planet_driver { ...@@ -139,18 +114,10 @@ struct fixture : host_fixture, planet_driver {
CAF_REQUIRE_EQUAL(mars.mpx->num_socket_managers(), 2); CAF_REQUIRE_EQUAL(mars.mpx->num_socket_managers(), 2);
} }
bool consume_message() override {
return earth.sched.try_run_once() || mars.sched.try_run_once();
}
bool handle_io_event() override { bool handle_io_event() override {
return earth.mpx->poll_once(false) || mars.mpx->poll_once(false); return earth.mpx->poll_once(false) || mars.mpx->poll_once(false);
} }
bool trigger_timeout() override {
return earth.sched.trigger_timeout() || mars.sched.trigger_timeout();
}
void set_thread_id() { void set_thread_id() {
earth.mpx->set_thread_id(); earth.mpx->set_thread_id();
mars.mpx->set_thread_id(); mars.mpx->set_thread_id();
...@@ -215,15 +182,8 @@ CAF_TEST(remote_actor) { ...@@ -215,15 +182,8 @@ CAF_TEST(remote_actor) {
earth.mm.publish(dummy, name); earth.mm.publish(dummy, name);
auto port = unbox(earth.mm.port("tcp")); auto port = unbox(earth.mm.port("tcp"));
auto ep_str = "tcp://localhost:"s + std::to_string(port); auto ep_str = "tcp://localhost:"s + std::to_string(port);
auto locator = unbox(make_uri(ep_str)); auto locator = unbox(make_uri(ep_str + "/name/"s + name));
CAF_MESSAGE("connecting mars to earth at " << CAF_ARG(locator));
CAF_CHECK(mars.mm.connect(locator));
handle_io_event();
CAF_CHECK_EQUAL(mars.mpx->num_socket_managers(), 3);
CAF_CHECK_EQUAL(earth.mpx->num_socket_managers(), 3);
locator = unbox(make_uri(ep_str + "/name/"s + name));
CAF_MESSAGE("resolve " << CAF_ARG(locator)); CAF_MESSAGE("resolve " << CAF_ARG(locator));
mars.mm.resolve(locator, mars.self);
bool running = true; bool running = true;
auto f = [&]() { auto f = [&]() {
set_thread_id(); set_thread_id();
......
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