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
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
|
#include "rxcpp/rx.hpp"
#include "rxcpp/rx-test.hpp"
#include "catch.hpp"
SCENARIO("range sample"){
printf("//! [range sample]\n");
auto values1 = rxcpp::observable<>::range(1, 5);
values1.
subscribe(
[](int v){printf("OnNext: %d\n", v);},
[](){printf("OnCompleted\n");});
printf("//! [range sample]\n");
}
#include "main.hpp"
SCENARIO("threaded range sample"){
printf("//! [threaded range sample]\n");
printf("[thread %s] Start task\n", get_pid().c_str());
auto values = rxcpp::observable<>::range(1, 3, rxcpp::observe_on_new_thread());
auto s = values.
map([](int v) { return std::make_tuple(get_pid(), v);});
s.
as_blocking().
subscribe(
rxcpp::util::apply_to(
[](const std::string pid, int v) {
printf("[thread %s] OnNext: %d\n", pid.c_str(), v);
}),
[](){printf("[thread %s] OnCompleted\n", get_pid().c_str());});
printf("[thread %s] Finish task\n", get_pid().c_str());
printf("//! [threaded range sample]\n");
}
SCENARIO("subscribe_on range sample"){
printf("//! [subscribe_on range sample]\n");
printf("[thread %s] Start task\n", get_pid().c_str());
auto values = rxcpp::observable<>::range(1, 3);
auto s = values.
subscribe_on(rxcpp::observe_on_new_thread()).
map([](int v) { return std::make_tuple(get_pid(), v);});
s.
as_blocking().
subscribe(
rxcpp::util::apply_to(
[](const std::string pid, int v) {
printf("[thread %s] OnNext: %d\n", pid.c_str(), v);
}),
[](){printf("[thread %s] OnCompleted\n", get_pid().c_str());});
printf("[thread %s] Finish task\n", get_pid().c_str());
printf("//! [subscribe_on range sample]\n");
}
SCENARIO("range concat sample"){
printf("//! [range concat sample]\n");
auto values = rxcpp::observable<>::range(1); // infinite (until overflow) stream of integers
auto s1 = values.
take(3).
map([](int v) { return std::make_tuple("1:", v);});
auto s2 = values.
take(3).
map([](int v) { return std::make_tuple("2:", v);});
s1.
concat(s2).
subscribe(rxcpp::util::apply_to(
[](const char* s, int p) {
printf("%s %d\n", s, p);
}));
printf("//! [range concat sample]\n");
}
SCENARIO("range merge sample"){
printf("//! [range merge sample]\n");
auto values = rxcpp::observable<>::range(1); // infinite (until overflow) stream of integers
auto s1 = values.
map([](int v) { return std::make_tuple("1:", v);});
auto s2 = values.
map([](int v) { return std::make_tuple("2:", v);});
s1.
merge(s2).
take(6).
as_blocking().
subscribe(rxcpp::util::apply_to(
[](const char* s, int p) {
printf("%s %d\n", s, p);
}));
printf("//! [range merge sample]\n");
}
SCENARIO("threaded range concat sample"){
printf("//! [threaded range concat sample]\n");
auto threads = rxcpp::observe_on_event_loop();
auto values = rxcpp::observable<>::range(1); // infinite (until overflow) stream of integers
auto s1 = values.
subscribe_on(threads).
take(3).
map([](int v) { std::this_thread::yield(); return std::make_tuple("1:", v);});
auto s2 = values.
subscribe_on(threads).
take(3).
map([](int v) { std::this_thread::yield(); return std::make_tuple("2:", v);});
s1.
concat(s2).
observe_on(threads).
as_blocking().
subscribe(rxcpp::util::apply_to(
[](const char* s, int p) {
printf("%s %d\n", s, p);
}));
printf("//! [threaded range concat sample]\n");
}
SCENARIO("threaded range merge sample"){
printf("//! [threaded range merge sample]\n");
auto threads = rxcpp::observe_on_event_loop();
auto values = rxcpp::observable<>::range(1); // infinite (until overflow) stream of integers
auto s1 = values.
subscribe_on(threads).
map([](int v) { std::this_thread::yield(); return std::make_tuple("1:", v);});
auto s2 = values.
subscribe_on(threads).
map([](int v) { std::this_thread::yield(); return std::make_tuple("2:", v);});
s1.
merge(s2).
take(6).
observe_on(threads).
as_blocking().
subscribe(rxcpp::util::apply_to(
[](const char* s, int p) {
printf("%s %d\n", s, p);
}));
printf("//! [threaded range merge sample]\n");
}
|
