Commit ca51c757 authored by Dominik Charousset's avatar Dominik Charousset

Merge branch 'integration/issue/1076'

Closes #1076.
parents ef25ccde c39db78a
...@@ -147,6 +147,8 @@ is based on [Keep a Changelog](https://keepachangelog.com). ...@@ -147,6 +147,8 @@ is based on [Keep a Changelog](https://keepachangelog.com).
- Fix build error in the OpenSSL module under some MSVC configurations - Fix build error in the OpenSSL module under some MSVC configurations
- Serializer and deserializer now accept `std::chrono::time_point` for all clock - Serializer and deserializer now accept `std::chrono::time_point` for all clock
types instead of hard-wiring `std::system_clock`. types instead of hard-wiring `std::system_clock`.
- In some edge cases, actors failed to shut down properly when hosting a stream
source (#1076). The handshake process for a graceful shutdown has been fixed.
## [0.17.4] - 2019-02-08 ## [0.17.4] - 2019-02-08
......
...@@ -1135,9 +1135,15 @@ scheduled_actor::advance_streams(actor_clock::time_point now) { ...@@ -1135,9 +1135,15 @@ scheduled_actor::advance_streams(actor_clock::time_point now) {
auto bitmask = stream_ticks_.timeouts(now, {max_batch_delay_ticks_, auto bitmask = stream_ticks_.timeouts(now, {max_batch_delay_ticks_,
credit_round_ticks_}); credit_round_ticks_});
// Force batches on all output paths. // Force batches on all output paths.
if ((bitmask & 0x01) != 0) { if ((bitmask & 0x01) != 0 && !stream_managers_.empty()) {
std::vector<stream_manager*> managers;
managers.reserve(stream_managers_.size());
for (auto& kvp : stream_managers_) for (auto& kvp : stream_managers_)
kvp.second->out().force_emit_batches(); managers.emplace_back(kvp.second.get());
std::sort(managers.begin(), managers.end());
auto e = std::unique(managers.begin(), managers.end());
for (auto i = managers.begin(); i != e; ++i)
(*i)->out().force_emit_batches();
} }
// Fill up credit on each input path. // Fill up credit on each input path.
if ((bitmask & 0x02) != 0) { if ((bitmask & 0x02) != 0) {
......
...@@ -113,7 +113,8 @@ void stream_manager::handle(stream_slots slots, upstream_msg::ack_batch& x) { ...@@ -113,7 +113,8 @@ 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&) {
CAF_LOG_TRACE(CAF_ARG(slots)); CAF_LOG_TRACE(CAF_ARG(slots));
out().close(slots.receiver); error reason;
out().remove_path(slots.receiver, reason, false);
} }
void stream_manager::handle(stream_slots slots, upstream_msg::forced_drop& x) { void stream_manager::handle(stream_slots slots, upstream_msg::forced_drop& 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