Commit e31a93fe authored by Joseph Noir's avatar Joseph Noir

Porting middleman_actor behavior to acccept URIs

parent 1be99cf3
...@@ -23,6 +23,8 @@ ...@@ -23,6 +23,8 @@
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/typed_actor.hpp" #include "caf/typed_actor.hpp"
#include "caf/io/uri.hpp"
namespace caf { namespace caf {
namespace io { namespace io {
...@@ -88,18 +90,18 @@ namespace io { ...@@ -88,18 +90,18 @@ namespace io {
/// ~~~ /// ~~~
using middleman_actor = using middleman_actor =
typed_actor< typed_actor<
replies_to<publish_atom, uint16_t, strong_actor_ptr, replies_to<publish_atom, strong_actor_ptr, std::set<std::string>, uri, bool>
std::set<std::string>, std::string, bool>
::with<uint16_t>, ::with<uint16_t>,
replies_to<open_atom, uint16_t, std::string, bool> replies_to<open_atom, uri, bool>
::with<uint16_t>, ::with<uint16_t>,
replies_to<connect_atom, std::string, uint16_t> replies_to<connect_atom, uri>
::with<node_id, strong_actor_ptr, std::set<std::string>>, ::with<node_id, strong_actor_ptr, std::set<std::string>>,
reacts_to<unpublish_atom, actor_addr, uint16_t>, reacts_to<unpublish_atom, actor_addr, uri>,
// TODO: Do these require URIs as well?
reacts_to<close_atom, uint16_t>, reacts_to<close_atom, uint16_t>,
replies_to<spawn_atom, node_id, std::string, message, std::set<std::string>> replies_to<spawn_atom, node_id, std::string, message, std::set<std::string>>
......
...@@ -116,17 +116,19 @@ expected<strong_actor_ptr> middleman::remote_spawn_impl(const node_id& nid, ...@@ -116,17 +116,19 @@ expected<strong_actor_ptr> middleman::remote_spawn_impl(const node_id& nid,
} }
expected<uint16_t> middleman::open(uint16_t port, const char* cstr, bool ru) { expected<uint16_t> middleman::open(uint16_t port, const char* cstr, bool ru) {
std::string str; std::stringstream u_stream;
if (cstr != nullptr) u_stream << default_protocol;
str = cstr; u_stream << (cstr != nullptr ? cstr : "0.0.0.0");
auto f = make_function_view(actor_handle()); u_stream << ":" << port;
return f(open_atom::value, port, std::move(str), ru); auto res = uri::make(u_stream.str());
if (!res)
return sec::invalid_argument;
return open(std::move(*res), ru);
} }
expected<uint16_t> middleman::open(uri u, bool ru) { expected<uint16_t> middleman::open(uri u, bool ru) {
std::string str(u.host().first, u.host().second);
auto f = make_function_view(actor_handle()); auto f = make_function_view(actor_handle());
return f(open_atom::value, u.port_as_int(), std::move(str), ru); return f(open_atom::value, std::move(u), ru);
} }
expected<void> middleman::close(uint16_t port) { expected<void> middleman::close(uint16_t port) {
...@@ -135,18 +137,19 @@ expected<void> middleman::close(uint16_t port) { ...@@ -135,18 +137,19 @@ expected<void> middleman::close(uint16_t port) {
} }
expected<node_id> middleman::connect(std::string host, uint16_t port) { expected<node_id> middleman::connect(std::string host, uint16_t port) {
auto f = make_function_view(actor_handle()); std::stringstream u_stream;
auto res = f(connect_atom::value, std::move(host), port); u_stream << default_protocol;
u_stream << host;
u_stream << ":" << port;
auto res = uri::make(u_stream.str());
if (!res) if (!res)
return std::move(res.error()); return sec::invalid_argument;
return std::get<0>(*res); return connect(*res);
} }
expected<node_id> middleman::connect(uri u) { expected<node_id> middleman::connect(uri u) {
auto f = make_function_view(actor_handle()); auto f = make_function_view(actor_handle());
auto res = f(connect_atom::value, auto res = f(connect_atom::value, std::move(u));
std::string(u.host().first, u.host().second),
u.port_as_int());
if (!res) if (!res)
return std::move(res.error()); return std::move(res.error());
return std::get<0>(*res);} return std::get<0>(*res);}
...@@ -157,10 +160,9 @@ expected<uint16_t> middleman::publish(const strong_actor_ptr& whom, ...@@ -157,10 +160,9 @@ expected<uint16_t> middleman::publish(const strong_actor_ptr& whom,
CAF_LOG_TRACE(CAF_ARG(whom) << CAF_ARG(sigs) << CAF_ARG(port)); CAF_LOG_TRACE(CAF_ARG(whom) << CAF_ARG(sigs) << CAF_ARG(port));
if (!whom) if (!whom)
return sec::cannot_publish_invalid_actor; return sec::cannot_publish_invalid_actor;
std::string in(u.host().first, u.host().second);
auto f = make_function_view(actor_handle()); auto f = make_function_view(actor_handle());
return f(publish_atom::value, u.port_as_int(), std::move(whom), return f(publish_atom::value, std::move(whom),
std::move(sigs), in, ru); std::move(sigs), std::move(u), ru);
} }
expected<uint16_t> middleman::publish_local_groups(uint16_t port, expected<uint16_t> middleman::publish_local_groups(uint16_t port,
...@@ -187,21 +189,19 @@ expected<uint16_t> middleman::publish_local_groups(uint16_t port, ...@@ -187,21 +189,19 @@ expected<uint16_t> middleman::publish_local_groups(uint16_t port,
expected<void> middleman::unpublish(const actor_addr& whom, uri u) { expected<void> middleman::unpublish(const actor_addr& whom, uri u) {
CAF_LOG_TRACE(CAF_ARG(whom) << CAF_ARG(port)); CAF_LOG_TRACE(CAF_ARG(whom) << CAF_ARG(port));
auto f = make_function_view(actor_handle()); auto f = make_function_view(actor_handle());
return f(unpublish_atom::value, whom, u.port_as_int()); return f(unpublish_atom::value, whom, u);
} }
expected<strong_actor_ptr> middleman::remote_actor(std::set<std::string> ifs, expected<strong_actor_ptr> middleman::remote_actor(std::set<std::string> ifs,
uri u) { uri u) {
CAF_LOG_TRACE(CAF_ARG(ifs) << CAF_ARG(u)); CAF_LOG_TRACE(CAF_ARG(ifs) << CAF_ARG(u));
auto f = make_function_view(actor_handle()); auto f = make_function_view(actor_handle());
auto host = std::string(u.host().first, u.host().second); auto res = f(connect_atom::value, std::move(u));
auto port = u.port_as_int();
auto res = f(connect_atom::value, std::move(host), port);
if (!res) if (!res)
return std::move(res.error()); return std::move(res.error());
strong_actor_ptr ptr = std::move(std::get<1>(*res)); strong_actor_ptr ptr = std::move(std::get<1>(*res));
if (!ptr) if (!ptr)
return make_error(sec::no_actor_published_at_port, port); return make_error(sec::no_actor_published_at_port, u.port_as_int());
if (!system().assignable(std::get<2>(*res), ifs)) if (!system().assignable(std::get<2>(*res), ifs))
return make_error(sec::unexpected_actor_messaging_interface, std::move(ifs), return make_error(sec::unexpected_actor_messaging_interface, std::move(ifs),
std::move(std::get<2>(*res))); std::move(std::get<2>(*res)));
......
...@@ -86,20 +86,24 @@ public: ...@@ -86,20 +86,24 @@ public:
behavior_type make_behavior() override { behavior_type make_behavior() override {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
return { return {
[=](publish_atom, uint16_t port, strong_actor_ptr& whom, [=](publish_atom, strong_actor_ptr& whom,
mpi_set& sigs, std::string& addr, bool reuse) -> put_res { mpi_set& sigs, uri& u, bool reuse) -> put_res {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
return put(port, whom, sigs, addr.c_str(), reuse); std::string addr(u.host().first, u.host().second);
return put(u.port_as_int(), whom, sigs, addr.c_str(), reuse);
}, },
[=](open_atom, uint16_t port, std::string& addr, bool reuse) -> put_res { [=](open_atom, uri& u, bool reuse) -> put_res {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
strong_actor_ptr whom; strong_actor_ptr whom;
mpi_set sigs; mpi_set sigs;
return put(port, whom, sigs, addr.c_str(), reuse); std::string addr(u.host().first, u.host().second);
return put(u.port_as_int(), whom, sigs, addr.c_str(), reuse);
}, },
[=](connect_atom, std::string& hostname, uint16_t port) -> get_res { [=](connect_atom, uri& u) -> get_res {
CAF_LOG_TRACE(CAF_ARG(hostname) << CAF_ARG(port)); CAF_LOG_TRACE(CAF_ARG(u));
auto rp = make_response_promise(); auto rp = make_response_promise();
std::string hostname(u.host().first, u.host().second);
auto port = u.port_as_int();
endpoint key{std::move(hostname), port}; endpoint key{std::move(hostname), port};
// respond immediately if endpoint is cached // respond immediately if endpoint is cached
auto x = cached(key); auto x = cached(key);
...@@ -150,9 +154,9 @@ public: ...@@ -150,9 +154,9 @@ public:
); );
return {}; return {};
}, },
[=](unpublish_atom atm, actor_addr addr, uint16_t p) -> del_res { [=](unpublish_atom atm, actor_addr addr, uri& u) -> del_res {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
delegate(broker_, atm, std::move(addr), p); delegate(broker_, atm, std::move(addr), u.port_as_int());
return {}; return {};
}, },
[=](close_atom atm, uint16_t p) -> del_res { [=](close_atom atm, uint16_t p) -> del_res {
......
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