Commit 5bc130bc authored by Dominik Charousset's avatar Dominik Charousset

Re-implement hooks, add test for auto connections

parent 854197b1
...@@ -89,8 +89,16 @@ template <class T, typename U> ...@@ -89,8 +89,16 @@ template <class T, typename U>
T actor_cast(const U&); T actor_cast(const U&);
namespace io { namespace io {
class broker;
class middleman; class broker;
class middleman;
namespace basp {
struct header;
} // namespace basp
} // namespace io } // namespace io
namespace scheduler { namespace scheduler {
......
...@@ -29,13 +29,14 @@ ...@@ -29,13 +29,14 @@
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/node_id.hpp" #include "caf/node_id.hpp"
#include "caf/callback.hpp"
#include "caf/actor_addr.hpp" #include "caf/actor_addr.hpp"
#include "caf/serializer.hpp" #include "caf/serializer.hpp"
#include "caf/deserializer.hpp" #include "caf/deserializer.hpp"
#include "caf/abstract_actor.hpp" #include "caf/abstract_actor.hpp"
#include "caf/callback.hpp"
#include "caf/actor_namespace.hpp" #include "caf/actor_namespace.hpp"
#include "caf/io/middleman.hpp"
#include "caf/io/receive_policy.hpp" #include "caf/io/receive_policy.hpp"
#include "caf/io/abstract_broker.hpp" #include "caf/io/abstract_broker.hpp"
#include "caf/io/system_messages.hpp" #include "caf/io/system_messages.hpp"
...@@ -372,7 +373,7 @@ public: ...@@ -372,7 +373,7 @@ public:
/// Provides a callback-based interface for certain BASP events. /// Provides a callback-based interface for certain BASP events.
class callee { class callee {
public: public:
explicit callee(actor_namespace::backend& mgm); explicit callee(actor_namespace::backend& mgm, middleman& mm);
virtual ~callee(); virtual ~callee();
...@@ -411,12 +412,17 @@ public: ...@@ -411,12 +412,17 @@ public:
virtual void learned_new_node_indirectly(const node_id& nid) = 0; virtual void learned_new_node_indirectly(const node_id& nid) = 0;
/// Returns the actor namespace associated to this BASP protocol instance. /// Returns the actor namespace associated to this BASP protocol instance.
actor_namespace& get_namespace() { inline actor_namespace& get_namespace() {
return namespace_; return namespace_;
} }
public: inline middleman& get_middleman() {
return middleman_;
}
protected:
actor_namespace namespace_; actor_namespace namespace_;
middleman& middleman_;
}; };
/// Describes a function object responsible for writing /// Describes a function object responsible for writing
...@@ -530,10 +536,16 @@ public: ...@@ -530,10 +536,16 @@ public:
actor_id aid, actor_id aid,
uint32_t rsn); uint32_t rsn);
const node_id& this_node() const { inline const node_id& this_node() const {
return this_node_; return this_node_;
} }
/// Invokes the callback(s) associated with given event.
template <hook::event_type Event, typename... Ts>
void notify(Ts&&... xs) {
callee_.get_middleman().template notify<Event>(std::forward<Ts>(xs)...);
}
private: private:
routing_table tbl_; routing_table tbl_;
published_actor_map published_actors_; published_actor_map published_actors_;
......
...@@ -20,6 +20,7 @@ ...@@ -20,6 +20,7 @@
#ifndef CAF_IO_HOOK_HPP #ifndef CAF_IO_HOOK_HPP
#define CAF_IO_HOOK_HPP #define CAF_IO_HOOK_HPP
#include <set>
#include <memory> #include <memory>
#include <vector> #include <vector>
...@@ -71,16 +72,17 @@ public: ...@@ -71,16 +72,17 @@ public:
const message& payload); const message& payload);
/// Called whenever a message is forwarded to a different node. /// Called whenever a message is forwarded to a different node.
virtual void message_forwarded_cb(const node_id& from, const node_id& dest, virtual void message_forwarded_cb(const basp::header& hdr,
const std::vector<char>* payload); const std::vector<char>* payload);
/// Called whenever no route for a forwarding request exists. /// Called whenever no route for a forwarding request exists.
virtual void message_forwarding_failed_cb(const node_id& from, virtual void message_forwarding_failed_cb(const basp::header& hdr,
const node_id& to,
const std::vector<char>* payload); const std::vector<char>* payload);
/// Called whenever an actor has been published. /// Called whenever an actor has been published.
virtual void actor_published_cb(const actor_addr& addr, uint16_t port); virtual void actor_published_cb(const actor_addr& addr,
const std::set<std::string>& ifs,
uint16_t port);
/// Called whenever a new remote actor appeared. /// Called whenever a new remote actor appeared.
virtual void new_remote_actor_cb(const actor_addr& addr); virtual void new_remote_actor_cb(const actor_addr& addr);
...@@ -93,6 +95,15 @@ public: ...@@ -93,6 +95,15 @@ public:
/// @param node The newly added entry to the routing table. /// @param node The newly added entry to the routing table.
virtual void new_route_added_cb(const node_id& via, const node_id& node); virtual void new_route_added_cb(const node_id& via, const node_id& node);
/// Called whenever a direct connection was lost.
virtual void connection_lost_cb(const node_id& dest);
/// Called whenever a route became unavailable.
/// @param hop The node that was either disconnected
/// or lost a connection itself.
/// @param dest The node that is no longer reachable via `hop`.
virtual void route_lost_cb(const node_id& hop, const node_id& dest);
/// Called whenever a message was discarded because a remote node /// Called whenever a message was discarded because a remote node
/// tried to send a message to an actor ID that could not be found /// tried to send a message to an actor ID that could not be found
/// in the registry. /// in the registry.
...@@ -115,6 +126,8 @@ public: ...@@ -115,6 +126,8 @@ public:
new_remote_actor, new_remote_actor,
new_connection_established, new_connection_established,
new_route_added, new_route_added,
connection_lost,
route_lost,
invalid_message_received, invalid_message_received,
before_shutdown before_shutdown
}; };
...@@ -150,6 +163,8 @@ private: ...@@ -150,6 +163,8 @@ private:
CAF_IO_HOOK_DISPATCH(new_remote_actor) CAF_IO_HOOK_DISPATCH(new_remote_actor)
CAF_IO_HOOK_DISPATCH(new_connection_established) CAF_IO_HOOK_DISPATCH(new_connection_established)
CAF_IO_HOOK_DISPATCH(new_route_added) CAF_IO_HOOK_DISPATCH(new_route_added)
CAF_IO_HOOK_DISPATCH(connection_lost)
CAF_IO_HOOK_DISPATCH(route_lost)
CAF_IO_HOOK_DISPATCH(invalid_message_received) CAF_IO_HOOK_DISPATCH(invalid_message_received)
CAF_IO_HOOK_DISPATCH(before_shutdown) CAF_IO_HOOK_DISPATCH(before_shutdown)
}; };
......
...@@ -99,6 +99,10 @@ public: ...@@ -99,6 +99,10 @@ public:
}); });
} }
inline bool has_hook() const {
return hooks_ != nullptr;
}
template <class F> template <class F>
void add_shutdown_cb(F fun) { void add_shutdown_cb(F fun) {
struct impl : hook { struct impl : hook {
......
...@@ -246,12 +246,20 @@ void routing_table::erase_direct(const connection_handle& hdl, ...@@ -246,12 +246,20 @@ void routing_table::erase_direct(const connection_handle& hdl,
if (i == direct_by_hdl_.end()) if (i == direct_by_hdl_.end())
return; return;
cb(i->second); cb(i->second);
parent_->parent().notify<hook::connection_lost>(i->second);
direct_by_nid_.erase(i->second); direct_by_nid_.erase(i->second);
direct_by_hdl_.erase(i); direct_by_hdl_.erase(i);
} }
bool routing_table::erase_indirect(const node_id& dest) { bool routing_table::erase_indirect(const node_id& dest) {
return indirect_.erase(dest) > 0; auto i = indirect_.find(dest);
if (i == indirect_.end())
return false;
if (parent_->parent().has_hook())
for (auto& nid : i->second)
parent_->parent().notify<hook::route_lost>(nid, dest);
indirect_.erase(i);
return true;
} }
void routing_table::add_direct(const connection_handle& hdl, void routing_table::add_direct(const connection_handle& hdl,
...@@ -260,6 +268,7 @@ void routing_table::add_direct(const connection_handle& hdl, ...@@ -260,6 +268,7 @@ void routing_table::add_direct(const connection_handle& hdl,
CAF_ASSERT(direct_by_nid_.count(nid) == 0); CAF_ASSERT(direct_by_nid_.count(nid) == 0);
direct_by_hdl_.emplace(hdl, nid); direct_by_hdl_.emplace(hdl, nid);
direct_by_nid_.emplace(nid, hdl); direct_by_nid_.emplace(nid, hdl);
parent_->parent().notify<hook::new_connection_established>(nid);
} }
bool routing_table::add_indirect(const node_id& hop, const node_id& dest) { bool routing_table::add_indirect(const node_id& hop, const node_id& dest) {
...@@ -268,6 +277,7 @@ bool routing_table::add_indirect(const node_id& hop, const node_id& dest) { ...@@ -268,6 +277,7 @@ bool routing_table::add_indirect(const node_id& hop, const node_id& dest) {
auto& hops = indirect_[dest]; auto& hops = indirect_[dest];
auto added_first = hops.empty(); auto added_first = hops.empty();
hops.emplace(hop); hops.emplace(hop);
parent_->parent().notify<hook::new_route_added>(hop, dest);
return added_first; return added_first;
} }
return false; // blacklisted return false; // blacklisted
...@@ -283,14 +293,17 @@ size_t routing_table::erase(const node_id& dest, erase_callback& cb) { ...@@ -283,14 +293,17 @@ size_t routing_table::erase(const node_id& dest, erase_callback& cb) {
auto i = indirect_.find(dest); auto i = indirect_.find(dest);
if (i != indirect_.end()) { if (i != indirect_.end()) {
res = i->second.size(); res = i->second.size();
for (auto& nid : i->second) for (auto& nid : i->second) {
cb(nid); cb(nid);
parent_->parent().notify<hook::route_lost>(nid, dest);
}
indirect_.erase(i); indirect_.erase(i);
} }
auto hdl = lookup_direct(dest); auto hdl = lookup_direct(dest);
if (hdl != invalid_connection_handle) { if (hdl != invalid_connection_handle) {
direct_by_hdl_.erase(hdl); direct_by_hdl_.erase(hdl);
direct_by_nid_.erase(dest); direct_by_nid_.erase(dest);
parent_->parent().notify<hook::connection_lost>(dest);
++res; ++res;
} }
return res; return res;
...@@ -300,7 +313,9 @@ size_t routing_table::erase(const node_id& dest, erase_callback& cb) { ...@@ -300,7 +313,9 @@ size_t routing_table::erase(const node_id& dest, erase_callback& cb) {
* callee * * callee *
******************************************************************************/ ******************************************************************************/
instance::callee::callee(actor_namespace::backend& mgm) : namespace_(mgm) { instance::callee::callee(actor_namespace::backend& mgm, middleman& mm)
: namespace_(mgm),
middleman_(mm) {
// nop // nop
} }
...@@ -358,6 +373,7 @@ connection_state instance::handle(const new_data_msg& dm, header& hdr, ...@@ -358,6 +373,7 @@ connection_state instance::handle(const new_data_msg& dm, header& hdr,
if (payload) if (payload)
bs.write_raw(payload->size(), payload->data()); bs.write_raw(payload->size(), payload->data());
tbl_.flush(*path); tbl_.flush(*path);
notify<hook::message_forwarded>(hdr, payload);
} else { } else {
CAF_LOG_INFO("cannot forward message, no route to destination"); CAF_LOG_INFO("cannot forward message, no route to destination");
if (hdr.source_node != this_node_) { if (hdr.source_node != this_node_) {
...@@ -375,6 +391,7 @@ connection_state instance::handle(const new_data_msg& dm, header& hdr, ...@@ -375,6 +391,7 @@ connection_state instance::handle(const new_data_msg& dm, header& hdr,
} else { } else {
CAF_LOG_WARNING("lost packet with probably spoofed source"); CAF_LOG_WARNING("lost packet with probably spoofed source");
} }
notify<hook::message_forwarding_failed>(hdr, payload);
} }
return await_header; return await_header;
} }
...@@ -385,16 +402,13 @@ connection_state instance::handle(const new_data_msg& dm, header& hdr, ...@@ -385,16 +402,13 @@ connection_state instance::handle(const new_data_msg& dm, header& hdr,
// handle message to ourselves // handle message to ourselves
switch (hdr.operation) { switch (hdr.operation) {
case message_type::server_handshake: { case message_type::server_handshake: {
if (! payload_valid()) {
CAF_LOG_WARNING("received server handshake without payload");
return err();
}
actor_id aid = invalid_actor_id; actor_id aid = invalid_actor_id;
std::set<std::string> sigs; std::set<std::string> sigs;
if (payload_valid()) {
binary_deserializer bd{payload->data(), payload->size(), binary_deserializer bd{payload->data(), payload->size(),
&get_namespace()}; &get_namespace()};
// read payload (ID and interface of published actor)
bd >> aid >> sigs; bd >> aid >> sigs;
}
// close self connection after handshake is done // close self connection after handshake is done
if (hdr.source_node == this_node_) { if (hdr.source_node == this_node_) {
CAF_LOG_INFO("close connection to self immediately"); CAF_LOG_INFO("close connection to self immediately");
...@@ -512,6 +526,7 @@ void instance::add_published_actor(uint16_t port, ...@@ -512,6 +526,7 @@ void instance::add_published_actor(uint16_t port,
auto& entry = published_actors_[port]; auto& entry = published_actors_[port];
swap(entry.first, published_actor); swap(entry.first, published_actor);
swap(entry.second, published_interface); swap(entry.second, published_interface);
notify<hook::actor_published>(entry.first, entry.second, port);
} }
size_t instance::remove_published_actor(uint16_t port, size_t instance::remove_published_actor(uint16_t port,
...@@ -555,11 +570,12 @@ size_t instance::remove_published_actor(const actor_addr& whom, uint16_t port, ...@@ -555,11 +570,12 @@ size_t instance::remove_published_actor(const actor_addr& whom, uint16_t port,
bool instance::dispatch(const actor_addr& sender, const actor_addr& receiver, bool instance::dispatch(const actor_addr& sender, const actor_addr& receiver,
message_id mid, const message& msg) { message_id mid, const message& msg) {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
if (! receiver.is_remote()) CAF_ASSERT(receiver.is_remote());
return false;
auto path = lookup(receiver->node()); auto path = lookup(receiver->node());
if (! path) if (! path) {
notify<hook::message_sending_failed>(sender, receiver, mid, msg);
return false; return false;
}
auto writer = make_callback([&](serializer& sink) { auto writer = make_callback([&](serializer& sink) {
msg.serialize(sink); msg.serialize(sink);
}); });
...@@ -568,6 +584,7 @@ bool instance::dispatch(const actor_addr& sender, const actor_addr& receiver, ...@@ -568,6 +584,7 @@ bool instance::dispatch(const actor_addr& sender, const actor_addr& receiver,
sender ? sender->id() : invalid_actor_id, receiver->id()}; sender ? sender->id() : invalid_actor_id, receiver->id()};
write(path->wr_buf, hdr, &writer); write(path->wr_buf, hdr, &writer);
flush(*path); flush(*path);
notify<hook::message_sent>(sender, path->next_hop, receiver, mid, msg);
return true; return true;
} }
......
...@@ -48,7 +48,8 @@ namespace io { ...@@ -48,7 +48,8 @@ namespace io {
******************************************************************************/ ******************************************************************************/
basp_broker_state::basp_broker_state(broker* selfptr) basp_broker_state::basp_broker_state(broker* selfptr)
: basp::instance::callee(static_cast<actor_namespace::backend&>(*this)), : basp::instance::callee(static_cast<actor_namespace::backend&>(*this),
selfptr->parent()),
self(selfptr), self(selfptr),
instance(selfptr, *this) { instance(selfptr, *this) {
CAF_ASSERT(this_node() != invalid_node_id); CAF_ASSERT(this_node() != invalid_node_id);
...@@ -97,7 +98,7 @@ actor_proxy_ptr basp_broker_state::make_proxy(const node_id& nid, ...@@ -97,7 +98,7 @@ actor_proxy_ptr basp_broker_state::make_proxy(const node_id& nid,
this_node(), nid, this_node(), nid,
invalid_actor_id, aid); invalid_actor_id, aid);
instance.tbl().flush(*path); instance.tbl().flush(*path);
self->parent().notify<hook::new_remote_actor>(res->address()); middleman_.notify<hook::new_remote_actor>(res->address());
return res; return res;
} }
...@@ -226,12 +227,16 @@ void basp_broker_state::deliver(const node_id& source_node, ...@@ -226,12 +227,16 @@ void basp_broker_state::deliver(const node_id& source_node,
} }
if (! dest) { if (! dest) {
CAF_LOG_INFO("cannot deliver message, destination not found"); CAF_LOG_INFO("cannot deliver message, destination not found");
self->parent().notify<hook::invalid_message_received>(source_node, src,
dest_actor, mid, msg);
if (mid.valid() && src != invalid_actor_addr) { if (mid.valid() && src != invalid_actor_addr) {
detail::sync_request_bouncer srb{rsn}; detail::sync_request_bouncer srb{rsn};
srb(src, mid); srb(src, mid);
} }
return; return;
} }
self->parent().notify<hook::message_received>(source_node, src,
dest->address(), mid, msg);
dest->enqueue(src, mid, std::move(msg), nullptr); dest->enqueue(src, mid, std::move(msg), nullptr);
} }
...@@ -321,11 +326,13 @@ void basp_broker_state::learned_new_node_indirectly(const node_id& nid) { ...@@ -321,11 +326,13 @@ void basp_broker_state::learned_new_node_indirectly(const node_id& nid) {
[&](uint16_t port, network::address_listing& addresses) { [&](uint16_t port, network::address_listing& addresses) {
auto& mx = middleman::instance()->backend(); auto& mx = middleman::instance()->backend();
for (auto& kvp : addresses) for (auto& kvp : addresses)
if (kvp.first != network::protocol::ethernet)
for (auto& addr : kvp.second) { for (auto& addr : kvp.second) {
try { try {
auto hdl = mx.new_tcp_scribe(addr, port); auto hdl = mx.new_tcp_scribe(addr, port);
// gotcha! send scribe to our BASP broker // gotcha! send scribe to our BASP broker
// to initiate handshake etc. // to initiate handshake etc.
CAF_LOGF_INFO("connected directly via " << addr);
helper->send(bb, connect_atom::value, hdl, port); helper->send(bb, connect_atom::value, hdl, port);
return; return;
} }
...@@ -511,7 +518,6 @@ behavior basp_broker::make_behavior() { ...@@ -511,7 +518,6 @@ behavior basp_broker::make_behavior() {
} }
detail::singletons::get_actor_registry()->put(whom->id(), whom); detail::singletons::get_actor_registry()->put(whom->id(), whom);
state.instance.add_published_actor(port, whom, std::move(sigs)); state.instance.add_published_actor(port, whom, std::move(sigs));
parent().notify<hook::actor_published>(whom, port);
}, },
// received from middleman actor (delegated) // received from middleman actor (delegated)
[=](connect_atom, connection_handle hdl, uint16_t port) { [=](connect_atom, connection_handle hdl, uint16_t port) {
...@@ -585,6 +591,7 @@ behavior basp_broker::make_behavior() { ...@@ -585,6 +591,7 @@ behavior basp_broker::make_behavior() {
value.apply([&](bool enabled) { value.apply([&](bool enabled) {
if (! enabled) if (! enabled)
return; return;
CAF_LOG_INFO("enable automatic connection");
// open a random port and store a record for others // open a random port and store a record for others
// how to connect to this port in the configuration server // how to connect to this port in the configuration server
auto port = add_tcp_doorman(uint16_t{0}).second; auto port = add_tcp_doorman(uint16_t{0}).second;
......
...@@ -40,9 +40,14 @@ void hook::message_sent_cb(const actor_addr& from, const node_id& dest_node, ...@@ -40,9 +40,14 @@ void hook::message_sent_cb(const actor_addr& from, const node_id& dest_node,
call_next<message_sent>(from, dest_node, dest, mid, payload); call_next<message_sent>(from, dest_node, dest, mid, payload);
} }
void hook::message_forwarded_cb(const node_id& from, const node_id& dest, void hook::message_forwarded_cb(const basp::header& hdr,
const std::vector<char>* payload) { const std::vector<char>* payload) {
call_next<message_forwarded>(from, dest, payload); call_next<message_forwarded>(hdr, payload);
}
void hook::message_forwarding_failed_cb(const basp::header& hdr,
const std::vector<char>* payload) {
call_next<message_forwarding_failed>(hdr, payload);
} }
void hook::message_sending_failed_cb(const actor_addr& from, void hook::message_sending_failed_cb(const actor_addr& from,
...@@ -52,13 +57,10 @@ void hook::message_sending_failed_cb(const actor_addr& from, ...@@ -52,13 +57,10 @@ void hook::message_sending_failed_cb(const actor_addr& from,
call_next<message_sending_failed>(from, dest, mid, payload); call_next<message_sending_failed>(from, dest, mid, payload);
} }
void hook::message_forwarding_failed_cb(const node_id& from, const node_id& to, void hook::actor_published_cb(const actor_addr& addr,
const std::vector<char>* payload) { const std::set<std::string>& ifs,
call_next<message_forwarding_failed>(from, to, payload); uint16_t port) {
} call_next<actor_published>(addr, ifs, port);
void hook::actor_published_cb(const actor_addr& addr, uint16_t port) {
call_next<actor_published>(addr, port);
} }
void hook::new_remote_actor_cb(const actor_addr& addr) { void hook::new_remote_actor_cb(const actor_addr& addr) {
...@@ -73,6 +75,14 @@ void hook::new_route_added_cb(const node_id& via, const node_id& node) { ...@@ -73,6 +75,14 @@ void hook::new_route_added_cb(const node_id& via, const node_id& node) {
call_next<new_route_added>(via, node); call_next<new_route_added>(via, node);
} }
void hook::connection_lost_cb(const node_id& dest) {
call_next<connection_lost>(dest);
}
void hook::route_lost_cb(const node_id& hop, const node_id& dest) {
call_next<route_lost>(hop, dest);
}
void hook::invalid_message_received_cb(const node_id& source, void hook::invalid_message_received_cb(const node_id& source,
const actor_addr& sender, const actor_addr& sender,
actor_id invalid_dest, actor_id invalid_dest,
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2015 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* 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. *
******************************************************************************/
#include "caf/config.hpp"
#define CAF_SUITE io_automatic_connection
#include "caf/test/unit_test.hpp"
#include <set>
#include <thread>
#include <vector>
#include "caf/all.hpp"
#include "caf/io/all.hpp"
#include "caf/experimental/whereis.hpp"
#include "caf/io/network/interfaces.hpp"
#include "caf/io/network/test_multiplexer.hpp"
#include "caf/detail/run_program.hpp"
using namespace caf;
using namespace caf::io;
using namespace caf::experimental;
using std::string;
using ping_atom = atom_constant<atom("ping")>;
using pong_atom = atom_constant<atom("pong")>;
/*
This test checks whether automatic connections work as expected
by first connecting three nodes "in line". In step 2, we send a
message across the line, forcing the nodes to build a mesh. In step 3,
we disconnect the node that originally connected the other two and expect
that the other two nodes communicate uninterrupted.
1) Initial setup:
Earth ---- Mars ---- Jupiter
2) After Jupiter has send a message to Earth:
Earth ---- Mars
\ /
\ /
\ /
Jupiter
3) After Earth has received the message and disconnected Mars:
Earth ---- Jupiter
*/
std::thread run_prog(const char* arg, uint16_t port, bool use_asio) {
return detail::run_program(invalid_actor, caf::test::engine::path(), "-n",
"-s", CAF_XSTR(CAF_SUITE), "--", arg, "-p", port,
(use_asio ? "--use-asio" : ""));
}
// we run the same code on all three nodes, a simple ping-pong client
struct testee_state {
std::set<actor> buddies;
uint16_t port = 0;
const char* name = "testee";
};
behavior testee(stateful_actor<testee_state>* self) {
return {
[self](ping_atom, actor buddy, bool please_broadcast) -> message {
if (please_broadcast)
for (auto& x : self->state.buddies)
if (x != buddy)
send_as(buddy, x, ping_atom::value, buddy, false);
self->state.buddies.emplace(std::move(buddy));
return make_message(pong_atom::value, self);
},
[self](pong_atom, actor buddy) {
self->state.buddies.emplace(std::move(buddy));
},
[self](put_atom, uint16_t new_port) {
self->state.port = new_port;
},
[self](get_atom) {
return self->state.port;
}
};
}
void run_earth(bool use_asio, bool as_server, uint16_t pub_port) {
scoped_actor self;
struct captain : hook {
public:
captain(actor parent) : parent_(std::move(parent)) {
// nop
}
void new_connection_established_cb(const node_id& node) override {
anon_send(parent_, put_atom::value, node);
call_next<hook::new_connection_established>(node);
}
void new_remote_actor_cb(const actor_addr& addr) override {
anon_send(parent_, put_atom::value, addr);
call_next<hook::new_remote_actor>(addr);
}
void connection_lost_cb(const node_id& dest) override {
anon_send(parent_, delete_atom::value, dest);
}
private:
actor parent_;
};
middleman::instance()->add_hook<captain>(self);
auto aut = spawn(testee);
auto port = publish(aut, pub_port);
CAF_TEST_VERBOSE("published testee at port " << port);
std::thread mars_process;
std::thread jupiter_process;
// launch process for Mars
if (! as_server) {
CAF_TEST_VERBOSE("launch process for Mars");
mars_process = run_prog("--mars", port, use_asio);
}
CAF_TEST_VERBOSE("wait for Mars to connect");
node_id mars;
self->receive(
[&](put_atom, const node_id& nid) {
mars = nid;
CAF_TEST_VERBOSE(CAF_TSARG(mars));
}
);
actor_addr mars_addr;
uint16_t mars_port;
self->receive_while([&] { return mars_addr == invalid_actor_addr; })(
[&](put_atom, const actor_addr& addr) {
auto hdl = actor_cast<actor>(addr);
self->sync_send(hdl, sys_atom::value, get_atom::value, "info").await(
[&](ok_atom, const string&, const actor_addr&, const string& name) {
if (name != "testee")
return;
mars_addr = addr;
CAF_TEST_VERBOSE(CAF_TSARG(mars_addr));
self->sync_send(actor_cast<actor>(mars_addr), get_atom::value).await(
[&](uint16_t mp) {
CAF_TEST_VERBOSE("mars published its actor at port " << mp);
mars_port = mp;
}
);
}
);
}
);
// launch process for Jupiter
if (! as_server) {
CAF_TEST_VERBOSE("launch process for Jupiter");
jupiter_process = run_prog("--jupiter", mars_port, use_asio);
}
CAF_TEST_VERBOSE("wait for Jupiter to connect");
self->receive(
[](put_atom, const node_id& jupiter) {
CAF_TEST_VERBOSE(CAF_TSARG(jupiter));
}
);
actor_addr jupiter_addr;
self->receive_while([&] { return jupiter_addr == invalid_actor_addr; })(
[&](put_atom, const actor_addr& addr) {
auto hdl = actor_cast<actor>(addr);
self->sync_send(hdl, sys_atom::value, get_atom::value, "info").await(
[&](ok_atom, const string&, const actor_addr&, const string& name) {
if (name != "testee")
return;
jupiter_addr = addr;
CAF_TEST_VERBOSE(CAF_TSARG(jupiter_addr));
}
);
}
);
CAF_TEST_VERBOSE("shutdown Mars");
anon_send_exit(mars_addr, exit_reason::kill);
if (mars_process.joinable())
mars_process.join();
self->receive(
[&](delete_atom, const node_id& nid) {
CAF_CHECK_EQUAL(nid, mars);
}
);
CAF_TEST_VERBOSE("check whether we still can talk to Jupiter");
self->send(aut, ping_atom::value, self, true);
std::set<actor_addr> found;
int i = 0;
self->receive_for(i, 2)(
[&](pong_atom, const actor&) {
found.emplace(self->current_sender());
}
);
std::set<actor_addr> expected{aut.address(), jupiter_addr};
CAF_CHECK_EQUAL(found, expected);
CAF_TEST_VERBOSE("shutdown Jupiter");
anon_send_exit(jupiter_addr, exit_reason::kill);
if (jupiter_process.joinable())
jupiter_process.join();
anon_send_exit(aut, exit_reason::kill);
}
void run_mars(uint16_t port_to_earth, uint16_t pub_port) {
auto aut = spawn(testee);
auto port = publish(aut, pub_port);
anon_send(aut, put_atom::value, port);
CAF_TEST_VERBOSE("published testee at port " << port);
auto earth = remote_actor("localhost", port_to_earth);
send_as(aut, earth, ping_atom::value, aut, false);
}
void run_jupiter(uint16_t port_to_mars) {
auto aut = spawn(testee);
auto mars = remote_actor("localhost", port_to_mars);
send_as(aut, mars, ping_atom::value, aut, true);
}
CAF_TEST(triangle_setup) {
uint16_t port = 0;
uint16_t publish_port = 0;
auto argv = caf::test::engine::argv();
auto argc = caf::test::engine::argc();
auto r = message_builder(argv, argv + argc).extract_opts({
{"port,p", "port of remote side (when running mars or jupiter)", port},
{"mars", "run mars"},
{"jupiter", "run jupiter"},
{"use-asio", "use ASIO network backend (if available)"},
{"server,s", "run in server mode (don't run clients)", publish_port}
});
// check arguments
bool is_mars = r.opts.count("mars") > 0;
bool is_jupiter = r.opts.count("jupiter") > 0;
bool has_port = r.opts.count("port") > 0;
if (((is_mars || is_jupiter) && ! has_port) || (is_mars && is_jupiter)) {
CAF_TEST_ERROR("need a port when running Mars or Jupiter and cannot "
"both at the same time");
return;
}
auto use_asio = r.opts.count("use-asio") > 0;
# ifdef CAF_USE_ASIO
if (use_asio) {
CAF_MESSAGE("enable ASIO backend");
set_middleman<network::asio_multiplexer>();
}
# endif // CAF_USE_ASIO
// enable automatic connections
anon_send(whereis(atom("ConfigServ")), put_atom::value,
"global.enable-automatic-connections", make_message(true));
auto as_server = r.opts.count("server") > 0;
if (is_mars)
run_mars(port, publish_port);
else if (is_jupiter)
run_jupiter(port);
else
run_earth(use_asio, as_server, publish_port);
await_all_actors_done();
shutdown();
}
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