Commit 044c483c authored by Joseph Noir's avatar Joseph Noir

Add custom timout handler to scheduled actor

parent 1befb65b
......@@ -205,6 +205,9 @@ public:
/// Function object for handling exit messages.
using exit_handler = std::function<void (pointer, exit_msg&)>;
/// Function object for handling timeouts.
using timeout_handler = std::function<void(pointer, timeout_msg&)>;
# ifndef CAF_NO_EXCEPTIONS
/// Function object for handling exit messages.
using exception_handler = std::function<error (pointer, std::exception_ptr&)>;
......@@ -245,6 +248,8 @@ public:
static void default_exit_handler(pointer ptr, exit_msg& x);
static void default_timeout_handler(pointer ptr, timeout_msg& x);
# ifndef CAF_NO_EXCEPTIONS
static error default_exception_handler(pointer ptr, std::exception_ptr& x);
# endif // CAF_NO_EXCEPTIONS
......@@ -389,6 +394,19 @@ public:
set_exit_handler([fun](scheduled_actor*, exit_msg& x) { fun(x); });
}
/// Sets a custom handler for timeout messages.
inline void set_timeout_handler(timeout_handler fun) {
if (fun)
timeout_handler_ = std::move(fun);
else
timeout_handler_ = default_timeout_handler;
}
template <class T>
auto set_timeout_handler(T fun) -> decltype (fun(std::declval<timeout_msg&>())) {
set_timeout_handler([fun](scheduled_actor*, timeout_msg& x) { fun(x); });
}
# ifndef CAF_NO_EXCEPTIONS
/// Sets a custom exception handler for this actor. If multiple handlers are
/// defined, only the functor that was added *last* is being executed.
......@@ -913,6 +931,9 @@ protected:
/// Customization point for setting a default `exit_msg` callback.
exit_handler exit_handler_;
/// Customization point for setting a default `timeout_msg` callback.
timeout_handler timeout_handler_;
/// Stores stream managers for established streams.
stream_manager_map stream_managers_;
......
......@@ -90,6 +90,12 @@ void scheduled_actor::default_exit_handler(scheduled_actor* ptr, exit_msg& x) {
default_error_handler(ptr, x.reason);
}
void scheduled_actor::default_timeout_handler(scheduled_actor* ptr,
timeout_msg& x) {
aout(ptr) << "*** unhandled timeout message [id: " << ptr->id()
<< "]: " << to_string(x) << std::endl;
}
# ifndef CAF_NO_EXCEPTIONS
error scheduled_actor::default_exception_handler(pointer ptr,
std::exception_ptr& x) {
......@@ -120,6 +126,7 @@ scheduled_actor::scheduled_actor(actor_config& cfg)
error_handler_(default_error_handler),
down_handler_(default_down_handler),
exit_handler_(default_exit_handler),
timeout_handler_(default_timeout_handler),
private_thread_(nullptr)
# ifndef CAF_NO_EXCEPTIONS
, exception_handler_(default_exception_handler)
......@@ -566,16 +573,17 @@ scheduled_actor::categorize(mailbox_element& x) {
return message_category::ordinary;
case make_type_token<timeout_msg>(): {
CAF_ASSERT(!x.mid.valid());
auto& tm = content.get_as<timeout_msg>(0);
auto tm = content.get_as<timeout_msg>(0);
auto tid = tm.timeout_id;
if (tm.type == receive_atom::value) {
CAF_LOG_DEBUG("handle ordinary timeout message");
if (is_active_receive_timeout(tid) && !bhvr_stack_.empty())
bhvr_stack_.back().handle_timeout();
} else {
CAF_ASSERT(tm.type == atom("stream"));
} else if (tm.type == atom("stream")) {
CAF_LOG_DEBUG("handle stream timeout message");
set_stream_timeout(advance_streams(clock().now()));
} else {
call_handler(timeout_handler_, this, tm);
}
return message_category::internal;
}
......
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