Commit 90fcde67 authored by Dominik Charousset's avatar Dominik Charousset

Expand flow API examples

parent 36a7a16f
......@@ -33,6 +33,8 @@ add_core_example(message_passing request)
add_core_example(message_passing typed_calculator)
# flow API
add_core_example(flow from-callable)
add_core_example(flow observe-on)
add_core_example(flow spsc-buffer-resource)
# dynamic behavior changes using 'become'
......
// Non-interactive example to showcase `from_callable`.
#include "caf/actor_system.hpp"
#include "caf/caf_main.hpp"
#include "caf/event_based_actor.hpp"
#include "caf/scheduled_actor/flow.hpp"
#include <iostream>
namespace {
struct config : caf::actor_system_config {
config() {
opt_group{custom_options_, "global"} //
.add(n, "num-values,n", "number of values produced by the source");
}
size_t n = 10;
};
// --(rst-from-callable-begin)--
void caf_main(caf::actor_system& sys, const config& cfg) {
sys.spawn([n = cfg.n](caf::event_based_actor* self) {
self
// Get an observable factory.
->make_observable()
// Produce an integer sequence starting at 1, i.e., 1, 2, 3, ...
.from_callable([i = 0]() mutable { return ++i; })
// Only take the requested number of items from the infinite sequence.
.take(n)
// Print each integer.
.for_each([](int x) { std::cout << x << '\n'; });
});
}
// --(rst-from-callable-end)--
} // namespace
CAF_MAIN()
// Non-interactive example to showcase `observe_on`.
#include "caf/actor_system.hpp"
#include "caf/caf_main.hpp"
#include "caf/event_based_actor.hpp"
#include "caf/scheduled_actor/flow.hpp"
#include <iostream>
namespace {
struct config : caf::actor_system_config {
config() {
opt_group{custom_options_, "global"} //
.add(n, "num-values,n", "number of values produced by the source");
}
size_t n = 10;
};
// --(rst-from-callable-begin)--
void caf_main(caf::actor_system& sys, const config& cfg) {
// Create two actors without actually running them yet.
using actor_t = caf::event_based_actor;
auto [src, launch_src] = sys.spawn_inactive<actor_t>();
auto [snk, launch_snk] = sys.spawn_inactive<actor_t>();
// Define our data flow: generate data on `src` and print it on `snk`.
src
// Get an observable factory.
->make_observable()
// Produce an integer sequence starting at 1, i.e., 1, 2, 3, ...
.from_callable([i = 0]() mutable { return ++i; })
// Only take the requested number of items from the infinite sequence.
.take(cfg.n)
// Switch to `snk` for further processing.
.observe_on(snk)
// Print each integer.
.for_each([](int x) { std::cout << x << '\n'; });
// Allow the actors to run. After this point, we may no longer dereference
// the `src` and `snk` pointers! Calling these manually is optional. When
// removing these two lines, CAF automatically launches the actors at scope
// exit.
launch_src();
launch_snk();
}
// --(rst-from-callable-end)--
} // namespace
CAF_MAIN()
// Non-interactive example to illustrate how to connect flows over an SPSC
// (Single Producer Single Consumer) buffer.
// Non-interactive example to illustrate how to connect flows over an
// asynchronous SPSC (Single Producer Single Consumer) buffer.
#include "caf/actor_system.hpp"
#include "caf/async/spsc_buffer.hpp"
......@@ -13,8 +13,8 @@ namespace {
// --(rst-source-begin)--
// Simple source for generating a stream of integers from 1 to n.
void int_source(caf::event_based_actor* self,
caf::async::producer_resource<int> out, size_t n) {
void source(caf::event_based_actor* self,
caf::async::producer_resource<int> out, size_t n) {
self
// Get an observable factory.
->make_observable()
......@@ -29,8 +29,7 @@ void int_source(caf::event_based_actor* self,
// --(rst-sink-begin)--
// Simple sink for consuming a stream of integers, printing it to stdout.
void int_sink(caf::event_based_actor* self,
caf::async::consumer_resource<int> in) {
void sink(caf::event_based_actor* self, caf::async::consumer_resource<int> in) {
self
// Get an observable factory.
->make_observable()
......@@ -53,8 +52,8 @@ struct config : caf::actor_system_config {
// --(rst-main-begin)--
void caf_main(caf::actor_system& sys, const config& cfg) {
auto [snk_res, src_res] = caf::async::make_spsc_buffer_resource<int>();
sys.spawn(int_sink, std::move(snk_res));
sys.spawn(int_source, std::move(src_res), cfg.n);
sys.spawn(sink, std::move(snk_res));
sys.spawn(source, std::move(src_res), cfg.n);
}
// --(rst-main-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