RxCpp 2 and API
The last article on rxcpp was based on a now obsolete version of RxCpp. The key contributor to the library, Kirk Shoop, has kindly provided a rewrite based on the newer, 2.0 API of the library: see the pull request, upon which this article is based.
Since the first article, the project has been enriched with somewhat more readable GIVEN/WHEN/THEN-style tests using Catch [1. i.e. create.cpp].
Still Ticking: Scheduler and Coordination in RxCpp 2
The previous articles give examples of managing periodic events, such as ticker ticks and measurements in c++. The following example creates an event loop that will be used for coordinated output of various events to the console:
auto scheduler = rxcpp::schedulers::make_same_worker(
rxcpp::schedulers::make_event_loop().create_worker()
);
auto coordination = rxcpp::identity_one_worker(scheduler);
One such sequence of events is some kind of measurement [2. Observe the convergence of the API towards the C# version.]
auto measure = rxcpp::observable<>::interval(
// when to start
scheduler.now() + std::chrono::milliseconds(250),
// measurement frequency
std::chrono::milliseconds(250),
coordination)
// take Hz values instead of a counter
.map([&FM](int) { return FM.Hz(); });
auto measure_subscription = measure
.subscribe([](int val) {
std::cout << val << std::endl;
});
Why didn’t it tick?
If this code were the end of the main program, there wouldn’t be any observable ticks, as all the objects would be destroyed before the first scheduled event. To see the code in action, we shall wait for some condition that will change when we’re done. This step is not necessary if there’s a GUI toolkit event loop that keeps objects alive, but it has to be simulated for a console example.
To demonstrate the subscription change and wait for some time, we’ll wait twice for an atomic variable to become zero:
std::atomic<long> pending(2);
...
// after all subscriptions defined
while (pending) {
sleep(1000); // wait for ticker and measure to finish
}
Tick and Stop
The other ticker will have another period, will only tick 10 times, and then decrement the pending counter:
auto ticker = rxcpp::observable<>::interval(
scheduler.now() + std::chrono::milliseconds(500),
std::chrono::milliseconds(500),
coordination);
ticker
.take(10)
.subscribe([](int val) {
std::cout << "tick " << val << std::endl;
},[&](){
--pending; // take completed the ticker
});
Now, we can schedule the termination of the measurement (decrement pending) subscription halfway through the 10-tick run. This scheduling is done on the same scheduler that is running all the subscriptions:
scheduler.create_worker().schedule(scheduler.now() + std::chrono::seconds(2),
[&](const rxcpp::schedulers::schedulable&) {
std::cout << "Canceling measurement ..." << std::endl;
measure_subscription.unsubscribe(); // cancel measurement
--pending; // signal measurement canceled
});
The result:
63 tick 1 63 61 tick 2 63 61 tick 3 63 62 Canceling measurement ... tick 4 tick 5 tick 6 tick 7 tick 8 tick 9 tick 10
Thanks, Kirk & other library contributors!