Commit 1e9fab4d authored by Dominik Charousset's avatar Dominik Charousset

Fix shutdown process in stream governor

parent c62ce9ee
......@@ -77,8 +77,9 @@ public:
// -- handler for stream-wide events -----------------------------------------
/// Shutdown the stream due to a fatal error.
virtual void abort(strong_actor_ptr& cause, const error& reason) = 0;
/// Signals an error at the up- or downstream actor `hdl`. This function is
/// called with `hdl == nullptr` if the parent actor shuts down.
virtual void abort(strong_actor_ptr& hdl, const error& reason) = 0;
// -- accessors --------------------------------------------------------------
......
......@@ -364,7 +364,7 @@ error stream_governor::close_upstream(strong_actor_ptr& hdl) {
void stream_governor::abort(strong_actor_ptr& hdl, const error& reason) {
CAF_LOG_TRACE(CAF_ARG(hdl) << CAF_ARG(reason));
CAF_IGNORE_UNUSED(reason);
if (hdl != nullptr) {
if (local_subscribers_.remove_path(hdl))
return;
auto i = peers_.find(hdl);
......@@ -373,6 +373,12 @@ void stream_governor::abort(strong_actor_ptr& hdl, const error& reason) {
state_->self->streams().erase(pd.incoming_sid);
peers_.erase(i);
}
} else {
local_subscribers_.abort(hdl, reason);
for (auto& kvp : peers_)
kvp.second->out.abort(hdl, reason);
peers_.clear();
}
}
bool stream_governor::done() const {
......@@ -666,27 +672,7 @@ CAF_TEST(two_peers) {
CAF_MESSAGE("Shutdown core actors.");
anon_send_exit(core1, exit_reason::user_shutdown);
anon_send_exit(core2, exit_reason::user_shutdown);
anon_send_exit(leaf, exit_reason::user_shutdown);
sched.run();
return;
// core1 <----(stream_msg::ack_open)------ core2
expect((stream_msg::ack_open), from(core2).to(core1).with(_, 5, _, false));
// core1 ----(stream_msg::batch)---> core2
expect((stream_msg::batch),
from(core1).to(core2).with(5, std::vector<int>{1, 2, 3, 4, 5}, 0));
// core1 <--(stream_msg::ack_batch)---- core2
expect((stream_msg::ack_batch), from(core2).to(core1).with(5, 0));
// core1 ----(stream_msg::batch)---> core2
expect((stream_msg::batch),
from(core1).to(core2).with(4, std::vector<int>{6, 7, 8, 9}, 1));
// core1 <--(stream_msg::ack_batch)---- core2
expect((stream_msg::ack_batch), from(core2).to(core1).with(4, 1));
// core1 ----(stream_msg::close)---> core2
expect((stream_msg::close), from(core1).to(core2).with());
// core2 ----(result: 25)---> core1
expect((int), from(core2).to(core1).with(45));
}
CAF_TEST_FIXTURE_SCOPE_END()
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