Commit 4dda7d9a authored by Dominik Charousset's avatar Dominik Charousset

Implement test for delayed stream handshake

parent 84361ff7
...@@ -107,7 +107,7 @@ struct stream_sink_trait<void(State&, In), void(State&)> ...@@ -107,7 +107,7 @@ struct stream_sink_trait<void(State&, In), void(State&)>
/// Specializes the trait for batch-wise processing with result. /// Specializes the trait for batch-wise processing with result.
template <class State, class In, class Out> template <class State, class In, class Out>
struct stream_sink_trait<void(State&, std::vector<In>&), Out(State&)> struct stream_sink_trait<void(State&, std::vector<In>&&), Out(State&)>
: stream_sink_trait_base<State, In, Out> { : stream_sink_trait_base<State, In, Out> {
/// Defines a helper for dispatching to the finalizing function object. /// Defines a helper for dispatching to the finalizing function object.
using finalize = detail::stream_sink_trait_default_finalize; using finalize = detail::stream_sink_trait_default_finalize;
...@@ -118,7 +118,7 @@ struct stream_sink_trait<void(State&, std::vector<In>&), Out(State&)> ...@@ -118,7 +118,7 @@ struct stream_sink_trait<void(State&, std::vector<In>&), Out(State&)>
/// Specializes the trait for batch-wise processing without result. /// Specializes the trait for batch-wise processing without result.
template <class State, class In> template <class State, class In>
struct stream_sink_trait<void(State&, std::vector<In>&), void(State&)> struct stream_sink_trait<void(State&, std::vector<In>&&), void(State&)>
: stream_sink_trait_base<State, In, void> { : stream_sink_trait_base<State, In, void> {
/// Defines a helper for dispatching to the finalizing function object. /// Defines a helper for dispatching to the finalizing function object.
using finalize = detail::stream_sink_trait_void_finalize; using finalize = detail::stream_sink_trait_void_finalize;
......
...@@ -33,15 +33,23 @@ using std::vector; ...@@ -33,15 +33,23 @@ using std::vector;
using namespace caf; using namespace caf;
namespace { #define TESTEE_SCAFFOLD(tname) \
struct tname##_state { \
static const char* name; \
}; \
const char* tname##_state::name = #tname
struct file_reader_state { #define TESTEE(tname) \
static const char* name; TESTEE_SCAFFOLD(tname); \
}; behavior tname(stateful_actor<tname##_state>* self)
const char* file_reader_state::name = "file_reader"; #define VARARGS_TESTEE(tname, ...) \
TESTEE_SCAFFOLD(tname); \
behavior tname(stateful_actor<tname##_state>* self, __VA_ARGS__)
namespace {
behavior file_reader(stateful_actor<file_reader_state>* self, size_t buf_size) { VARARGS_TESTEE(file_reader, size_t buf_size) {
using buf = std::deque<int>; using buf = std::deque<int>;
return { return {
[=](string& fname) -> output_stream<int, string> { [=](string& fname) -> output_stream<int, string> {
...@@ -72,13 +80,7 @@ behavior file_reader(stateful_actor<file_reader_state>* self, size_t buf_size) { ...@@ -72,13 +80,7 @@ behavior file_reader(stateful_actor<file_reader_state>* self, size_t buf_size) {
}; };
} }
struct sum_up_state { TESTEE(sum_up) {
static const char* name;
};
const char* sum_up_state::name = "sum_up";
behavior sum_up(stateful_actor<sum_up_state>* self) {
return { return {
[=](stream<int>& in, const string& fname) { [=](stream<int>& in, const string& fname) {
CAF_CHECK_EQUAL(fname, "numbers.txt"); CAF_CHECK_EQUAL(fname, "numbers.txt");
...@@ -102,16 +104,41 @@ behavior sum_up(stateful_actor<sum_up_state>* self) { ...@@ -102,16 +104,41 @@ behavior sum_up(stateful_actor<sum_up_state>* self) {
}; };
} }
struct broken_sink_state { TESTEE(delayed_sum_up) {
static const char* name; self->set_default_handler(skip);
}; return {
[=](ok_atom) {
const char* broken_sink_state::name = "broken_sink"; self->become(
[=](stream<int>& in, const std::string& fname) {
CAF_CHECK_EQUAL(fname, "numbers.txt");
return self->make_sink(
// input stream
in,
// initialize state
[](int& x) {
x = 0;
},
// processing step
[](int& x, std::vector<int>&& ys) {
for (auto y : ys)
x += y;
},
// cleanup and produce result message
[](int& x) -> int {
return x;
}
);
}
);
}
};
}
behavior broken_sink(stateful_actor<broken_sink_state>*) { TESTEE(broken_sink) {
CAF_IGNORE_UNUSED(self);
return { return {
[=](stream<int>&, const std::string&) { [=](stream<int>&, const std::string& fname) {
// nop CAF_CHECK_EQUAL(fname, "numbers.txt");
} }
}; };
} }
...@@ -157,6 +184,38 @@ CAF_TEST(depth_2_pipeline_10_items) { ...@@ -157,6 +184,38 @@ CAF_TEST(depth_2_pipeline_10_items) {
CAF_CHECK_EQUAL(fail_state(src), exit_reason::normal); CAF_CHECK_EQUAL(fail_state(src), exit_reason::normal);
} }
CAF_TEST(delayed_depth_2_pipeline_10_items) {
std::chrono::microseconds cycle{cfg.streaming_credit_round_interval_us};
sched.clock().current_time += cycle;
auto src = sys.spawn(file_reader, 10);
auto snk = sys.spawn(delayed_sum_up);
auto pipeline = snk * src;
CAF_MESSAGE(CAF_ARG(self) << CAF_ARG(src) << CAF_ARG(snk));
CAF_MESSAGE("initiate stream handshake");
self->send(pipeline, "numbers.txt");
expect((string), from(self).to(src).with("numbers.txt"));
expect((open_stream_msg), from(self).to(snk));
disallow((upstream_msg::ack_open), from(snk).to(src));
disallow((upstream_msg::forced_drop), from(snk).to(src));
CAF_MESSAGE("send 'ok' to trigger sink to handle open_stream_msg");
self->send(snk, ok_atom::value);
expect((ok_atom), from(self).to(snk));
expect((open_stream_msg), from(self).to(snk));
expect((upstream_msg::ack_open), from(snk).to(src));
CAF_MESSAGE("start data transmission (a single batch)");
expect((downstream_msg::batch), from(src).to(snk));
sched.clock().current_time += cycle;
sched.dispatch();
expect((timeout_msg), from(snk).to(snk));
expect((timeout_msg), from(src).to(src));
expect((upstream_msg::ack_batch), from(snk).to(src));
CAF_MESSAGE("expect close message from src and then result from snk");
expect((downstream_msg::close), from(src).to(snk));
expect((int), from(snk).to(self).with(45));
CAF_CHECK_EQUAL(fail_state(snk), exit_reason::normal);
CAF_CHECK_EQUAL(fail_state(src), exit_reason::normal);
}
CAF_TEST(depth_2_pipeline_500_items) { CAF_TEST(depth_2_pipeline_500_items) {
std::chrono::microseconds cycle{cfg.streaming_credit_round_interval_us}; std::chrono::microseconds cycle{cfg.streaming_credit_round_interval_us};
sched.clock().current_time += cycle; sched.clock().current_time += cycle;
...@@ -200,4 +259,19 @@ CAF_TEST(broken_pipeline) { ...@@ -200,4 +259,19 @@ CAF_TEST(broken_pipeline) {
expect((error), from(snk).to(self).with(sec::stream_init_failed)); expect((error), from(snk).to(self).with(sec::stream_init_failed));
} }
CAF_TEST(delayed_pipeline) {
CAF_MESSAGE("streams must abort if a stage fails to initialize its state");
auto src = sys.spawn(file_reader, 50);
auto snk = sys.spawn(broken_sink);
auto pipeline = snk * src;
sched.run();
CAF_MESSAGE("initiate stream handshake");
self->send(pipeline, "test.txt");
expect((std::string), from(self).to(src).with("test.txt"));
expect((open_stream_msg), from(self).to(snk));
expect((upstream_msg::forced_drop), from(snk).to(src));
expect((error), from(snk).to(self).with(sec::stream_init_failed));
}
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