Commit 3f135a62 authored by Dominik Charousset's avatar Dominik Charousset

Allow setting high priority for down messages

parent 191679e8
...@@ -40,14 +40,14 @@ public: ...@@ -40,14 +40,14 @@ public:
bool matches(const token& what) override; bool matches(const token& what) override;
inline static attachable_ptr make_monitor(actor_addr observed, static attachable_ptr
actor_addr observer) { make_monitor(actor_addr observed, actor_addr observer,
return attachable_ptr{new default_attachable(std::move(observed), message_priority prio = message_priority::normal) {
std::move(observer), monitor)}; return attachable_ptr{new default_attachable(
std::move(observed), std::move(observer), monitor, prio)};
} }
inline static attachable_ptr make_link(actor_addr observed, static attachable_ptr make_link(actor_addr observed, actor_addr observer) {
actor_addr observer) {
return attachable_ptr{new default_attachable(std::move(observed), return attachable_ptr{new default_attachable(std::move(observed),
std::move(observer), link)}; std::move(observer), link)};
} }
...@@ -70,10 +70,21 @@ public: ...@@ -70,10 +70,21 @@ public:
}; };
private: private:
default_attachable(actor_addr observed, actor_addr observer, observe_type type); default_attachable(actor_addr observed, actor_addr observer,
observe_type type,
message_priority prio = message_priority::normal);
/// Holds a weak reference to the observed actor.
actor_addr observed_; actor_addr observed_;
/// Holds a weak reference to the observing actor.
actor_addr observer_; actor_addr observer_;
/// Defines the type of message we wish to send.
observe_type type_; observe_type type_;
/// Defines the priority for the message.
message_priority priority_;
}; };
} // namespace caf } // namespace caf
......
...@@ -208,7 +208,7 @@ public: ...@@ -208,7 +208,7 @@ public:
/// @cond PRIVATE /// @cond PRIVATE
void monitor(abstract_actor* ptr); void monitor(abstract_actor* ptr, message_priority prio);
/// @endcond /// @endcond
...@@ -289,9 +289,9 @@ public: ...@@ -289,9 +289,9 @@ public:
/// Adds a unidirectional `monitor` to `whom`. /// Adds a unidirectional `monitor` to `whom`.
/// @note Each call to `monitor` creates a new, independent monitor. /// @note Each call to `monitor` creates a new, independent monitor.
template <class Handle> template <message_priority P = message_priority::normal, class Handle = actor>
void monitor(const Handle& whom) { void monitor(const Handle& whom) {
monitor(actor_cast<abstract_actor*>(whom)); monitor(actor_cast<abstract_actor*>(whom), P);
} }
/// Removes a monitor from `whom`. /// Removes a monitor from `whom`.
......
...@@ -87,9 +87,9 @@ public: ...@@ -87,9 +87,9 @@ public:
return self_->make_response_promise<Ts...>(); return self_->make_response_promise<Ts...>();
} }
template <class T> template <message_priority P = message_priority::normal, class Handle = actor>
void monitor(const T& x) { void monitor(const Handle& x) {
self_->monitor(x); self_->monitor<P>(x);
} }
template <class T> template <class T>
......
...@@ -40,7 +40,7 @@ void default_attachable::actor_exited(const error& rsn, execution_unit* host) { ...@@ -40,7 +40,7 @@ void default_attachable::actor_exited(const error& rsn, execution_unit* host) {
auto observer = actor_cast<strong_actor_ptr>(observer_); auto observer = actor_cast<strong_actor_ptr>(observer_);
auto observed = actor_cast<strong_actor_ptr>(observed_); auto observed = actor_cast<strong_actor_ptr>(observed_);
if (observer) if (observer)
observer->enqueue(std::move(observed), make_message_id(), observer->enqueue(std::move(observed), make_message_id(priority_),
factory(actor_cast<abstract_actor*>(observed_), rsn), factory(actor_cast<abstract_actor*>(observed_), rsn),
host); host);
} }
...@@ -53,10 +53,12 @@ bool default_attachable::matches(const token& what) { ...@@ -53,10 +53,12 @@ bool default_attachable::matches(const token& what) {
} }
default_attachable::default_attachable(actor_addr observed, actor_addr observer, default_attachable::default_attachable(actor_addr observed, actor_addr observer,
observe_type type) observe_type type,
message_priority priority)
: observed_(std::move(observed)), : observed_(std::move(observed)),
observer_(std::move(observer)), observer_(std::move(observer)),
type_(type) { type_(type),
priority_(priority) {
// nop // nop
} }
......
...@@ -66,9 +66,10 @@ void local_actor::request_response_timeout(const duration& d, message_id mid) { ...@@ -66,9 +66,10 @@ void local_actor::request_response_timeout(const duration& d, message_id mid) {
clock().set_request_timeout(t, this, mid.response_id()); clock().set_request_timeout(t, this, mid.response_id());
} }
void local_actor::monitor(abstract_actor* ptr) { void local_actor::monitor(abstract_actor* ptr, message_priority priority) {
if (ptr != nullptr) if (ptr != nullptr)
ptr->attach(default_attachable::make_monitor(ptr->address(), address())); ptr->attach(default_attachable::make_monitor(ptr->address(), address(),
priority));
} }
void local_actor::demonitor(const actor_addr& whom) { void local_actor::demonitor(const actor_addr& whom) {
......
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