1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
|
#include "rxcpp/rx.hpp"
#include "rxcpp/rx-test.hpp"
#include "catch.hpp"
SCENARIO("flat_map sample"){
printf("//! [flat_map sample]\n");
auto values = rxcpp::observable<>::range(1, 3).
flat_map(
[](int v){
return
rxcpp::observable<>::interval(std::chrono::steady_clock::now() + std::chrono::milliseconds(10 * v), std::chrono::milliseconds(50)).
take(3);
},
[](int v_main, long v_sub){
return std::make_tuple(v_main, v_sub);
});
values.
subscribe(
[](std::tuple<int, long> v){printf("OnNext: %d - %ld\n", std::get<0>(v), std::get<1>(v));},
[](){printf("OnCompleted\n");});
printf("//! [flat_map sample]\n");
}
std::string get_pid();
SCENARIO("threaded flat_map sample"){
printf("//! [threaded flat_map sample]\n");
printf("[thread %s] Start task\n", get_pid().c_str());
auto values = rxcpp::observable<>::range(1, 3).
flat_map(
[](int v){
printf("[thread %s] Call CollectionSelector(v = %d)\n", get_pid().c_str(), v);
return
rxcpp::observable<>::interval(std::chrono::steady_clock::now() + std::chrono::milliseconds(10 * v), std::chrono::milliseconds(50)).
take(3);
},
[](int v_main, int v_sub){
printf("[thread %s] Call ResultSelector(v_main = %d, v_sub = %d)\n", get_pid().c_str(), v_main, v_sub);
return std::make_tuple(v_main, v_sub);
},
rxcpp::observe_on_new_thread());
values.
as_blocking().
subscribe(
[](std::tuple<int, long> v){printf("[thread %s] OnNext: %d - %ld\n", get_pid().c_str(), std::get<0>(v), std::get<1>(v));},
[](){printf("[thread %s] OnCompleted\n", get_pid().c_str());});
printf("[thread %s] Finish task\n", get_pid().c_str());
printf("//! [threaded flat_map sample]\n");
}
|
