Commit e22b2bcc authored by Dominik Charousset's avatar Dominik Charousset

Add missing hooks in BASP broker and publish()

parent da8d52bd
...@@ -84,14 +84,14 @@ class hook { ...@@ -84,14 +84,14 @@ class hook {
* 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 node_id& from, const node_id& dest,
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 node_id& from,
const node_id& to, const node_id& to,
std::vector<char>* payload); const std::vector<char>* payload);
/** /**
* Called whenever an actor has been published. * Called whenever an actor has been published.
......
...@@ -32,15 +32,16 @@ namespace caf { ...@@ -32,15 +32,16 @@ namespace caf {
namespace io { namespace io {
template <class ActorHandle, class SocketAcceptor> template <class ActorHandle, class SocketAcceptor>
void publish_impl(ActorHandle whom, SocketAcceptor fd) { void publish_impl(ActorHandle whom, SocketAcceptor fd, uint16_t port) {
using namespace detail; using namespace detail;
auto mm = middleman::instance(); auto mm = middleman::instance();
// we can't move fd into our lambda in C++11 ... // we can't move fd into our lambda in C++11 ...
using pair_type = std::pair<ActorHandle, SocketAcceptor>; using pair_type = std::pair<ActorHandle, SocketAcceptor>;
auto data = std::make_shared<pair_type>(std::move(whom), std::move(fd)); auto data = std::make_shared<pair_type>(whom, std::move(fd));
mm->run_later([mm, data] { mm->run_later([mm, data, whom, port] {
auto bro = mm->get_named_broker<basp_broker>(atom("_BASP")); auto bro = mm->get_named_broker<basp_broker>(atom("_BASP"));
bro->publish(std::move(data->first), std::move(data->second)); bro->publish(std::move(data->first), std::move(data->second));
mm->notify<hook::actor_published>(whom->address(), port);
}); });
} }
...@@ -48,10 +49,9 @@ inline void publish_impl(abstract_actor_ptr whom, uint16_t port, ...@@ -48,10 +49,9 @@ inline void publish_impl(abstract_actor_ptr whom, uint16_t port,
const char* ipaddr) { const char* ipaddr) {
using namespace detail; using namespace detail;
auto mm = middleman::instance(); auto mm = middleman::instance();
auto addr = whom->address();
network::default_socket_acceptor fd{mm->backend()}; network::default_socket_acceptor fd{mm->backend()};
network::ipv4_bind(fd, port, ipaddr); network::ipv4_bind(fd, port, ipaddr);
publish_impl(std::move(whom), std::move(fd)); publish_impl(std::move(whom), std::move(fd), port);
} }
} // namespace io } // namespace io
......
...@@ -261,9 +261,10 @@ basp_broker::handle_basp_header(connection_context& ctx, ...@@ -261,9 +261,10 @@ basp_broker::handle_basp_header(connection_context& ctx,
if (hdr.dest_node != invalid_node_id && hdr.dest_node != node()) { if (hdr.dest_node != invalid_node_id && hdr.dest_node != node()) {
auto route = get_route(hdr.dest_node); auto route = get_route(hdr.dest_node);
if (route.invalid()) { if (route.invalid()) {
// TODO: signalize that we don't have route to given node
CAF_LOG_INFO("message dropped: no route to node " CAF_LOG_INFO("message dropped: no route to node "
<< to_string(hdr.dest_node)); << to_string(hdr.dest_node));
parent().notify<hook::message_forwarding_failed>(hdr.source_node,
hdr.dest_node, payload);
return close_connection; return close_connection;
} }
auto& buf = wr_buf(route.hdl); auto& buf = wr_buf(route.hdl);
...@@ -271,6 +272,8 @@ basp_broker::handle_basp_header(connection_context& ctx, ...@@ -271,6 +272,8 @@ basp_broker::handle_basp_header(connection_context& ctx,
write(bs, hdr); write(bs, hdr);
if (payload) buf.insert(buf.end(), payload->begin(), payload->end()); if (payload) buf.insert(buf.end(), payload->begin(), payload->end());
flush(route.hdl); flush(route.hdl);
parent().notify<hook::message_forwarded>(hdr.source_node,
hdr.dest_node, payload);
return await_header; return await_header;
} }
// handle a message that is addressed to us // handle a message that is addressed to us
...@@ -334,6 +337,7 @@ basp_broker::handle_basp_header(connection_context& ctx, ...@@ -334,6 +337,7 @@ basp_broker::handle_basp_header(connection_context& ctx,
CAF_LOG_INFO("multiple incoming connections from the same node"); CAF_LOG_INFO("multiple incoming connections from the same node");
return close_connection; return close_connection;
} }
parent().notify<hook::new_connection_established>(ctx.remote_id );
break; break;
} }
case basp::server_handshake: { case basp::server_handshake: {
...@@ -427,6 +431,7 @@ basp_broker::handle_basp_header(connection_context& ctx, ...@@ -427,6 +431,7 @@ basp_broker::handle_basp_header(connection_context& ctx,
ctx.published_actor = proxy; ctx.published_actor = proxy;
ctx.handshake_data->result->set_value(std::move(proxy)); ctx.handshake_data->result->set_value(std::move(proxy));
ctx.handshake_data = nullptr; ctx.handshake_data = nullptr;
parent().notify<hook::new_connection_established>(nid);
break; break;
} }
} }
...@@ -516,6 +521,7 @@ void basp_broker::erase_proxy(const id_type& nid, actor_id aid) { ...@@ -516,6 +521,7 @@ void basp_broker::erase_proxy(const id_type& nid, actor_id aid) {
void basp_broker::add_route(const id_type& nid, connection_handle hdl) { void basp_broker::add_route(const id_type& nid, connection_handle hdl) {
if (m_blacklist.count(std::make_pair(nid, hdl)) == 0) { if (m_blacklist.count(std::make_pair(nid, hdl)) == 0) {
parent().notify<hook::new_route_added>(m_current_context->remote_id, nid);
m_routes[nid].second.insert({hdl, nid}); m_routes[nid].second.insert({hdl, nid});
} }
} }
......
...@@ -41,7 +41,7 @@ void hook::message_sent_cb(const actor_addr& from, const node_id& dest_node, ...@@ -41,7 +41,7 @@ void hook::message_sent_cb(const actor_addr& from, const node_id& dest_node,
} }
void hook::message_forwarded_cb(const node_id& from, const node_id& dest, void hook::message_forwarded_cb(const node_id& from, const node_id& dest,
std::vector<char>* payload) { const std::vector<char>* payload) {
call_next<message_forwarded>(from, dest, payload); call_next<message_forwarded>(from, dest, payload);
} }
...@@ -53,7 +53,7 @@ void hook::message_sending_failed_cb(const actor_addr& from, ...@@ -53,7 +53,7 @@ void hook::message_sending_failed_cb(const actor_addr& from,
} }
void hook::message_forwarding_failed_cb(const node_id& from, const node_id& to, void hook::message_forwarding_failed_cb(const node_id& from, const node_id& to,
std::vector<char>* payload) { const std::vector<char>* payload) {
call_next<message_forwarding_failed>(from, to, payload); call_next<message_forwarding_failed>(from, to, payload);
} }
......
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