Commit 89f9b6f3 authored by Dominik Charousset's avatar Dominik Charousset

Add `remove_input_path` function to stream manager

parent 67fcb31b
......@@ -408,6 +408,10 @@ public:
/// Silently closes incoming stream traffic on `slot`.
virtual void erase_inbound_path_later(stream_slot slot);
/// Closes incoming stream traffic on `slot`. Emits a drop message on the
/// path if `reason == none` and a `forced_drop` message otherwise.
virtual void erase_inbound_path_later(stream_slot slot, error reason);
/// Silently closes all inbound paths for `mgr`.
virtual void erase_inbound_paths_later(const stream_manager* mgr);
......
......@@ -737,6 +737,8 @@ public:
void erase_inbound_path_later(stream_slot slot) override;
void erase_inbound_path_later(stream_slot slot, error reason) override;
void erase_inbound_paths_later(const stream_manager* mgr) override;
void erase_inbound_paths_later(const stream_manager* mgr,
......
......@@ -138,6 +138,9 @@ public:
/// This function is called from the destructor of `inbound_path`.
virtual void deregister_input_path(inbound_path* x) noexcept;
/// Removes an input path
virtual void remove_input_path(stream_slot slot, error reason, bool silent);
// -- mutators ---------------------------------------------------------------
/// Adds a response promise to a sink for delivering the final result.
......
......@@ -134,6 +134,10 @@ void local_actor::erase_inbound_path_later(stream_slot) {
CAF_LOG_ERROR("local_actor::erase_inbound_path_later called");
}
void local_actor::erase_inbound_path_later(stream_slot, error) {
CAF_LOG_ERROR("local_actor::erase_inbound_path_later called");
}
void local_actor::erase_inbound_paths_later(const stream_manager*) {
CAF_LOG_ERROR("local_actor::erase_inbound_paths_later called");
}
......
......@@ -842,6 +842,22 @@ void scheduled_actor::erase_inbound_path_later(stream_slot slot) {
get_downstream_queue().erase_later(slot);
}
void scheduled_actor::erase_inbound_path_later(stream_slot slot, error reason) {
auto& q = get_downstream_queue();
auto e = q.queues().end();
auto i = q.queues().find(slot);
if (i != e) {
auto& path = i->second.policy().handler;
if (path != nullptr) {
if (reason == none)
path->emit_regular_shutdown(this);
else
path->emit_irregular_shutdown(this, std::move(reason));
}
q.erase_later(slot);
}
}
void scheduled_actor::erase_inbound_paths_later(const stream_manager* ptr) {
CAF_LOG_TRACE("");
for (auto& kvp : get_downstream_queue().queues()) {
......
......@@ -176,6 +176,14 @@ void stream_manager::deregister_input_path(inbound_path* ptr) noexcept {
CAF_LOG_DEBUG(inbound_paths_.size() << "paths remaining");
}
void stream_manager::remove_input_path(stream_slot slot, error reason,
bool silent) {
if (silent)
self_->erase_inbound_path_later(slot);
else
self_->erase_inbound_path_later(slot, std::move(reason));
}
void stream_manager::add_promise(response_promise x) {
CAF_LOG_TRACE(CAF_ARG(x));
CAF_ASSERT(out().terminal());
......
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