1 #include "../test.h"
2 #include <rxcpp/operators/rx-element_at.hpp>
3 
4 SCENARIO("element_at - never", "[element_at][operators]"){
5     GIVEN("a source"){
6         auto sc = rxsc::make_test();
7         auto w = sc.create_worker();
8         const rxsc::test::messages<int> on;
9 
10         auto xs = sc.make_hot_observable({
11             on.next(150, 1)
12         });
13 
14         WHEN("element_at is taken"){
15 
16             auto res = w.start(
__anon5c1a3f480102() 17                 [xs]() {
18                     return xs | rxo::element_at(3);
19                 }
20             );
21 
22             THEN("the output is empty"){
23                 auto required = std::vector<rxsc::test::messages<int>::recorded_type>();
24                 auto actual = res.get_observer().messages();
25                 REQUIRE(required == actual);
26             }
27 
28             THEN("there was 1 subscription/unsubscription to the source"){
29                 auto required = rxu::to_vector({
30                     on.subscribe(200, 1000)
31                 });
32                 auto actual = xs.subscriptions();
33                 REQUIRE(required == actual);
34             }
35         }
36     }
37 }
38 
39 SCENARIO("element_at - empty", "[element_at][operators]"){
40     GIVEN("a source"){
41         auto sc = rxsc::make_test();
42         auto w = sc.create_worker();
43         const rxsc::test::messages<int> on;
44 
45         std::runtime_error ex("element_at on_error from source");
46 
47         auto xs = sc.make_hot_observable({
48             on.next(150, 1),
49             on.completed(250)
50         });
51 
52         WHEN("element_at is taken"){
53 
54             auto res = w.start(
__anon5c1a3f480202() 55                 [xs]() {
56                     return xs.element_at(0);
57                 }
58             );
59 
60             THEN("the output only contains an error"){
61                 auto required = rxu::to_vector({
62                     on.error(250, ex)
63                 });
64                 auto actual = res.get_observer().messages();
65                 REQUIRE(required == actual);
66             }
67 
68             THEN("there was 1 subscription/unsubscription to the source"){
69                 auto required = rxu::to_vector({
70                     on.subscribe(200, 250)
71                 });
72                 auto actual = xs.subscriptions();
73                 REQUIRE(required == actual);
74             }
75 
76         }
77     }
78 }
79 
80 SCENARIO("element_at - first", "[element_at][operators]"){
81     GIVEN("a source"){
82         auto sc = rxsc::make_test();
83         auto w = sc.create_worker();
84         const rxsc::test::messages<int> on;
85 
86         auto xs = sc.make_hot_observable({
87             on.next(150, 1),
88             on.next(210, 2),
89             on.completed(250)
90         });
91 
92         WHEN("element_at is taken"){
93 
94             auto res = w.start(
__anon5c1a3f480302() 95                 [xs]() {
96                     return xs.element_at(0);
97                 }
98             );
99 
100             THEN("the output contains the first element"){
101                 auto required = rxu::to_vector({
102                     on.next(210, 2),
103                     on.completed(210)
104                 });
105                 auto actual = res.get_observer().messages();
106                 REQUIRE(required == actual);
107             }
108 
109             THEN("there was 1 subscription/unsubscription to the source"){
110                 auto required = rxu::to_vector({
111                     on.subscribe(200, 210)
112                 });
113                 auto actual = xs.subscriptions();
114                 REQUIRE(required == actual);
115             }
116 
117         }
118     }
119 }
120 
121 SCENARIO("element_at - throw", "[element_at][operators]"){
122     GIVEN("a source"){
123         auto sc = rxsc::make_test();
124         auto w = sc.create_worker();
125         const rxsc::test::messages<int> on;
126 
127         std::runtime_error ex("element_at on_error from source");
128 
129         auto xs = sc.make_hot_observable({
130             on.next(150, 1),
131             on.error(250, ex)
132         });
133 
134         WHEN("element_at is taken"){
135 
136             auto res = w.start(
__anon5c1a3f480402() 137                 [xs]() {
138                     return xs.element_at(3);
139                 }
140             );
141 
142             THEN("the output contains an error"){
143                 auto required = rxu::to_vector({
144                     on.error(250, ex)
145                 });
146                 auto actual = res.get_observer().messages();
147                 REQUIRE(required == actual);
148             }
149 
150             THEN("there was 1 subscription/unsubscription to the source"){
151                 auto required = rxu::to_vector({
152                     on.subscribe(200, 250)
153                 });
154                 auto actual = xs.subscriptions();
155                 REQUIRE(required == actual);
156             }
157 
158         }
159     }
160 }
161 
162 SCENARIO("element_at - non-first", "[element_at][operators]"){
163     GIVEN("a source"){
164         auto sc = rxsc::make_test();
165         auto w = sc.create_worker();
166         const rxsc::test::messages<int> on;
167 
168         auto xs = sc.make_hot_observable({
169             on.next(150, 1),
170             on.next(210, 2),
171             on.next(220, 3),
172             on.next(230, 4), //
173             on.next(240, 5),
174             on.completed(250)
175         });
176 
177         WHEN("element_at is taken"){
178 
179             auto res = w.start(
__anon5c1a3f480502() 180                 [xs]() {
181                     return xs.element_at(2);
182                 }
183             );
184 
185             THEN("the output contains the element at requested index"){
186                 auto required = rxu::to_vector({
187                     on.next(230, 4),
188                     on.completed(230)
189                 });
190                 auto actual = res.get_observer().messages();
191                 REQUIRE(required == actual);
192             }
193 
194             THEN("there was 1 subscription/unsubscription to the source"){
195                 auto required = rxu::to_vector({
196                     on.subscribe(200, 230)
197                 });
198                 auto actual = xs.subscriptions();
199                 REQUIRE(required == actual);
200             }
201 
202         }
203     }
204 }
205 
206 SCENARIO("element_at - last in a sequence", "[element_at][operators]"){
207     GIVEN("a source"){
208         auto sc = rxsc::make_test();
209         auto w = sc.create_worker();
210         const rxsc::test::messages<int> on;
211 
212         auto xs = sc.make_hot_observable({
213             on.next(150, 1),
214             on.next(210, 2),
215             on.next(220, 3),
216             on.next(230, 4),
217             on.next(240, 5), //
218             on.completed(250)
219         });
220 
221         WHEN("element_at is taken"){
222 
223             auto res = w.start(
__anon5c1a3f480602() 224                 [xs]() {
225                     return xs.element_at(3);
226                 }
227             );
228 
229             THEN("the output contains the element at requested index"){
230                 auto required = rxu::to_vector({
231                     on.next(240, 5),
232                     on.completed(240)
233                 });
234                 auto actual = res.get_observer().messages();
235                 REQUIRE(required == actual);
236             }
237 
238             THEN("there was 1 subscription/unsubscription to the source"){
239                 auto required = rxu::to_vector({
240                     on.subscribe(200, 240)
241                 });
242                 auto actual = xs.subscriptions();
243                 REQUIRE(required == actual);
244             }
245 
246         }
247     }
248 }
249 
250 SCENARIO("element_at - invalid index", "[element_at][operators]"){
251     GIVEN("a source"){
252         auto sc = rxsc::make_test();
253         auto w = sc.create_worker();
254         const rxsc::test::messages<int> on;
255 
256         std::runtime_error ex("element_at on_error from source");
257 
258         auto xs = sc.make_hot_observable({
259             on.next(150, 1),
260             on.next(210, 2), // #0
261             on.next(220, 3), // #1
262             on.next(230, 4), // #2
263             on.next(240, 5), // #3
264             on.completed(250)
265         });
266 
267         WHEN("element_at is taken"){
268 
269             auto res = w.start(
__anon5c1a3f480702() 270                 [xs]() {
271                     return xs.element_at(4);
272                 }
273             );
274 
275             THEN("the output contains an error"){
276                 auto required = rxu::to_vector({
277                     on.error(250, ex)
278                 });
279                 auto actual = res.get_observer().messages();
280                 REQUIRE(required == actual);
281             }
282 
283             THEN("there was 1 subscription/unsubscription to the source"){
284                 auto required = rxu::to_vector({
285                     on.subscribe(200, 250)
286                 });
287                 auto actual = xs.subscriptions();
288                 REQUIRE(required == actual);
289             }
290 
291         }
292     }
293 }
294