Commit 6e8ef53f authored by Dominik Charousset's avatar Dominik Charousset

Remove unused and undocumented MM hooks

parent 0ff9f69e
...@@ -18,7 +18,6 @@ set(LIBCAF_IO_SRCS ...@@ -18,7 +18,6 @@ set(LIBCAF_IO_SRCS
src/default_multiplexer.cpp src/default_multiplexer.cpp
src/doorman.cpp src/doorman.cpp
src/header.cpp src/header.cpp
src/hook.cpp
src/instance.cpp src/instance.cpp
src/interfaces.cpp src/interfaces.cpp
src/ip_endpoint.cpp src/ip_endpoint.cpp
......
...@@ -29,7 +29,6 @@ ...@@ -29,7 +29,6 @@
#include "caf/io/basp/header.hpp" #include "caf/io/basp/header.hpp"
#include "caf/io/basp/message_type.hpp" #include "caf/io/basp/message_type.hpp"
#include "caf/io/basp/routing_table.hpp" #include "caf/io/basp/routing_table.hpp"
#include "caf/io/hook.hpp"
#include "caf/io/middleman.hpp" #include "caf/io/middleman.hpp"
#include "caf/variant.hpp" #include "caf/variant.hpp"
...@@ -208,12 +207,6 @@ public: ...@@ -208,12 +207,6 @@ public:
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) {
system().middleman().template notify<Event>(std::forward<Ts>(xs)...);
}
actor_system& system() { actor_system& system() {
return callee_.system(); return callee_.system();
} }
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 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. *
******************************************************************************/
#pragma once
#include <set>
#include <string>
#include <memory>
#include <vector>
#include "caf/fwd.hpp"
namespace caf {
namespace io {
class hook;
// safes us some typing for the static dispatching
#define CAF_IO_HOOK_DISPATCH(eventname) \
template <typename... Ts> \
void dispatch(event<eventname>, Ts&&... ts) { \
eventname##_cb(std::forward<Ts>(ts)...); \
}
/// @relates hook
using hook_uptr = std::unique_ptr<hook>;
/// Interface to define hooks into the IO layer.
class hook {
public:
explicit hook(actor_system& sys);
virtual ~hook();
/// Called whenever a message has arrived via the network.
virtual void message_received_cb(const node_id& source,
const strong_actor_ptr& from,
const strong_actor_ptr& dest,
message_id mid,
const message& msg);
/// Called whenever a message has been sent to the network.
/// @param from The address of the sending actor.
/// @param hop The node in the network we've sent the message to.
/// @param dest The address of the receiving actor. Note that the node ID
/// of `dest` can differ from `hop` in case we don't
/// have a direct connection to `dest_node`.
/// @param mid The ID of the message.
/// @param payload The message we've sent.
virtual void message_sent_cb(const strong_actor_ptr& from, const node_id& hop,
const strong_actor_ptr& dest, message_id mid,
const message& payload);
/// Called whenever no route for sending a message exists.
virtual void message_sending_failed_cb(const strong_actor_ptr& from,
const strong_actor_ptr& dest,
message_id mid,
const message& payload);
/// Called whenever a message is forwarded to a different node.
virtual void message_forwarded_cb(const basp::header& hdr,
const std::vector<char>* payload);
/// Called whenever no route for a forwarding request exists.
virtual void message_forwarding_failed_cb(const basp::header& hdr,
const std::vector<char>* payload);
/// Called whenever an actor has been published.
virtual void actor_published_cb(const strong_actor_ptr& addr,
const std::set<std::string>& ifs,
uint16_t port);
/// Called whenever a new remote actor appeared.
virtual void new_remote_actor_cb(const strong_actor_ptr& addr);
/// Called whenever a handshake via a direct TCP connection succeeded.
virtual void new_connection_established_cb(const node_id& node);
/// Called whenever a message from or to a yet unknown node was received.
/// @param via The node that has sent us the message.
/// @param node The newly added entry to the routing table.
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
/// tried to send a message to an actor ID that could not be found
/// in the registry.
virtual void invalid_message_received_cb(const node_id& source,
const strong_actor_ptr& sender,
actor_id invalid_dest,
message_id mid, const message& msg);
/// Called before middleman shuts down.
virtual void before_shutdown_cb();
/// All possible events for IO hooks.
enum event_type {
message_received,
message_sent,
message_forwarded,
message_sending_failed,
message_forwarding_failed,
actor_published,
new_remote_actor,
new_connection_established,
new_route_added,
connection_lost,
route_lost,
invalid_message_received,
before_shutdown
};
/// Handles an event by invoking the associated callback.
template <event_type Event, typename... Ts>
void handle(Ts&&... ts) {
dispatch(event<Event>{}, std::forward<Ts>(ts)...);
}
inline actor_system& system() const {
return system_;
}
private:
// ------------ convenience interface based on static dispatching ------------
template <event_type Id>
using event = std::integral_constant<event_type, Id>;
CAF_IO_HOOK_DISPATCH(message_received)
CAF_IO_HOOK_DISPATCH(message_sent)
CAF_IO_HOOK_DISPATCH(message_forwarded)
CAF_IO_HOOK_DISPATCH(message_sending_failed)
CAF_IO_HOOK_DISPATCH(message_forwarding_failed)
CAF_IO_HOOK_DISPATCH(actor_published)
CAF_IO_HOOK_DISPATCH(new_remote_actor)
CAF_IO_HOOK_DISPATCH(new_connection_established)
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(before_shutdown)
actor_system& system_;
};
} // namespace io
} // namespace caf
...@@ -31,7 +31,6 @@ ...@@ -31,7 +31,6 @@
#include "caf/proxy_registry.hpp" #include "caf/proxy_registry.hpp"
#include "caf/send.hpp" #include "caf/send.hpp"
#include "caf/io/hook.hpp"
#include "caf/io/broker.hpp" #include "caf/io/broker.hpp"
#include "caf/io/middleman_actor.hpp" #include "caf/io/middleman_actor.hpp"
#include "caf/io/network/multiplexer.hpp" #include "caf/io/network/multiplexer.hpp"
...@@ -44,8 +43,6 @@ class middleman : public actor_system::module { ...@@ -44,8 +43,6 @@ class middleman : public actor_system::module {
public: public:
friend class ::caf::actor_system; friend class ::caf::actor_system;
using hook_vector = std::vector<hook_uptr>;
~middleman() override; ~middleman() override;
/// Tries to open a port for other CAF instances to connect to. /// Tries to open a port for other CAF instances to connect to.
...@@ -149,23 +146,6 @@ public: ...@@ -149,23 +146,6 @@ public:
/// Returns the IO backend used by this middleman. /// Returns the IO backend used by this middleman.
virtual network::multiplexer& backend() = 0; virtual network::multiplexer& backend() = 0;
/// Invokes the callback(s) associated with given event.
template <hook::event_type Event, typename... Ts>
void notify(Ts&&... ts) {
for (auto& hook : hooks_)
hook->handle<Event>(std::forward<Ts>(ts)...);
}
/// Returns whether this middleman has any hooks installed.
inline bool has_hook() const {
return !hooks_.empty();
}
/// Returns all installed hooks.
const hook_vector& hooks() const {
return hooks_;
}
/// Returns the actor associated with `name` at `nid` or /// Returns the actor associated with `name` at `nid` or
/// `invalid_actor` if `nid` is not connected or has no actor /// `invalid_actor` if `nid` is not connected or has no actor
/// associated to this `name`. /// associated to this `name`.
...@@ -348,8 +328,6 @@ private: ...@@ -348,8 +328,6 @@ private:
std::thread thread_; std::thread thread_;
// keeps track of "singleton-like" brokers // keeps track of "singleton-like" brokers
std::map<atom_value, actor> named_brokers_; std::map<atom_value, actor> named_brokers_;
// user-defined hooks
hook_vector hooks_;
// actor offering asyncronous IO by managing this singleton instance // actor offering asyncronous IO by managing this singleton instance
middleman_actor manager_; middleman_actor manager_;
}; };
......
...@@ -428,7 +428,6 @@ behavior basp_broker::make_behavior() { ...@@ -428,7 +428,6 @@ behavior basp_broker::make_behavior() {
state.instance.write_monitor_message(context(), state.get_buffer(hdl), state.instance.write_monitor_message(context(), state.get_buffer(hdl),
proxy->node(), proxy->id()); proxy->node(), proxy->id());
flush(hdl); flush(hdl);
system().middleman().notify<hook::new_remote_actor>(proxy);
}, },
// received from underlying broker implementation // received from underlying broker implementation
[=](const new_connection_msg& msg) { [=](const new_connection_msg& msg) {
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 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. *
******************************************************************************/
#include "caf/io/hook.hpp"
#include "caf/message_id.hpp"
namespace caf {
namespace io {
hook::hook(actor_system& sys) : system_(sys) {
// nop
}
hook::~hook() {
// nop
}
void hook::message_received_cb(const node_id&, const strong_actor_ptr&,
const strong_actor_ptr&, message_id,
const message&) {
// nop
}
void hook::message_sent_cb(const strong_actor_ptr&, const node_id&,
const strong_actor_ptr&, message_id,
const message&) {
// nop
}
void hook::message_forwarded_cb(const basp::header&, const std::vector<char>*) {
// nop
}
void hook::message_forwarding_failed_cb(const basp::header&,
const std::vector<char>*) {
// nop
}
void hook::message_sending_failed_cb(const strong_actor_ptr&,
const strong_actor_ptr&,
message_id, const message&) {
// nop
}
void hook::actor_published_cb(const strong_actor_ptr&,
const std::set<std::string>&, uint16_t) {
// nop
}
void hook::new_remote_actor_cb(const strong_actor_ptr&) {
// nop
}
void hook::new_connection_established_cb(const node_id&) {
// nop
}
void hook::new_route_added_cb(const node_id&, const node_id&) {
// nop
}
void hook::connection_lost_cb(const node_id&) {
// nop
}
void hook::route_lost_cb(const node_id&, const node_id&) {
// nop
}
void hook::invalid_message_received_cb(const node_id&, const strong_actor_ptr&,
actor_id, message_id, const message&) {
// nop
}
void hook::before_shutdown_cb() {
// nop
}
} // namespace io
} // namespace caf
...@@ -116,7 +116,6 @@ void instance::add_published_actor(uint16_t port, ...@@ -116,7 +116,6 @@ 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,
...@@ -168,10 +167,8 @@ bool instance::dispatch(execution_unit* ctx, const strong_actor_ptr& sender, ...@@ -168,10 +167,8 @@ bool instance::dispatch(execution_unit* ctx, const strong_actor_ptr& sender,
<< CAF_ARG(msg)); << CAF_ARG(msg));
CAF_ASSERT(dest_node && this_node_ != dest_node); CAF_ASSERT(dest_node && this_node_ != dest_node);
auto path = lookup(dest_node); auto path = lookup(dest_node);
if (!path) { if (!path)
//notify<hook::message_sending_failed>(sender, receiver, mid, msg);
return false; return false;
}
auto& source_node = sender ? sender->node() : this_node_; auto& source_node = sender ? sender->node() : this_node_;
if (dest_node == path->next_hop && source_node == this_node_) { if (dest_node == path->next_hop && source_node == this_node_) {
header hdr{message_type::direct_message, flags, 0, mid.integer_value(), header hdr{message_type::direct_message, flags, 0, mid.integer_value(),
...@@ -189,7 +186,6 @@ bool instance::dispatch(execution_unit* ctx, const strong_actor_ptr& sender, ...@@ -189,7 +186,6 @@ bool instance::dispatch(execution_unit* ctx, const strong_actor_ptr& sender,
write(ctx, callee_.get_buffer(path->hdl), hdr, &writer); write(ctx, callee_.get_buffer(path->hdl), hdr, &writer);
} }
flush(*path); flush(*path);
//notify<hook::message_sent>(sender, path->next_hop, receiver, mid, msg);
return true; return true;
} }
...@@ -468,10 +464,8 @@ void instance::forward(execution_unit* ctx, const node_id& dest_node, ...@@ -468,10 +464,8 @@ void instance::forward(execution_unit* ctx, const node_id& dest_node,
return; return;
} }
flush(*path); flush(*path);
notify<hook::message_forwarded>(hdr, &payload);
} else { } else {
CAF_LOG_WARNING("cannot forward message, no route to destination"); CAF_LOG_WARNING("cannot forward message, no route to destination");
notify<hook::message_forwarding_failed>(hdr, &payload);
} }
} }
......
...@@ -270,9 +270,6 @@ strong_actor_ptr middleman::remote_lookup(atom_value name, const node_id& nid) { ...@@ -270,9 +270,6 @@ strong_actor_ptr middleman::remote_lookup(atom_value name, const node_id& nid) {
void middleman::start() { void middleman::start() {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
// Create hooks.
for (auto& f : system().config().hook_factories)
hooks_.emplace_back(f(system_));
// Launch backend. // Launch backend.
if (!get_or(config(), "middleman.manual-multiplexing", false)) if (!get_or(config(), "middleman.manual-multiplexing", false))
backend_supervisor_ = backend().make_supervisor(); backend_supervisor_ = backend().make_supervisor();
...@@ -312,7 +309,6 @@ void middleman::stop() { ...@@ -312,7 +309,6 @@ void middleman::stop() {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
backend().dispatch([=] { backend().dispatch([=] {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
notify<hook::before_shutdown>();
// managers_ will be modified while we are stopping each manager, // managers_ will be modified while we are stopping each manager,
// because each manager will call remove(...) // because each manager will call remove(...)
for (auto& kvp : named_brokers_) { for (auto& kvp : named_brokers_) {
...@@ -333,7 +329,6 @@ void middleman::stop() { ...@@ -333,7 +329,6 @@ void middleman::stop() {
while (backend().try_run_once()) while (backend().try_run_once())
; // nop ; // nop
} }
hooks_.clear();
named_brokers_.clear(); named_brokers_.clear();
scoped_actor self{system(), true}; scoped_actor self{system(), true};
self->send_exit(manager_, exit_reason::kill); self->send_exit(manager_, exit_reason::kill);
......
...@@ -100,9 +100,6 @@ bool routing_table::erase_indirect(const node_id& dest) { ...@@ -100,9 +100,6 @@ bool routing_table::erase_indirect(const node_id& dest) {
auto i = indirect_.find(dest); auto i = indirect_.find(dest);
if (i == indirect_.end()) if (i == indirect_.end())
return false; return false;
if (parent_->parent().has_hook())
for (auto& nid : i->second)
parent_->parent().notify<hook::route_lost>(nid, dest);
indirect_.erase(i); indirect_.erase(i);
return true; return true;
} }
...@@ -115,7 +112,6 @@ void routing_table::add_direct(const connection_handle& hdl, ...@@ -115,7 +112,6 @@ void routing_table::add_direct(const connection_handle& hdl,
CAF_ASSERT(hdl_added && nid_added); CAF_ASSERT(hdl_added && nid_added);
CAF_IGNORE_UNUSED(hdl_added); CAF_IGNORE_UNUSED(hdl_added);
CAF_IGNORE_UNUSED(nid_added); CAF_IGNORE_UNUSED(nid_added);
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) {
...@@ -129,8 +125,7 @@ bool routing_table::add_indirect(const node_id& hop, const node_id& dest) { ...@@ -129,8 +125,7 @@ bool routing_table::add_indirect(const node_id& hop, const node_id& dest) {
// Add entry to our node ID set. // Add entry to our node ID set.
auto& hops = indirect_[dest]; auto& hops = indirect_[dest];
auto result = hops.empty(); auto result = hops.empty();
if (hops.emplace(hop).second) hops.emplace(hop);
parent_->parent().notify<hook::new_route_added>(hop, dest);
return result; return result;
} }
......
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