Commit 75218859 authored by Dominik Charousset's avatar Dominik Charousset

Rename {bool remove => void erase}_stream_manager

parent ea00044b
...@@ -1111,7 +1111,7 @@ public: ...@@ -1111,7 +1111,7 @@ public:
bool add_stream_manager(stream_slots id, stream_manager_ptr ptr); bool add_stream_manager(stream_slots id, stream_manager_ptr ptr);
/// Removes a new stream manager. /// Removes a new stream manager.
bool remove_stream_manager(stream_slots id); void erase_stream_manager(stream_slots id);
/// @pre `x.content().match_elements<open_stream_msg>()` /// @pre `x.content().match_elements<open_stream_msg>()`
invoke_message_result handle_open_stream_msg(mailbox_element& x); invoke_message_result handle_open_stream_msg(mailbox_element& x);
......
...@@ -284,6 +284,7 @@ struct downstream_msg_visitor { ...@@ -284,6 +284,7 @@ struct downstream_msg_visitor {
auto& inptr = q_ref.policy().handler; auto& inptr = q_ref.policy().handler;
if (inptr == nullptr) if (inptr == nullptr)
return intrusive::task_result::stop; return intrusive::task_result::stop;
// Do *not* store a reference here since we potentially reset `inptr`.
auto mgr = inptr->mgr; auto mgr = inptr->mgr;
inptr->handle(x); inptr->handle(x);
// TODO: replace with `if constexpr` when switching to C++17 // TODO: replace with `if constexpr` when switching to C++17
...@@ -291,7 +292,7 @@ struct downstream_msg_visitor { ...@@ -291,7 +292,7 @@ struct downstream_msg_visitor {
|| std::is_same<T, downstream_msg::forced_close>::value) { || std::is_same<T, downstream_msg::forced_close>::value) {
inptr.reset(); inptr.reset();
qs_ref.erase_later(dm.slots.receiver); qs_ref.erase_later(dm.slots.receiver);
selfptr->remove_stream_manager(dm.slots); selfptr->erase_stream_manager(dm.slots);
if (mgr->done()) if (mgr->done())
mgr->stop(); mgr->stop();
return intrusive::task_result::stop; return intrusive::task_result::stop;
...@@ -299,7 +300,7 @@ struct downstream_msg_visitor { ...@@ -299,7 +300,7 @@ struct downstream_msg_visitor {
else if (mgr->done()) { else if (mgr->done()) {
CAF_LOG_DEBUG("path is done receiving and closes its manager"); CAF_LOG_DEBUG("path is done receiving and closes its manager");
mgr->stop(); mgr->stop();
selfptr->remove_stream_manager(dm.slots); selfptr->erase_stream_manager(dm.slots);
return intrusive::task_result::stop; return intrusive::task_result::stop;
} }
return intrusive::task_result::resume; return intrusive::task_result::resume;
...@@ -312,6 +313,7 @@ intrusive::task_result scheduled_actor::mailbox_visitor:: ...@@ -312,6 +313,7 @@ intrusive::task_result scheduled_actor::mailbox_visitor::
operator()(size_t, downstream_queue& qs, stream_slot, operator()(size_t, downstream_queue& qs, stream_slot,
policy::downstream_messages::nested_queue_type& q, policy::downstream_messages::nested_queue_type& q,
mailbox_element& x) { mailbox_element& x) {
CAF_LOG_TRACE(CAF_ARG(x) << CAF_ARG(handled_msgs));
CAF_ASSERT(x.content().type_token() == make_type_token<downstream_msg>()); CAF_ASSERT(x.content().type_token() == make_type_token<downstream_msg>());
auto& dm = x.content().get_mutable_as<downstream_msg>(0); auto& dm = x.content().get_mutable_as<downstream_msg>(0);
downstream_msg_visitor f{self, qs, q, dm}; downstream_msg_visitor f{self, qs, q, dm};
...@@ -924,13 +926,10 @@ bool scheduled_actor::add_stream_manager(stream_slots id, ...@@ -924,13 +926,10 @@ bool scheduled_actor::add_stream_manager(stream_slots id,
return result; return result;
} }
bool scheduled_actor::remove_stream_manager(stream_slots id) { void scheduled_actor::erase_stream_manager(stream_slots id) {
CAF_LOG_TRACE(CAF_ARG(id)); CAF_LOG_TRACE(CAF_ARG(id));
if (stream_managers_.erase(id) == 0) if (stream_managers_.erase(id) != 0 && stream_managers_.empty())
return false;
if (stream_managers_.empty())
stream_ticks_.stop(); stream_ticks_.stop();
return true;
} }
invoke_message_result invoke_message_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