Unverified Commit 3b33fe79 authored by Dominik Charousset's avatar Dominik Charousset Committed by GitHub

Merge pull request #1

Add unit test to container_source
parents 555b27a4 c160a53f
......@@ -45,7 +45,7 @@ struct container_source_state {
// -- properties -------------------------------------------------------------
size_t remaining() const {
size_t remaining() {
return static_cast<size_t>(std::distance(i, xs.end()));
}
......@@ -80,21 +80,21 @@ behavior container_source(container_source_type<Container>* self, Container xs,
// Spin up stream manager and connect the first sink.
self->state.init(std::move(xs));
auto src = self->make_source(
std::move(sink),
[&](unit_t&) {
// nop
},
[self](unit_t&, downstream<value_type>& out, size_t hint) {
auto& st = self->state;
auto n = std::min(hint, st.remaining());
for (size_t pushed = 0; pushed < n; ++pushed)
out.push(std::move(*st.i++));
},
[self](const unit_t&) { return self->state.at_end(); });
std::move(sink),
[&](unit_t&) {
// nop
},
[self](unit_t&, downstream<value_type>& out, size_t hint) {
auto& st = self->state;
auto n = std::min(hint, st.remaining());
for (size_t pushed = 0; pushed < n; ++pushed)
out.push(std::move(*st.i++));
},
[self](const unit_t&) { return self->state.at_end(); });
// Add the remaining sinks.
std::initializer_list<unit_t>{src.ptr()->add_outbound_path(sinks)...};
unit(src.ptr()->add_outbound_path(sinks)...);
return {};
};
}
/// Convenience function for spawning container sources.
......
......@@ -24,20 +24,93 @@
#include <vector>
#include "caf/actor_system.hpp"
#include "caf/actor_system_config.hpp"
using namespace caf;
namespace {
struct fixture {
using container_type = std::vector<int>;
TESTEE_SETUP();
TESTEE_STATE(container_sink) {
container_type con;
};
TESTEE(container_sink) {
return {[=](stream<container_type::value_type> in) {
return self->make_sink(
// input stream
in,
// initialize state
[=](unit_t&) {
// nop
},
// Consumer
[=](unit_t&, container_type::value_type val) {
self->state.con.emplace_back(std::move(val));
},
// cleanup and produce result message
[=](unit_t&, const error&) { CAF_MESSAGE(self->name() << " is done"); });
}};
}
TESTEE_STATE(container_monitor) {
actor streamer;
};
TESTEE(container_monitor) {
self->set_down_handler([=](const down_msg& dm) {
CAF_CHECK_EQUAL(dm.source, self->state.streamer);
});
return {[=](join_atom, actor streamer) {
self->state.streamer = streamer;
self->monitor(streamer);
}};
}
struct config : actor_system_config {
config() {
add_message_type<container_type::value_type>("value_type");
}
};
} // namespace <anonymous>
using fixture = test_coordinator_fixture<config>;
} // namespace
CAF_TEST_FIXTURE_SCOPE(container_source_tests, fixture)
CAF_TEST(todo) {
// implement me
CAF_TEST(stream_to_sink) {
scoped_actor self{sys};
container_type test_container{0, 1, 2, 3, 4, 5};
auto sink = sys.spawn(container_sink);
auto src = sys.spawn(bb::container_source<container_type, actor>,
test_container, sink);
auto mon = sys.spawn(container_monitor);
self->send(mon, join_atom::value, src);
run();
CAF_CHECK_EQUAL(deref<container_sink_actor>(sink).state.con, test_container);
}
CAF_TEST(stream_to_sinks) {
scoped_actor self{sys};
container_type test_container{0, 1, 2, 3, 4, 5};
auto snk1 = sys.spawn(container_sink);
auto snk2 = sys.spawn(container_sink);
auto snk3 = sys.spawn(container_sink);
auto src
= sys.spawn(bb::container_source<container_type, actor, actor, actor>,
test_container, snk1, snk2, snk3);
auto mon = sys.spawn(container_monitor);
self->send(mon, join_atom::value, src);
run();
CAF_CHECK_EQUAL(deref<container_sink_actor>(snk1).state.con, test_container);
CAF_CHECK_EQUAL(deref<container_sink_actor>(snk2).state.con, test_container);
CAF_CHECK_EQUAL(deref<container_sink_actor>(snk3).state.con, test_container);
}
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