Commit 600bc956 authored by Dominik Charousset's avatar Dominik Charousset

Properly shutdown sources, only clear pending data

parent e14558b0
...@@ -52,6 +52,11 @@ public: ...@@ -52,6 +52,11 @@ public:
// -- implementation of virtual functions ------------------------------------ // -- implementation of virtual functions ------------------------------------
void shutdown() override {
super::shutdown();
at_end_ = true;
}
bool done() const override { bool done() const override {
return this->pending_handshakes_ == 0 && at_end_ && this->out_.clean(); return this->pending_handshakes_ == 0 && at_end_ && this->out_.clean();
} }
......
...@@ -74,6 +74,28 @@ std::function<void (T&, const error&)> fin(scheduled_actor* self) { ...@@ -74,6 +74,28 @@ std::function<void (T&, const error&)> fin(scheduled_actor* self) {
}; };
} }
TESTEE(infinite_source) {
return {
[=](string& fname) -> output_stream<int> {
CAF_CHECK_EQUAL(fname, "numbers.txt");
CAF_CHECK_EQUAL(self->mailbox().empty(), true);
return self->make_source(
[](int& x) {
x = 0;
},
[](int& x, downstream<int>& out, size_t num) {
for (size_t i = 0; i < num; ++i)
out.push(x++);
},
[](const int&) {
return false;
},
fin<int>(self)
);
}
};
}
VARARGS_TESTEE(file_reader, size_t buf_size) { VARARGS_TESTEE(file_reader, size_t buf_size) {
return { return {
[=](string& fname) -> output_stream<int> { [=](string& fname) -> output_stream<int> {
...@@ -441,4 +463,21 @@ CAF_TEST(depth_3_pipeline_graceful_shutdown) { ...@@ -441,4 +463,21 @@ CAF_TEST(depth_3_pipeline_graceful_shutdown) {
CAF_CHECK_EQUAL(deref<sum_up_actor>(snk).state.x, 625); CAF_CHECK_EQUAL(deref<sum_up_actor>(snk).state.x, 625);
} }
CAF_TEST(depth_3_pipeline_infinite_source) {
auto src = sys.spawn(infinite_source);
auto stg = sys.spawn(filter);
auto snk = sys.spawn(sum_up);
CAF_MESSAGE(CAF_ARG(self) << CAF_ARG(src) << CAF_ARG(stg) << CAF_ARG(snk));
CAF_MESSAGE("initiate stream handshake");
self->send(snk * stg * src, "numbers.txt");
expect((string), from(self).to(src).with("numbers.txt"));
expect((open_stream_msg), from(self).to(stg));
expect((open_stream_msg), from(self).to(snk));
expect((upstream_msg::ack_open), from(snk).to(stg));
expect((upstream_msg::ack_open), from(stg).to(src));
CAF_MESSAGE("send exit to the source and expect the stream to terminate");
anon_send_exit(src, exit_reason::user_shutdown);
run();
}
CAF_TEST_FIXTURE_SCOPE_END() 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