Commit 229f4935 authored by Dominik Charousset's avatar Dominik Charousset

Check content of storages in manual_stream test

parent 7c77a514
...@@ -196,10 +196,10 @@ struct core_state { ...@@ -196,10 +196,10 @@ struct core_state {
void init(event_based_actor* s, filter_type initial_filter) { void init(event_based_actor* s, filter_type initial_filter) {
self = s; self = s;
filter = std::move(initial_filter); filter = std::move(initial_filter);
governor = make_counted<stream_governor>(this);
sid = sid =
stream_id{self->ctrl(), stream_id{self->ctrl(),
self->new_request_id(message_priority::normal).integer_value()}; self->new_request_id(message_priority::normal).integer_value()};
governor = make_counted<stream_governor>(this);
self->streams().emplace(sid, governor); self->streams().emplace(sid, governor);
} }
...@@ -558,7 +558,12 @@ void driver(event_based_actor* self, const actor& sink) { ...@@ -558,7 +558,12 @@ void driver(event_based_actor* self, const actor& sink) {
); );
} }
void consumer(event_based_actor* self, filter_type ts, const actor& src) { struct consumer_state {
std::vector<element_type> xs;
};
void consumer(stateful_actor<consumer_state>* self, filter_type ts,
const actor& src) {
self->send(self * src, join_atom::value, std::move(ts)); self->send(self * src, join_atom::value, std::move(ts));
self->become( self->become(
[=](const stream_type& in) { [=](const stream_type& in) {
...@@ -570,14 +575,17 @@ void consumer(event_based_actor* self, filter_type ts, const actor& src) { ...@@ -570,14 +575,17 @@ void consumer(event_based_actor* self, filter_type ts, const actor& src) {
// nop // nop
}, },
// Process single element. // Process single element.
[](unit_t&, element_type) { [=](unit_t&, element_type x) {
// nop self->state.xs.emplace_back(std::move(x));
}, },
// Cleanup. // Cleanup.
[](unit_t&) { [](unit_t&) {
// nop // nop
} }
); );
},
[=](get_atom) {
return self->state.xs;
} }
); );
} }
...@@ -644,6 +652,16 @@ CAF_TEST(two_peers) { ...@@ -644,6 +652,16 @@ CAF_TEST(two_peers) {
.with(2, buf{{"b", 0}, {"b", 1}}, 0)); .with(2, buf{{"b", 0}, {"b", 1}}, 0));
expect((stream_msg::ack_batch), from(core2).to(core1).with(5, 0)); expect((stream_msg::ack_batch), from(core2).to(core1).with(5, 0));
expect((stream_msg::ack_batch), from(core1).to(d1).with(5, 0)); expect((stream_msg::ack_batch), from(core1).to(d1).with(5, 0));
// Check log of the consumer.
self->send(leaf, get_atom::value);
sched.prioritize(leaf);
sched.run_once();
self->receive(
[](const buf& xs) {
buf expected{{"b", 0}, {"b", 1}};
CAF_REQUIRE_EQUAL(xs, expected);
}
);
// Shutdown. // Shutdown.
CAF_MESSAGE("Shutdown core actors."); CAF_MESSAGE("Shutdown core actors.");
anon_send_exit(core1, exit_reason::user_shutdown); anon_send_exit(core1, exit_reason::user_shutdown);
......
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