Commit 97024e8c authored by Dominik Charousset's avatar Dominik Charousset

Add trailing commas for nicer formatting

parent 460a1953
...@@ -58,38 +58,46 @@ void push(std::deque<T>& xs, downstream<T>& out, size_t num) { ...@@ -58,38 +58,46 @@ void push(std::deque<T>& xs, downstream<T>& out, size_t num) {
VARARGS_TESTEE(int_file_reader, size_t buf_size) { VARARGS_TESTEE(int_file_reader, size_t buf_size) {
using buf = std::deque<int32_t>; using buf = std::deque<int32_t>;
return {[=](string& fname) -> result<stream<int32_t>> { return {
CAF_CHECK_EQUAL(fname, "numbers.txt"); [=](string& fname) -> result<stream<int32_t>> {
return attach_stream_source( CAF_CHECK_EQUAL(fname, "numbers.txt");
self, return attach_stream_source(
// initialize state self,
[=](buf& xs) { // initialize state
xs.resize(buf_size); [=](buf& xs) {
std::iota(xs.begin(), xs.end(), 1); xs.resize(buf_size);
}, std::iota(xs.begin(), xs.end(), 1);
// get next element },
[](buf& xs, downstream<int32_t>& out, size_t num) { push(xs, out, num); }, // get next element
// check whether we reached the end [](buf& xs, downstream<int32_t>& out, size_t num) {
[=](const buf& xs) { return xs.empty(); }); push(xs, out, num);
}}; },
// check whether we reached the end
[=](const buf& xs) { return xs.empty(); });
},
};
} }
VARARGS_TESTEE(string_file_reader, size_t buf_size) { VARARGS_TESTEE(string_file_reader, size_t buf_size) {
using buf = std::deque<string>; using buf = std::deque<string>;
return {[=](string& fname) -> result<stream<string>> { return {
CAF_CHECK_EQUAL(fname, "strings.txt"); [=](string& fname) -> result<stream<string>> {
return attach_stream_source( CAF_CHECK_EQUAL(fname, "strings.txt");
self, return attach_stream_source(
// initialize state self,
[=](buf& xs) { // initialize state
for (size_t i = 0; i < buf_size; ++i) [=](buf& xs) {
xs.emplace_back("some string data"); for (size_t i = 0; i < buf_size; ++i)
}, xs.emplace_back("some string data");
// get next element },
[](buf& xs, downstream<string>& out, size_t num) { push(xs, out, num); }, // get next element
// check whether we reached the end [](buf& xs, downstream<string>& out, size_t num) {
[=](const buf& xs) { return xs.empty(); }); push(xs, out, num);
}}; },
// check whether we reached the end
[=](const buf& xs) { return xs.empty(); });
},
};
} }
TESTEE_STATE(sum_up) { TESTEE_STATE(sum_up) {
...@@ -98,24 +106,26 @@ TESTEE_STATE(sum_up) { ...@@ -98,24 +106,26 @@ TESTEE_STATE(sum_up) {
TESTEE(sum_up) { TESTEE(sum_up) {
using intptr = int*; using intptr = int*;
return {[=](stream<int32_t>& in) { return {
return attach_stream_sink( [=](stream<int32_t>& in) {
self, return attach_stream_sink(
// input stream self,
in, // input stream
// initialize state in,
[=](intptr& x) { x = &self->state.x; }, // initialize state
// processing step [=](intptr& x) { x = &self->state.x; },
[](intptr& x, int32_t y) { *x += y; }, // processing step
// cleanup and produce result message [](intptr& x, int32_t y) { *x += y; },
[=](intptr&, const error&) { // cleanup and produce result message
CAF_MESSAGE(self->name() << " is done"); [=](intptr&, const error&) {
}); CAF_MESSAGE(self->name() << " is done");
}, });
[=](join_atom, actor src) { },
CAF_MESSAGE(self->name() << " joins a stream"); [=](join_atom, actor src) {
self->send(self * src, join_atom::value, ints_atom::value); CAF_MESSAGE(self->name() << " joins a stream");
}}; self->send(self * src, join_atom::value, ints_atom::value);
},
};
} }
TESTEE_STATE(collect) { TESTEE_STATE(collect) {
...@@ -123,28 +133,30 @@ TESTEE_STATE(collect) { ...@@ -123,28 +133,30 @@ TESTEE_STATE(collect) {
}; };
TESTEE(collect) { TESTEE(collect) {
return {[=](stream<string>& in) { return {
return attach_stream_sink( [=](stream<string>& in) {
self, return attach_stream_sink(
// input stream self,
in, // input stream
// initialize state in,
[](unit_t&) { // initialize state
// nop [](unit_t&) {
}, // nop
// processing step },
[=](unit_t&, string y) { // processing step
self->state.strings.emplace_back(std::move(y)); [=](unit_t&, string y) {
}, self->state.strings.emplace_back(std::move(y));
// cleanup and produce result message },
[=](unit_t&, const error&) { // cleanup and produce result message
CAF_MESSAGE(self->name() << " is done"); [=](unit_t&, const error&) {
}); CAF_MESSAGE(self->name() << " is done");
}, });
[=](join_atom, actor src) { },
CAF_MESSAGE(self->name() << " joins a stream"); [=](join_atom, actor src) {
self->send(self * src, join_atom::value, strings_atom::value); CAF_MESSAGE(self->name() << " joins a stream");
}}; self->send(self * src, join_atom::value, strings_atom::value);
},
};
} }
using int_downstream_manager = broadcast_downstream_manager<int>; using int_downstream_manager = broadcast_downstream_manager<int>;
...@@ -210,28 +222,30 @@ TESTEE_STATE(stream_multiplexer) { ...@@ -210,28 +222,30 @@ TESTEE_STATE(stream_multiplexer) {
TESTEE(stream_multiplexer) { TESTEE(stream_multiplexer) {
self->state.stage = make_counted<fused_stage>(self); self->state.stage = make_counted<fused_stage>(self);
return {[=](join_atom, ints_atom) { return {
auto& stg = self->state.stage; [=](join_atom, ints_atom) {
CAF_MESSAGE("received 'join' request for integers"); auto& stg = self->state.stage;
auto result = stg->add_unchecked_outbound_path<int>(); CAF_MESSAGE("received 'join' request for integers");
stg->out().assign<int_downstream_manager>(result); auto result = stg->add_unchecked_outbound_path<int>();
return result; stg->out().assign<int_downstream_manager>(result);
}, return result;
[=](join_atom, strings_atom) { },
auto& stg = self->state.stage; [=](join_atom, strings_atom) {
CAF_MESSAGE("received 'join' request for strings"); auto& stg = self->state.stage;
auto result = stg->add_unchecked_outbound_path<string>(); CAF_MESSAGE("received 'join' request for strings");
stg->out().assign<string_downstream_manager>(result); auto result = stg->add_unchecked_outbound_path<string>();
return result; stg->out().assign<string_downstream_manager>(result);
}, return result;
[=](const stream<int32_t>& in) { },
CAF_MESSAGE("received handshake for integers"); [=](const stream<int32_t>& in) {
return self->state.stage->add_unchecked_inbound_path(in); CAF_MESSAGE("received handshake for integers");
}, return self->state.stage->add_unchecked_inbound_path(in);
[=](const stream<string>& in) { },
CAF_MESSAGE("received handshake for strings"); [=](const stream<string>& in) {
return self->state.stage->add_unchecked_inbound_path(in); CAF_MESSAGE("received handshake for strings");
}}; return self->state.stage->add_unchecked_inbound_path(in);
},
};
} }
struct config : actor_system_config { struct config : actor_system_config {
......
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