Commit 84285fb1 authored by Dominik Charousset's avatar Dominik Charousset

Fix formatting

parent 8607ad2f
...@@ -91,18 +91,20 @@ TESTEE(infinite_source) { ...@@ -91,18 +91,20 @@ TESTEE(infinite_source) {
} }
VARARGS_TESTEE(file_reader, size_t buf_size) { VARARGS_TESTEE(file_reader, size_t buf_size) {
return {[=](string& fname) -> result<stream<int>> { return {
CAF_CHECK_EQUAL(fname, "numbers.txt"); [=](string& fname) -> result<stream<int>> {
CAF_CHECK_EQUAL(self->mailbox().empty(), true); CAF_CHECK_EQUAL(fname, "numbers.txt");
return attach_stream_source(self, init(buf_size), push_from_buf, CAF_CHECK_EQUAL(self->mailbox().empty(), true);
is_done(self), fin<buf>(self)); return attach_stream_source(self, init(buf_size), push_from_buf,
}, is_done(self), fin<buf>(self));
[=](string& fname, actor next) { },
CAF_CHECK_EQUAL(fname, "numbers.txt"); [=](string& fname, actor next) {
CAF_CHECK_EQUAL(self->mailbox().empty(), true); CAF_CHECK_EQUAL(fname, "numbers.txt");
attach_stream_source(self, next, init(buf_size), push_from_buf, CAF_CHECK_EQUAL(self->mailbox().empty(), true);
is_done(self), fin<buf>(self)); attach_stream_source(self, next, init(buf_size), push_from_buf,
}}; is_done(self), fin<buf>(self));
},
};
} }
TESTEE_STATE(sum_up) { TESTEE_STATE(sum_up) {
...@@ -111,16 +113,18 @@ TESTEE_STATE(sum_up) { ...@@ -111,16 +113,18 @@ TESTEE_STATE(sum_up) {
TESTEE(sum_up) { TESTEE(sum_up) {
using intptr = int*; using intptr = int*;
return {[=](stream<int>& in) { return {
return attach_stream_sink( [=](stream<int>& 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, int y) { *x += y; }, fin<intptr>(self)); // processing step
}}; [](intptr& x, int y) { *x += y; }, fin<intptr>(self));
},
};
} }
TESTEE_STATE(delayed_sum_up) { TESTEE_STATE(delayed_sum_up) {
...@@ -130,67 +134,75 @@ TESTEE_STATE(delayed_sum_up) { ...@@ -130,67 +134,75 @@ TESTEE_STATE(delayed_sum_up) {
TESTEE(delayed_sum_up) { TESTEE(delayed_sum_up) {
using intptr = int*; using intptr = int*;
self->set_default_handler(skip); self->set_default_handler(skip);
return {[=](ok_atom) { return {
self->become([=](stream<int>& in) { [=](ok_atom) {
self->set_default_handler(print_and_drop); self->become([=](stream<int>& in) {
return attach_stream_sink( self->set_default_handler(print_and_drop);
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, int y) { *x += y; }, // processing step
// cleanup [](intptr& x, int y) { *x += y; },
fin<intptr>(self)); // cleanup
}); fin<intptr>(self));
}}; });
},
};
} }
TESTEE(broken_sink) { TESTEE(broken_sink) {
CAF_IGNORE_UNUSED(self); CAF_IGNORE_UNUSED(self);
return {[=](stream<int>&, const actor&) { return {
// nop [=](stream<int>&, const actor&) {
}}; // nop
},
};
} }
TESTEE(filter) { TESTEE(filter) {
CAF_IGNORE_UNUSED(self); CAF_IGNORE_UNUSED(self);
return {[=](stream<int>& in) { return {
return attach_stream_stage( [=](stream<int>& in) {
self, return attach_stream_stage(
// input stream self,
in, // input stream
// initialize state in,
[](unit_t&) { // initialize state
// nop [](unit_t&) {
}, // nop
// processing step },
[](unit_t&, downstream<int>& out, int x) { // processing step
if ((x & 0x01) != 0) [](unit_t&, downstream<int>& out, int x) {
out.push(x); if ((x & 0x01) != 0)
}, out.push(x);
// cleanup },
fin<unit_t>(self)); // cleanup
}}; fin<unit_t>(self));
},
};
} }
TESTEE(doubler) { TESTEE(doubler) {
CAF_IGNORE_UNUSED(self); CAF_IGNORE_UNUSED(self);
return {[=](stream<int>& in) { return {
return attach_stream_stage( [=](stream<int>& in) {
self, return attach_stream_stage(
// input stream self,
in, // input stream
// initialize state in,
[](unit_t&) { // initialize state
// nop [](unit_t&) {
}, // nop
// processing step },
[](unit_t&, downstream<int>& out, int x) { out.push(x * 2); }, // processing step
// cleanup [](unit_t&, downstream<int>& out, int x) { out.push(x * 2); },
fin<unit_t>(self)); // cleanup
}}; fin<unit_t>(self));
},
};
} }
struct fixture : test_coordinator_fixture<> { struct fixture : test_coordinator_fixture<> {
......
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