Commit ba1b06a7 authored by Dominik Charousset's avatar Dominik Charousset

Add new resolve function to the middleman

parent 56a953bf
......@@ -50,6 +50,8 @@ public:
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;
......
......@@ -41,6 +41,8 @@ enum class ec : uint8_t {
unimplemented = 10,
app_identifiers_mismatch,
invalid_payload,
invalid_scheme,
invalid_locator,
};
/// @relates ec
......
......@@ -52,7 +52,7 @@ public:
struct event : intrusive::singly_linked<event> {
struct resolve_request {
std::string path;
uri locator;
actor listener;
};
......@@ -61,7 +61,7 @@ public:
uint64_t id;
};
event(std::string path, actor listener);
event(uri locator, actor listener);
event(atom_value type, uint64_t id);
......@@ -138,7 +138,7 @@ public:
// -- event management -------------------------------------------------------
/// Resolves a path to a remote actor.
void resolve(std::string path, actor listener);
void resolve(uri locator, actor listener);
/// Enqueues a message to the endpoint.
void enqueue(mailbox_element_ptr msg, strong_actor_ptr receiver,
......
......@@ -96,8 +96,7 @@ public:
using timeout = endpoint_manager::event::timeout;
using resolve_request = endpoint_manager::event::resolve_request;
if (auto rr = get_if<resolve_request>(&ptr->value)) {
transport_.resolve(*this, std::move(rr->path),
std::move(rr->listener));
transport_.resolve(*this, rr->locator, rr->listener);
} else {
auto& t = get<timeout>(ptr->value);
transport_.timeout(*this, t.type, t.id);
......
......@@ -66,6 +66,11 @@ public:
return result.release();
}
// -- remoting ---------------------------------------------------------------
/// Resolves a path to a remote actor.
void resolve(const uri& locator, const actor& listener);
// -- properties -------------------------------------------------------------
actor_system& system() {
......
......@@ -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;
/// Resolves a path to a remote actor.
virtual void resolve(const uri& locator, const actor& listener) = 0;
// -- properties -------------------------------------------------------------
const std::string& id() const noexcept {
......
......@@ -131,8 +131,8 @@ public:
}
template <class Parent>
void resolve(Parent&, const std::string& path, actor listener) {
worker_.resolve(*this, path, listener);
void resolve(Parent&, const uri& locator, const actor& listener) {
worker_.resolve(*this, locator.path(), listener);
}
template <class... Ts>
......
......@@ -82,7 +82,7 @@ public:
}
template <class Parent>
void resolve(Parent& parent, const std::string& path, actor listener) {
void resolve(Parent& parent, string_view path, const actor& listener) {
auto decorator = make_write_packet_decorator(*this, parent);
application_.resolve(decorator, path, listener);
}
......
......@@ -42,6 +42,7 @@ string_view ec_names[] = {
"unimplemented",
"app_identifiers_mismatch",
"invalid_payload",
"invalid_scheme",
};
} // namespace
......
......@@ -26,8 +26,8 @@
namespace caf {
namespace net {
endpoint_manager::event::event(std::string path, actor listener)
: value(resolve_request{std::move(path), std::move(listener)}) {
endpoint_manager::event::event(uri locator, actor listener)
: value(resolve_request{std::move(locator), std::move(listener)}) {
// nop
}
......@@ -74,9 +74,9 @@ std::unique_ptr<endpoint_manager::message> endpoint_manager::next_message() {
return result;
}
void endpoint_manager::resolve(std::string path, actor listener) {
void endpoint_manager::resolve(uri locator, actor listener) {
using intrusive::inbox_result;
auto ptr = new event(std::move(path), std::move(listener));
auto ptr = new event(std::move(locator), std::move(listener));
switch (events_.push_back(ptr)) {
default:
break;
......
......@@ -21,10 +21,12 @@
#include "caf/expected.hpp"
#include "caf/net/actor_proxy_impl.hpp"
#include "caf/net/basp/application.hpp"
#include "caf/net/basp/ec.hpp"
#include "caf/net/make_endpoint_manager.hpp"
#include "caf/net/middleman.hpp"
#include "caf/net/stream_transport.hpp"
#include "caf/raise_error.hpp"
#include "caf/send.hpp"
namespace caf {
namespace net {
......@@ -50,6 +52,14 @@ endpoint_manager_ptr test::peer(const node_id& id) {
return get_peer(id).second;
}
void test::resolve(const uri& locator, const actor& listener) {
auto id = locator.authority_only();
if (id)
peer(make_node_id(*id))->resolve(locator, listener);
else
anon_send(listener, error(basp::ec::invalid_locator));
}
strong_actor_ptr test::make_proxy(node_id nid, actor_id aid) {
using impl_type = actor_proxy_impl;
using hdl_type = strong_actor_ptr;
......
......@@ -19,9 +19,11 @@
#include "caf/net/middleman.hpp"
#include "caf/actor_system_config.hpp"
#include "caf/net/basp/ec.hpp"
#include "caf/net/middleman_backend.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/raise_error.hpp"
#include "caf/send.hpp"
#include "caf/uri.hpp"
namespace caf {
......@@ -70,6 +72,14 @@ void* middleman::subtype_ptr() {
return this;
}
void middleman::resolve(const uri& locator, const actor& listener) {
auto ptr = backend(locator.scheme());
if (ptr != nullptr)
ptr->resolve(locator, listener);
else
anon_send(listener, error{basp::ec::invalid_scheme});
}
middleman_backend* middleman::backend(string_view scheme) const noexcept {
auto predicate = [&](const middleman_backend_ptr& ptr) {
return ptr->id() == scheme;
......
......@@ -121,7 +121,7 @@ public:
}
template <class Manager>
void resolve(Manager& mgr, std::string path, actor listener) {
void resolve(Manager& mgr, const uri& locator, const actor& listener) {
actor_id aid = 42;
auto hid = "0011223344556677889900112233445566778899";
auto nid = unbox(make_node_id(42, hid));
......@@ -129,6 +129,7 @@ public:
auto p = make_actor<actor_proxy_impl, strong_actor_ptr>(aid, nid,
&mgr.system(), cfg,
&mgr);
std::string path{locator.path().begin(), locator.path().end()};
anon_send(listener, resolve_atom::value, std::move(path), p);
}
......@@ -189,7 +190,7 @@ CAF_TEST(resolve and proxy communication) {
mpx->handle_updates();
run();
CAF_CHECK_EQUAL(read(sockets.second, make_span(read_buf)), hello_test.size());
mgr->resolve("/id/42", self);
mgr->resolve(unbox(make_uri("test:id/42")), self);
run();
self->receive(
[&](resolve_atom, const std::string&, const strong_actor_ptr& p) {
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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.test
#include "caf/net/backend/test.hpp"
#include "caf/test/dsl.hpp"
using namespace caf;
namespace {
struct fixture {
};
} // namespace
CAF_TEST_FIXTURE_SCOPE(test_tests, fixture)
CAF_TEST(todo) {
// implement me
}
CAF_TEST_FIXTURE_SCOPE_END()
......@@ -85,7 +85,7 @@ public:
}
template <class Parent>
void resolve(Parent& parent, const std::string& path, actor listener) {
void resolve(Parent& parent, string_view path, actor listener) {
actor_id aid = 42;
auto hid = "0011223344556677889900112233445566778899";
auto nid = unbox(make_node_id(42, hid));
......@@ -95,7 +95,8 @@ public:
&parent.system(),
cfg,
std::move(ptr));
anon_send(listener, resolve_atom::value, std::move(path), p);
anon_send(listener, resolve_atom::value,
std::string{path.begin(), path.end()}, p);
}
template <class Transport>
......@@ -163,7 +164,7 @@ CAF_TEST(resolve and proxy communication) {
CAF_CHECK_EQUAL(mgr->init(), none);
mpx->handle_updates();
run();
mgr->resolve("/id/42", self);
mgr->resolve(unbox(make_uri("test:/id/42")), self);
run();
self->receive(
[&](resolve_atom, const std::string&, const strong_actor_ptr& p) {
......
......@@ -159,7 +159,7 @@ public:
}
template <class Parent>
void resolve(Parent& parent, const std::string& path, actor listener) {
void resolve(Parent& parent, string_view path, actor listener) {
actor_id aid = 42;
auto hid = "0011223344556677889900112233445566778899";
auto nid = unbox(make_node_id(aid, hid));
......@@ -168,7 +168,8 @@ public:
auto mgr = &parent.manager();
auto p = make_actor<net::actor_proxy_impl, strong_actor_ptr>(aid, nid, sys,
cfg, mgr);
anon_send(listener, resolve_atom::value, std::move(path), p);
anon_send(listener, resolve_atom::value,
std::string{path.begin(), path.end()}, p);
}
template <class Transport>
......@@ -214,7 +215,7 @@ CAF_TEST(receive) {
mpx->handle_updates();
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 3u);
CAF_MESSAGE("resolve actor-proxy");
mgr1->resolve("/id/42", self);
mgr1->resolve(unbox(make_uri("test:/id/42")), self);
run();
self->receive(
[&](resolve_atom, const std::string&, const strong_actor_ptr& p) {
......
......@@ -84,9 +84,9 @@ public:
}
template <class Parent>
void resolve(Parent&, const std::string& path, actor listener) {
res_->resolve_path = path;
res_->resolve_listener = std::move(listener);
void resolve(Parent&, string_view path, const actor& listener) {
res_->resolve_path.assign(path.begin(), path.end());
res_->resolve_listener = listener;
}
template <class Parent>
......
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