Commit 73f9eb8b authored by Dominik Charousset's avatar Dominik Charousset

Implement `source -> stage -> (sink1, sink2)` test

parent 73715e00
......@@ -104,6 +104,10 @@ TESTEE(sum_up) {
return x;
}
);
},
[=](join_atom atm, actor src) {
CAF_MESSAGE(self->name() << " joins a stream");
self->send(self * src, atm);
}
};
}
......@@ -175,12 +179,52 @@ TESTEE(filter) {
};
}
struct stream_multiplexer_state {
stream_stage_ptr<int, int, std::string> stage;
static const char* name;
};
const char* stream_multiplexer_state::name = "stream_multiplexer";
behavior stream_multiplexer(stateful_actor<stream_multiplexer_state>* self) {
self->state.stage = self->make_continuous_stage(
// handshake data
std::make_tuple(std::string{"numbers.txt"}),
// initialize state
[=](unit_t&) {
// nop
},
// processing step
[=](unit_t&, downstream<int>& out, int x) {
out.push(x);
},
// cleanup
[=](unit_t&) {
CAF_MESSAGE("stream_multiplexer is done");
}
);
return {
[=](join_atom) {
CAF_MESSAGE("received 'join' request");
return self->add_output_path(self->state.stage);
},
[=](const stream<int>& in, std::string& fname) {
CAF_CHECK_EQUAL(fname, "numbers.txt");
return self->add_input_path(in, self->state.stage);
},
};
}
struct fixture : test_coordinator_fixture<> {
std::chrono::microseconds cycle;
fixture() : cycle(cfg.streaming_credit_round_interval_us) {
// Configure the clock to measure each batch item with 1us.
sched.clock().time_per_unit.emplace(atom("batch"), timespan{1000});
}
~fixture() {
sched.run();
}
};
error fail_state(const actor& x) {
......@@ -375,4 +419,32 @@ CAF_TEST(depth_3_pipeline_50_items) {
expect((int), from(snk).to(self).with(625));
}
CAF_TEST(depth_3_pipeline_with_fork) {
sched.clock().current_time += cycle;
auto src = sys.spawn(file_reader, 50);
auto stg = sys.spawn(stream_multiplexer);
auto snk1 = sys.spawn(sum_up);
auto snk2 = sys.spawn(sum_up);
auto pipeline = stg * src;
auto& st = deref<stateful_actor<stream_multiplexer_state>>(stg).state;
CAF_MESSAGE("connect sinks to the stage (fork)");
self->send(snk1, join_atom::value, stg);
self->send(snk2, join_atom::value, stg);
sched.run();
CAF_CHECK_EQUAL(st.stage->out().paths().size(), 2u);
CAF_MESSAGE("connect source to the stage (fork)");
self->send(pipeline, "numbers.txt");
sched.run();
CAF_CHECK_EQUAL(st.stage->out().paths().size(), 2u);
CAF_CHECK_EQUAL(st.stage->inbound_paths().size(), 1u);
auto predicate = [&] {
return st.stage->inbound_paths().empty();
};
sched.run_dispatch_loop(predicate, cycle);
self->send_exit(stg, exit_reason::kill);
CAF_CHECK_EQUAL(st.stage->out().paths().size(), 2u);
CAF_CHECK_EQUAL(st.stage->inbound_paths().size(), 0u);
}
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