Commit c57d18c2 authored by Dominik Charousset's avatar Dominik Charousset

Merge branch 'topic/streaming-fixes'

parents e9398439 d93ec704
...@@ -583,11 +583,24 @@ scheduled_actor::categorize(mailbox_element& x) { ...@@ -583,11 +583,24 @@ scheduled_actor::categorize(mailbox_element& x) {
auto em = content.move_if_unshared<exit_msg>(0); auto em = content.move_if_unshared<exit_msg>(0);
// make sure to get rid of attachables if they're no longer needed // make sure to get rid of attachables if they're no longer needed
unlink_from(em.source); unlink_from(em.source);
// exit_reason::kill is always fatal // exit_reason::kill is always fatal and also aborts streams.
if (em.reason == exit_reason::kill) if (em.reason == exit_reason::kill) {
quit(std::move(em.reason)); quit(std::move(em.reason));
else std::vector<stream_manager_ptr> xs;
for (auto& kvp : stream_managers_)
xs.emplace_back(kvp.second);
for (auto& kvp : pending_stream_managers_)
xs.emplace_back(kvp.second);
std::sort(xs.begin(), xs.end());
auto last = std::unique(xs.begin(), xs.end());
std::for_each(xs.begin(), last, [&](stream_manager_ptr& mgr) {
mgr->stop(exit_reason::kill);
});
stream_managers_.clear();
pending_stream_managers_.clear();
} else {
call_handler(exit_handler_, this, em); call_handler(exit_handler_, this, em);
}
return message_category::internal; return message_category::internal;
} }
case make_type_token<down_msg>(): { case make_type_token<down_msg>(): {
......
...@@ -51,16 +51,23 @@ void stream_manager::handle(inbound_path*, downstream_msg::batch&) { ...@@ -51,16 +51,23 @@ void stream_manager::handle(inbound_path*, downstream_msg::batch&) {
CAF_LOG_WARNING("unimplemented base handler for batches called"); CAF_LOG_WARNING("unimplemented base handler for batches called");
} }
void stream_manager::handle(inbound_path*, downstream_msg::close&) { void stream_manager::handle(inbound_path* in, downstream_msg::close&) {
// nop // Reset the actor handle to make sure no further messages travel upstream.
in->hdl = nullptr;
} }
void stream_manager::handle(inbound_path* in, downstream_msg::forced_close& x) { void stream_manager::handle(inbound_path* in, downstream_msg::forced_close& x) {
CAF_ASSERT(in != nullptr); CAF_ASSERT(in != nullptr);
CAF_LOG_TRACE(CAF_ARG2("slots", in->slots) << CAF_ARG(x)); CAF_LOG_TRACE(CAF_ARG2("slots", in->slots) << CAF_ARG(x));
// Reset the actor handle to make sure no further message travels upstream. // Reset the actor handle to make sure no further messages travel upstream.
in->hdl = nullptr; in->hdl = nullptr;
// A continuous stream exists independent of sources. Hence, we ignore
// upstream errors in this case.
if (!continuous()) {
stop(std::move(x.reason)); stop(std::move(x.reason));
} else {
CAF_LOG_INFO("received (and ignored) forced_close from a source");
}
} }
bool stream_manager::handle(stream_slots slots, upstream_msg::ack_open& x) { bool stream_manager::handle(stream_slots slots, upstream_msg::ack_open& x) {
......
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