Commit b7e99c51 authored by Joseph Noir's avatar Joseph Noir

Adapt to middleman actor behavior accepting URIs

parent 8c140ab5
...@@ -93,8 +93,7 @@ struct state { ...@@ -93,8 +93,7 @@ struct state {
behavior unconnected(stateful_actor<state>*); behavior unconnected(stateful_actor<state>*);
// prototype definition for transition to `connecting` with Host and Port // prototype definition for transition to `connecting` with Host and Port
void connecting(stateful_actor<state>*, void connecting(stateful_actor<state>*, io::uri);
const std::string& host, uint16_t port);
// prototype definition for transition to `running` with Calculator // prototype definition for transition to `running` with Calculator
behavior running(stateful_actor<state>*, actor calculator); behavior running(stateful_actor<state>*, actor calculator);
...@@ -121,27 +120,28 @@ behavior unconnected(stateful_actor<state>* self) { ...@@ -121,27 +120,28 @@ behavior unconnected(stateful_actor<state>* self) {
self->state.tasks.emplace_back(task{op, x, y}); self->state.tasks.emplace_back(task{op, x, y});
}, },
[=](connect_atom, const std::string& host, uint16_t port) { [=](connect_atom, const std::string& host, uint16_t port) {
connecting(self, host, port); auto u = io::uri::make("tcp://" + host + std::to_string(port));
if (!u)
throw sec::invalid_argument;
connecting(self, std::move(*u));
} }
}; };
} }
void connecting(stateful_actor<state>* self, void connecting(stateful_actor<state>* self, io::uri u) {
const std::string& host, uint16_t port) {
// make sure we are not pointing to an old server // make sure we are not pointing to an old server
self->state.current_server = nullptr; self->state.current_server = nullptr;
// use request().await() to suspend regular behavior until MM responded // use request().await() to suspend regular behavior until MM responded
auto mm = self->system().middleman().actor_handle(); auto mm = self->system().middleman().actor_handle();
self->request(mm, infinite, connect_atom::value, host, port).await( self->request(mm, infinite, connect_atom::value, u).await(
[=](const node_id&, strong_actor_ptr serv, const std::set<std::string>& ifs) { [=](const node_id&, strong_actor_ptr serv, const std::set<std::string>& ifs) {
if (!serv) { if (!serv) {
aout(self) << "*** no server found at \"" << host << "\":" aout(self) << "*** no server found at \"" << u << endl;
<< port << endl;
return; return;
} }
if (!ifs.empty()) { if (!ifs.empty()) {
aout(self) << "*** typed actor found at \"" << host << "\":" aout(self) << "*** typed actor found at \"" << u
<< port << ", but expected an untyped actor "<< endl; << ", but expected an untyped actor "<< endl;
return; return;
} }
aout(self) << "*** successfully connected to server" << endl; aout(self) << "*** successfully connected to server" << endl;
...@@ -151,8 +151,8 @@ void connecting(stateful_actor<state>* self, ...@@ -151,8 +151,8 @@ void connecting(stateful_actor<state>* self,
self->become(running(self, hdl)); self->become(running(self, hdl));
}, },
[=](const error& err) { [=](const error& err) {
aout(self) << "*** cannot connect to \"" << host << "\":" aout(self) << "*** cannot connect to \"" << u
<< port << " => " << self->system().render(err) << endl; << " => " << self->system().render(err) << endl;
self->become(unconnected(self)); self->become(unconnected(self));
} }
); );
...@@ -183,7 +183,10 @@ behavior running(stateful_actor<state>* self, actor calculator) { ...@@ -183,7 +183,10 @@ behavior running(stateful_actor<state>* self, actor calculator) {
send_task(task{op, x, y}); send_task(task{op, x, y});
}, },
[=](connect_atom, const std::string& host, uint16_t port) { [=](connect_atom, const std::string& host, uint16_t port) {
connecting(self, host, port); auto u = io::uri::make("tcp://" + host + std::to_string(port));
if (!u)
throw sec::invalid_argument;
connecting(self, std::move(*u));
} }
}; };
} }
......
...@@ -575,13 +575,15 @@ CAF_TEST(publish_and_connect) { ...@@ -575,13 +575,15 @@ CAF_TEST(publish_and_connect) {
CAF_TEST(remote_actor_and_send) { CAF_TEST(remote_actor_and_send) {
constexpr const char* lo = "localhost"; constexpr const char* lo = "localhost";
constexpr uint16_t port = 4242;
auto u = uri::make("tcp://" + std::string(lo) + ":" + std::to_string(port));
CAF_REQUIRE(u);
CAF_MESSAGE("self: " << to_string(self()->address())); CAF_MESSAGE("self: " << to_string(self()->address()));
mpx()->provide_scribe(lo, 4242, jupiter().connection); mpx()->provide_scribe(lo, 4242, jupiter().connection);
CAF_REQUIRE(mpx()->has_pending_scribe(lo, 4242)); CAF_REQUIRE(mpx()->has_pending_scribe(lo, 4242));
auto mm1 = system.middleman().actor_handle(); auto mm1 = system.middleman().actor_handle();
actor result; actor result;
auto f = self()->request(mm1, infinite, auto f = self()->request(mm1, infinite, connect_atom::value, *u);
connect_atom::value, lo, uint16_t{4242});
// wait until BASP broker has received and processed the connect message // wait until BASP broker has received and processed the connect message
while (!aut()->valid(jupiter().connection)) while (!aut()->valid(jupiter().connection))
mpx()->exec_runnable(); mpx()->exec_runnable();
......
...@@ -73,9 +73,8 @@ struct fixture { ...@@ -73,9 +73,8 @@ struct fixture {
actor result; actor result;
scoped_actor self{system, true}; scoped_actor self{system, true};
auto host = std::string(u.host().first, u.host().second); auto host = std::string(u.host().first, u.host().second);
auto port = u.port_as_int();
self->request(system.middleman().actor_handle(), infinite, self->request(system.middleman().actor_handle(), infinite,
connect_atom::value, host, port).receive( connect_atom::value, u).receive(
[&](node_id&, strong_actor_ptr& res, std::set<std::string>& xs) { [&](node_id&, strong_actor_ptr& res, std::set<std::string>& xs) {
CAF_REQUIRE(xs.empty()); CAF_REQUIRE(xs.empty());
if (res) if (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