Commit 7b7870ad authored by Jakob Otto's avatar Jakob Otto

Add test for mm.remote_actor

parent 794358ff
...@@ -53,16 +53,12 @@ public: ...@@ -53,16 +53,12 @@ public:
endpoint_manager_ptr peer(const node_id& id) override; endpoint_manager_ptr peer(const node_id& id) override;
void publish(actor handle, string_view path) override;
void resolve(const uri& locator, const actor& listener) override; void resolve(const uri& locator, const actor& listener) override;
strong_actor_ptr make_proxy(node_id nid, actor_id aid) override; strong_actor_ptr make_proxy(node_id nid, actor_id aid) override;
void set_last_hop(node_id*) override; void set_last_hop(node_id*) override;
void publish();
// -- properties ------------------------------------------------------------- // -- properties -------------------------------------------------------------
tcp_stream_socket socket(const node_id& peer_id) { tcp_stream_socket socket(const node_id& peer_id) {
......
...@@ -53,8 +53,6 @@ public: ...@@ -53,8 +53,6 @@ public:
endpoint_manager_ptr connect(const uri& locator) override; endpoint_manager_ptr connect(const uri& locator) override;
void publish(actor handle, string_view path) override;
void resolve(const uri& locator, const actor& listener) override; void resolve(const uri& locator, const actor& listener) override;
strong_actor_ptr make_proxy(node_id nid, actor_id aid) override; strong_actor_ptr make_proxy(node_id nid, actor_id aid) override;
......
...@@ -80,12 +80,8 @@ public: ...@@ -80,12 +80,8 @@ public:
// Publishes an actor. // Publishes an actor.
template <class Handle = actor> template <class Handle = actor>
error publish(Handle whom, const uri& locator) { void publish(Handle whom, const std::string& path) {
if (auto be = backend(locator.scheme())) system().registry().put(path, whom);
be->publish(whom, locator.path());
else
return sec::runtime_error;
return none;
} }
/// Resolves a path to a remote actor. /// Resolves a path to a remote actor.
...@@ -96,28 +92,22 @@ public: ...@@ -96,28 +92,22 @@ public:
// TODO: Use function view? // TODO: Use function view?
scoped_actor self{sys_}; scoped_actor self{sys_};
resolve(locator, self); resolve(locator, self);
strong_actor_ptr actor_ptr; Handle actor_ptr;
error err = none; error err = none;
self->receive( self->receive(
[&actor_ptr](strong_actor_ptr& ptr, const std::set<std::string>&) { [&actor_ptr](strong_actor_ptr& ptr, const std::set<std::string>&) {
actor_ptr = ptr; actor_ptr = actor_cast<Handle>(std::move(ptr));
}, },
[&err](const error& e) { [&err](const error& e) { err = e; },
err = e; // TODO: how long should the middleman wait before erroring out?
} /*, after(std::chrono::seconds(5)) >>
after(std::chrono::seconds(5)) >> [&err] {
[&err] { err = make_error(sec::runtime_error,
err = make_error(sec::runtime_error, "manager did not respond with a proxy.");
"manager did not respond with a proxy."); });
}*/);
if (err) if (err)
return err; return err;
auto res = actor_cast<Handle>(actor_ptr); return actor_ptr;
if (res)
return res;
else
return make_error(sec::runtime_error,
"cannot cast actor to specified type");
} }
// -- properties ------------------------------------------------------------- // -- properties -------------------------------------------------------------
......
...@@ -49,9 +49,6 @@ public: ...@@ -49,9 +49,6 @@ public:
/// Establishes a connection to a remote node. /// Establishes a connection to a remote node.
virtual endpoint_manager_ptr connect(const uri& locator) = 0; virtual endpoint_manager_ptr connect(const uri& locator) = 0;
/// Publishes an actor.
virtual void publish(actor handle, string_view path) = 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;
......
...@@ -101,10 +101,6 @@ endpoint_manager_ptr tcp::peer(const node_id& id) { ...@@ -101,10 +101,6 @@ endpoint_manager_ptr tcp::peer(const node_id& id) {
return get_peer(id); return get_peer(id);
} }
void tcp::publish(actor handle, string_view path) {
mm_.system().registry().put(std::string(path.begin(), path.end()), handle);
}
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)
......
...@@ -58,10 +58,6 @@ endpoint_manager_ptr test::connect(const uri&) { ...@@ -58,10 +58,6 @@ endpoint_manager_ptr test::connect(const uri&) {
return nullptr; return nullptr;
} }
void test::publish(actor, string_view) {
// nop
}
void test::resolve(const uri& locator, const actor& listener) { void test::resolve(const uri& locator, const actor& listener) {
auto id = locator.authority_only(); auto id = locator.authority_only();
if (id) if (id)
......
...@@ -142,6 +142,11 @@ struct fixture : host_fixture, planet_driver { ...@@ -142,6 +142,11 @@ struct fixture : host_fixture, planet_driver {
return earth.sched.trigger_timeout() || mars.sched.trigger_timeout(); return earth.sched.trigger_timeout() || mars.sched.trigger_timeout();
} }
void set_thread_id() {
earth.mpx->set_thread_id();
mars.mpx->set_thread_id();
}
void run() { void run() {
earth.run(); earth.run();
} }
...@@ -177,50 +182,53 @@ CAF_TEST(connect) { ...@@ -177,50 +182,53 @@ CAF_TEST(connect) {
auto port = unbox(local_port(acc_guard.socket())); auto port = unbox(local_port(acc_guard.socket()));
auto uri_str = std::string("tcp://localhost:") + std::to_string(port); auto uri_str = std::string("tcp://localhost:") + std::to_string(port);
CAF_MESSAGE("connecting to " << CAF_ARG(uri_str)); CAF_MESSAGE("connecting to " << CAF_ARG(uri_str));
unbox(earth.mm.connect(*make_uri(uri_str))); CAF_CHECK(earth.mm.connect(*make_uri(uri_str)));
CAF_CHECK(accept(acc_guard.socket())); auto sock = unbox(accept(acc_guard.socket()));
auto sock_guard = make_socket_guard(sock);
handle_io_event(); handle_io_event();
CAF_REQUIRE_EQUAL(earth.mpx->num_socket_managers(), 3); CAF_REQUIRE_EQUAL(earth.mpx->num_socket_managers(), 3);
} }
/* CAF_TEST(publish) {
CAF_TEST(publish and resolve) { auto dummy = earth.sys.spawn(dummy_actor);
auto dummy = sys.spawn(dummy_actor); auto path = "name/dummy"s;
auto port = unbox(mm.port("tcp")); CAF_MESSAGE("publishing actor " << CAF_ARG(path));
auto locator = unbox(make_uri("tcp://localhost:"s + std::to_string(port))); earth.mm.publish(dummy, path);
mm.publish(dummy, locator); CAF_MESSAGE("check registry for " << CAF_ARG(path));
auto ret = sys.registry().get( CAF_CHECK(earth.sys.registry().get(path) != nullptr);
std::string(locator.path().begin(), locator.path().end()));
if (!ret)
CAF_FAIL("could not retrieve published actor");
config<mars_node> mars_cfg;
actor_system mars_sys{mars_cfg};
auto& mars_mm = mars_sys.network_manager();
auto mars_mpx = mars_mm.mpx();
mars_mpx->set_thread_id();
mars_mpx->poll_once(false);
CAF_REQUIRE(mars_mpx->num_socket_managers(), 2);
CAF_MESSAGE("connecting to " << CAF_ARG(locator));
unbox(mars_mm.connect(locator));
run();
mars_mpx->poll_once(false);
auto locator
= unbox(make_uri("tcp://localhost:"s + std::to_string(port) + "/dummy"));
CAF_MESSAGE("resolving actor " << CAF_ARG(actor_locator));
mars_mm.resolve(locator, self);
mars_mpx->poll_once(false);
run();
mars_mpx->poll_once(false);
CAF_MESSAGE("receiving actor");
self->receive(
[](strong_actor_ptr&, const std::set<std::string>&) {
},
[](const error& err) {
CAF_FAIL("error resolving actor " << CAF_ARG(err));
},
after(std::chrono::seconds(5)) >>
[] { CAF_FAIL("manager did not respond with a proxy."); });
} }
*/
CAF_TEST(remote_actor) {
using std::chrono::milliseconds;
using std::chrono::seconds;
auto dummy = earth.sys.spawn(dummy_actor);
auto path = "name/dummy"s;
earth.mm.publish(dummy, path);
auto port = unbox(earth.mm.port("tcp"));
auto ep_str = "tcp://localhost:"s + std::to_string(port);
auto locator = unbox(make_uri(ep_str));
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 + "/"s + path));
CAF_MESSAGE("resolve " << CAF_ARG(locator));
mars.mm.resolve(locator, mars.self);
bool running = true;
auto f = [&]() {
set_thread_id();
while (running) {
handle_io_event();
std::this_thread::sleep_for(milliseconds(100));
}
};
std::thread t{f};
auto proxy = unbox(mars.mm.remote_actor(locator));
CAF_MESSAGE("resolved actor");
running = false;
t.join();
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