Commit a27358d3 authored by Dominik Charousset's avatar Dominik Charousset

Use endl to improve example usability on Windows

On Windows, STDOUT is "fully buffered" by default. This means that
messages don't appear after a newline on screen, like they do on UNIX
systems that use line buffering by default. Using `std::endl` makes sure
that written messages actually appear on screen for Windows users as
they should.
parent 72c484ad
...@@ -29,7 +29,7 @@ void caf_main(caf::actor_system& sys, const config& cfg) { ...@@ -29,7 +29,7 @@ void caf_main(caf::actor_system& sys, const config& cfg) {
// Only take the requested number of items from the infinite sequence. // Only take the requested number of items from the infinite sequence.
.take(n) .take(n)
// Print each integer. // Print each integer.
.for_each([](int x) { std::cout << x << '\n'; }); .for_each([](int x) { std::cout << x << std::endl; });
}); });
} }
// --(rst-main-end)-- // --(rst-main-end)--
......
...@@ -35,7 +35,7 @@ void caf_main(caf::actor_system& sys, const config& cfg) { ...@@ -35,7 +35,7 @@ void caf_main(caf::actor_system& sys, const config& cfg) {
// Switch to `snk` for further processing. // Switch to `snk` for further processing.
.observe_on(snk) .observe_on(snk)
// Print each integer. // Print each integer.
.for_each([](int x) { std::cout << x << '\n'; }); .for_each([](int x) { std::cout << x << std::endl; });
// Allow the actors to run. After this point, we may no longer dereference // Allow the actors to run. After this point, we may no longer dereference
// the `src` and `snk` pointers! Calling these manually is optional. When // the `src` and `snk` pointers! Calling these manually is optional. When
// removing these two lines, CAF automatically launches the actors at scope // removing these two lines, CAF automatically launches the actors at scope
......
...@@ -37,7 +37,7 @@ void sink(caf::event_based_actor* self, caf::async::consumer_resource<int> in) { ...@@ -37,7 +37,7 @@ void sink(caf::event_based_actor* self, caf::async::consumer_resource<int> in) {
// Lift the input to an observable flow. // Lift the input to an observable flow.
.from_resource(std::move(in)) .from_resource(std::move(in))
// Print each integer. // Print each integer.
.for_each([](int x) { std::cout << x << '\n'; }); .for_each([](int x) { std::cout << x << std::endl; });
} }
// --(rst-sink-end)-- // --(rst-sink-end)--
......
...@@ -77,7 +77,8 @@ int caf_main(caf::actor_system& sys, const config& cfg) { ...@@ -77,7 +77,8 @@ int caf_main(caf::actor_system& sys, const config& cfg) {
} }
// Note: the actor system will only wait for actors on default. Since we don't // Note: the actor system will only wait for actors on default. Since we don't
// start actors, we need to block on something else. // start actors, we need to block on something else.
std::cout << "Server is up and running. Press <enter> to shut down.\n"; std::cout << "Server is up and running. Press <enter> to shut down."
<< std::endl;
getchar(); getchar();
std::cout << "Terminating.\n"; std::cout << "Terminating.\n";
return EXIT_SUCCESS; return EXIT_SUCCESS;
......
...@@ -70,11 +70,13 @@ int caf_main(caf::actor_system& sys, const config& cfg) { ...@@ -70,11 +70,13 @@ int caf_main(caf::actor_system& sys, const config& cfg) {
pull pull
.observe_on(self) // .observe_on(self) //
.do_on_error([](const caf::error& err) { .do_on_error([](const caf::error& err) {
std::cout << "*** connection error: " << to_string(err) << '\n'; std::cout << "*** connection error: " << to_string(err)
<< std::endl;
}) })
.do_finally([self] { .do_finally([self] {
std::cout << "*** lost connection to server -> quit\n" std::cout << "*** lost connection to server -> quit\n"
<< "*** use CTRL+D or CTRL+C to terminate\n"; "*** use CTRL+D or CTRL+C to terminate"
<< std::endl;
self->quit(); self->quit();
}) })
.for_each([](const lp::frame& frame) { .for_each([](const lp::frame& frame) {
...@@ -83,10 +85,10 @@ int caf_main(caf::actor_system& sys, const config& cfg) { ...@@ -83,10 +85,10 @@ int caf_main(caf::actor_system& sys, const config& cfg) {
auto str = std::string_view{ auto str = std::string_view{
reinterpret_cast<const char*>(bytes.data()), bytes.size()}; reinterpret_cast<const char*>(bytes.data()), bytes.size()};
if (std::all_of(str.begin(), str.end(), ::isprint)) { if (std::all_of(str.begin(), str.end(), ::isprint)) {
std::cout << str << '\n'; std::cout << str << std::endl;
} else { } else {
std::cout << "<non-ascii-data of size " << bytes.size() std::cout << "<non-ascii-data of size " << bytes.size() << ">"
<< ">\n"; << std::endl;
} }
}); });
}); });
......
...@@ -49,7 +49,7 @@ void worker_impl(caf::event_based_actor* self, ...@@ -49,7 +49,7 @@ void worker_impl(caf::event_based_actor* self,
messages.for_each([](const message_t& msg) { messages.for_each([](const message_t& msg) {
const auto& [conn, frame] = msg; const auto& [conn, frame] = msg;
std::cout << "*** got message of size " << frame.size() << " from " std::cout << "*** got message of size " << frame.size() << " from "
<< to_string(conn) << '\n'; << to_string(conn) << std::endl;
}); });
// Connect the flows for each incoming connection. // Connect the flows for each incoming connection.
events events
...@@ -58,7 +58,8 @@ void worker_impl(caf::event_based_actor* self, ...@@ -58,7 +58,8 @@ void worker_impl(caf::event_based_actor* self,
[self, messages, pub = std::move(pub)](const auto& event) mutable { [self, messages, pub = std::move(pub)](const auto& event) mutable {
// Each connection gets a unique ID. // Each connection gets a unique ID.
auto conn = caf::uuid::random(); auto conn = caf::uuid::random();
std::cout << "*** accepted new connection " << to_string(conn) << '\n'; std::cout << "*** accepted new connection " << to_string(conn)
<< std::endl;
auto& [pull, push] = event.data(); auto& [pull, push] = event.data();
// Subscribe the `push` end to the central merge point. // Subscribe the `push` end to the central merge point.
messages messages
...@@ -72,14 +73,16 @@ void worker_impl(caf::event_based_actor* self, ...@@ -72,14 +73,16 @@ void worker_impl(caf::event_based_actor* self,
}) })
.subscribe(push); .subscribe(push);
// Feed messages from the `pull` end into the central merge point. // Feed messages from the `pull` end into the central merge point.
auto inputs auto inputs //
= pull.observe_on(self) = pull.observe_on(self)
.do_on_error([](const caf::error& err) { .do_on_error([](const caf::error& err) {
std::cout << "*** connection error: " << to_string(err) << '\n'; std::cout << "*** connection error: " << to_string(err)
<< std::endl;
}) })
.on_error_complete() // Cary on if a connection breaks. .on_error_complete() // Cary on if a connection breaks.
.do_on_complete([conn] { .do_on_complete([conn] {
std::cout << "*** lost connection " << to_string(conn) << '\n'; std::cout << "*** lost connection " << to_string(conn)
<< std::endl;
}) })
.map([conn](const lp::frame& frame) { .map([conn](const lp::frame& frame) {
return message_t{conn, frame}; return message_t{conn, frame};
......
...@@ -142,17 +142,17 @@ void caf_main(actor_system& sys) { ...@@ -142,17 +142,17 @@ void caf_main(actor_system& sys) {
for (int row = 0; row < rows; ++row) { for (int row = 0; row < rows; ++row) {
for (int column = 0; column < columns; ++column) for (int column = 0; column < columns; ++column)
cout << std::setw(4) << f(get_atom_v, row, column) << ' '; cout << std::setw(4) << f(get_atom_v, row, column) << ' ';
cout << '\n'; cout << std::endl;
} }
// Print out AVG for each row and column. // Print out AVG for each row and column.
for (int row = 0; row < rows; ++row) for (int row = 0; row < rows; ++row)
cout << "AVG(row " << row << ") = " cout << "AVG(row " << row << ") = "
<< caf::to_string(f(get_atom_v, average_atom_v, row_atom_v, row)) << caf::to_string(f(get_atom_v, average_atom_v, row_atom_v, row))
<< '\n'; << std::endl;
for (int column = 0; column < columns; ++column) for (int column = 0; column < columns; ++column)
cout << "AVG(column " << column << ") = " cout << "AVG(column " << column << ") = "
<< caf::to_string(f(get_atom_v, average_atom_v, column_atom_v, column)) << caf::to_string(f(get_atom_v, average_atom_v, column_atom_v, column))
<< '\n'; << std::endl;
} }
CAF_MAIN(id_block::fan_out_request) CAF_MAIN(id_block::fan_out_request)
...@@ -96,7 +96,7 @@ void client_repl(function_view<calculator> f) { ...@@ -96,7 +96,7 @@ void client_repl(function_view<calculator> f) {
cout << " = " cout << " = "
<< caf::to_string(words[1] == "+" ? f(add_atom_v, *x, *y) << caf::to_string(words[1] == "+" ? f(add_atom_v, *x, *y)
: f(sub_atom_v, *x, *y)) : f(sub_atom_v, *x, *y))
<< "\n"; << std::endl;
} }
} }
......
...@@ -66,7 +66,7 @@ int caf_main(caf::actor_system& sys, const config& cfg) { ...@@ -66,7 +66,7 @@ int caf_main(caf::actor_system& sys, const config& cfg) {
// The header parameter contains fields from the WebSocket handshake // The header parameter contains fields from the WebSocket handshake
// such as the path and HTTP header fields.. // such as the path and HTTP header fields..
auto path = acc.header().path(); auto path = acc.header().path();
std::cout << "*** new client request for path " << path << '\n'; std::cout << "*** new client request for path " << path << std::endl;
// Accept the WebSocket connection only if the path is "/". // Accept the WebSocket connection only if the path is "/".
if (path == "/") { if (path == "/") {
// Calling `accept` causes the server to acknowledge the client and // Calling `accept` causes the server to acknowledge the client and
...@@ -95,20 +95,20 @@ int caf_main(caf::actor_system& sys, const config& cfg) { ...@@ -95,20 +95,20 @@ int caf_main(caf::actor_system& sys, const config& cfg) {
pull.observe_on(self) pull.observe_on(self)
.do_on_error([](const caf::error& what) { // .do_on_error([](const caf::error& what) { //
std::cout << "*** connection closed: " << to_string(what) std::cout << "*** connection closed: " << to_string(what)
<< "\n"; << std::endl;
}) })
.do_on_complete([] { // .do_on_complete([] { //
std::cout << "*** connection closed\n"; std::cout << "*** connection closed" << std::endl;
}) })
.do_on_next([](const ws::frame& x) { .do_on_next([](const ws::frame& x) {
if (x.is_binary()) { if (x.is_binary()) {
std::cout std::cout
<< "*** received a binary WebSocket frame of size " << "*** received a binary WebSocket frame of size "
<< x.size() << '\n'; << x.size() << std::endl;
} else { } else {
std::cout std::cout
<< "*** received a text WebSocket frame of size " << "*** received a text WebSocket frame of size "
<< x.size() << '\n'; << x.size() << std::endl;
} }
}) })
.subscribe(push); .subscribe(push);
......
...@@ -52,7 +52,7 @@ int caf_main(caf::actor_system& sys, const config& cfg) { ...@@ -52,7 +52,7 @@ int caf_main(caf::actor_system& sys, const config& cfg) {
// Print errors to stderr. // Print errors to stderr.
.do_on_error([](const caf::error& what) { .do_on_error([](const caf::error& what) {
std::cerr << "*** error while reading from the WebSocket: " std::cerr << "*** error while reading from the WebSocket: "
<< to_string(what) << '\n'; << to_string(what) << std::endl;
}) })
// Restrict how many messages we receive if the user configured // Restrict how many messages we receive if the user configured
// a limit. // a limit.
...@@ -65,15 +65,15 @@ int caf_main(caf::actor_system& sys, const config& cfg) { ...@@ -65,15 +65,15 @@ int caf_main(caf::actor_system& sys, const config& cfg) {
}) })
// Print a bye-bye message if the server closes the connection. // Print a bye-bye message if the server closes the connection.
.do_on_complete([] { // .do_on_complete([] { //
std::cout << "Server has closed the connection\n"; std::cout << "Server has closed the connection" << std::endl;
}) })
// Print everything from the server to stdout. // Print everything from the server to stdout.
.for_each([](const ws::frame& msg) { .for_each([](const ws::frame& msg) {
if (msg.is_text()) { if (msg.is_text()) {
std::cout << "Server: " << msg.as_text() << '\n'; std::cout << "Server: " << msg.as_text() << std::endl;
} else if (msg.is_binary()) { } else if (msg.is_binary()) {
std::cout << "Server: [binary message of size " std::cout << "Server: [binary message of size "
<< msg.as_binary().size() << "]\n"; << msg.as_binary().size() << "]" << std::endl;
} }
}); });
// Send our hello message and wait until the server closes the // Send our hello message and wait until the server closes the
...@@ -87,7 +87,7 @@ int caf_main(caf::actor_system& sys, const config& cfg) { ...@@ -87,7 +87,7 @@ int caf_main(caf::actor_system& sys, const config& cfg) {
}); });
if (!conn) { if (!conn) {
std::cerr << "*** unable to connect to " << server->str() << ": " std::cerr << "*** unable to connect to " << server->str() << ": "
<< to_string(conn.error()) << '\n'; << to_string(conn.error()) << std::endl;
return EXIT_FAILURE; return EXIT_FAILURE;
} }
// Note: the actor system will keep the application running for as long as the // Note: the actor system will keep the application running for as long as the
......
...@@ -83,7 +83,7 @@ struct random_feed_state { ...@@ -83,7 +83,7 @@ struct random_feed_state {
auto& x = update(); auto& x = update();
if (!writer.apply(x)) { if (!writer.apply(x)) {
std::cerr << "*** failed to generate JSON: " std::cerr << "*** failed to generate JSON: "
<< to_string(writer.get_error()) << '\n'; << to_string(writer.get_error()) << std::endl;
return frame{}; return frame{};
} }
return frame{writer.str()}; return frame{writer.str()};
...@@ -95,18 +95,19 @@ struct random_feed_state { ...@@ -95,18 +95,19 @@ struct random_feed_state {
.share(); .share();
// Subscribe once to start the feed immediately and to keep it running. // Subscribe once to start the feed immediately and to keep it running.
feed.for_each([n = 1](const frame&) mutable { feed.for_each([n = 1](const frame&) mutable {
std::cout << "*** tick " << n++ << "\n"; std::cout << "*** tick " << n++ << std::endl;
}); });
// Add each incoming WebSocket listener to the feed. // Add each incoming WebSocket listener to the feed.
auto n = std::make_shared<int>(0); auto n = std::make_shared<int>(0);
events events
.observe_on(self) // .observe_on(self) //
.for_each([this, n](const accept_event& ev) { .for_each([this, n](const accept_event& ev) {
std::cout << "*** added listener (n = " << ++*n << ")\n"; std::cout << "*** added listener (n = " << ++*n << ")" << std::endl;
auto [pull, push] = ev.data(); auto [pull, push] = ev.data();
pull.observe_on(self) pull.observe_on(self)
.do_finally([n] { // .do_finally([n] { //
std::cout << "*** removed listener (n = " << --*n << ")\n"; std::cout << "*** removed listener (n = " << --*n << ")"
<< std::endl;
}) })
.subscribe(std::ignore); .subscribe(std::ignore);
feed.subscribe(push); feed.subscribe(push);
......
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