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