Commit 67fcb31b authored by Dominik Charousset's avatar Dominik Charousset

Rename `stream_scatterer::{take => remove}_path`

parent cb652868
...@@ -168,13 +168,14 @@ public: ...@@ -168,13 +168,14 @@ public:
return result; return result;
} }
unique_path_ptr take_path(stream_slot slot) noexcept override { bool remove_path(stream_slot slot, error reason,
bool silent) noexcept override {
auto i = paths_.find(slot); auto i = paths_.find(slot);
if (i == paths_.end()) if (i == paths_.end())
return nullptr; return false;
auto owner = i->second.owner; auto owner = i->second.owner;
paths_.erase(i); paths_.erase(i);
return owner->take_path(slot); return owner->remove_path(slot, std::move(reason), silent);
} }
path_ptr path(stream_slot slot) noexcept override { path_ptr path(stream_slot slot) noexcept override {
......
...@@ -34,7 +34,8 @@ public: ...@@ -34,7 +34,8 @@ public:
path_ptr add_path(stream_slots slots, strong_actor_ptr target) override; path_ptr add_path(stream_slots slots, strong_actor_ptr target) override;
unique_path_ptr take_path(stream_slot slots) noexcept override; bool remove_path(stream_slot slots, error reason,
bool silent) noexcept override;
path_ptr path(stream_slot slots) noexcept override; path_ptr path(stream_slot slots) noexcept override;
......
...@@ -120,7 +120,8 @@ public: ...@@ -120,7 +120,8 @@ public:
virtual path_ptr add_path(stream_slots slots, strong_actor_ptr target) = 0; virtual path_ptr add_path(stream_slots slots, strong_actor_ptr target) = 0;
/// Removes a path from the scatterer and returns it. /// Removes a path from the scatterer and returns it.
virtual unique_path_ptr take_path(stream_slot slots) noexcept = 0; virtual bool remove_path(stream_slot slots, error reason,
bool silent) noexcept = 0;
/// Returns the path associated to `slots` or `nullptr`. /// Returns the path associated to `slots` or `nullptr`.
virtual path_ptr path(stream_slot slots) noexcept = 0; virtual path_ptr path(stream_slot slots) noexcept = 0;
......
...@@ -61,7 +61,8 @@ public: ...@@ -61,7 +61,8 @@ public:
path_ptr add_path(stream_slots slots, strong_actor_ptr target) override; path_ptr add_path(stream_slots slots, strong_actor_ptr target) override;
unique_path_ptr take_path(stream_slot slots) noexcept override; bool remove_path(stream_slot slots, error reason,
bool silent) noexcept override;
path_ptr path(stream_slot slots) noexcept override; path_ptr path(stream_slot slots) noexcept override;
......
...@@ -42,9 +42,8 @@ auto invalid_stream_scatterer::add_path(stream_slots, strong_actor_ptr) ...@@ -42,9 +42,8 @@ auto invalid_stream_scatterer::add_path(stream_slots, strong_actor_ptr)
return nullptr; return nullptr;
} }
auto invalid_stream_scatterer::take_path(stream_slot) noexcept bool invalid_stream_scatterer::remove_path(stream_slot, error, bool) noexcept {
-> unique_path_ptr { return false;
return unique_path_ptr{nullptr};
} }
auto invalid_stream_scatterer::path(stream_slot) noexcept -> path_ptr { auto invalid_stream_scatterer::path(stream_slot) noexcept -> path_ptr {
......
...@@ -83,11 +83,12 @@ void stream_manager::handle(stream_slots slots, upstream_msg::ack_batch& x) { ...@@ -83,11 +83,12 @@ void stream_manager::handle(stream_slots slots, upstream_msg::ack_batch& x) {
} }
void stream_manager::handle(stream_slots slots, upstream_msg::drop&) { void stream_manager::handle(stream_slots slots, upstream_msg::drop&) {
out().take_path(slots.receiver); error tmp;
out().remove_path(slots.receiver, std::move(tmp), true);
} }
void stream_manager::handle(stream_slots slots, upstream_msg::forced_drop& x) { void stream_manager::handle(stream_slots slots, upstream_msg::forced_drop& x) {
if (out().path(slots.receiver) != nullptr) if (out().remove_path(slots.receiver, x.reason, true))
abort(std::move(x.reason)); abort(std::move(x.reason));
} }
......
...@@ -49,16 +49,15 @@ size_t stream_scatterer_impl::num_paths() const noexcept { ...@@ -49,16 +49,15 @@ size_t stream_scatterer_impl::num_paths() const noexcept {
return paths_.size(); return paths_.size();
} }
auto stream_scatterer_impl::take_path(stream_slot slot) noexcept bool stream_scatterer_impl::remove_path(stream_slot slot, error reason,
-> unique_path_ptr { bool silent) noexcept {
unique_path_ptr result;
auto i = paths_.find(slot); auto i = paths_.find(slot);
if (i != paths_.end()) { if (i != paths_.end()) {
about_to_erase(i->second.get(), true, nullptr); about_to_erase(i->second.get(), silent, &reason);
result.swap(i->second);
paths_.erase(i); paths_.erase(i);
return true;
} }
return result; return false;
} }
auto stream_scatterer_impl::path(stream_slot slot) noexcept -> path_ptr { auto stream_scatterer_impl::path(stream_slot slot) noexcept -> path_ptr {
......
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