1 #include "rxcpp/rx.hpp" 2 3 #include "rxcpp/rx-test.hpp" 4 #include "catch.hpp" 5 6 SCENARIO("flat_map sample"){ 7 printf("//! [flat_map sample]\n"); 8 auto values = rxcpp::observable<>::range(1, 3). 9 flat_map( __anonc1b991d30102(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 }, __anonc1b991d30202(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( __anonc1b991d30302(std::tuple<int, long> v)20 [](std::tuple<int, long> v){printf("OnNext: %d - %ld\n", std::get<0>(v), std::get<1>(v));}, __anonc1b991d30402()21 [](){printf("OnCompleted\n");}); 22 printf("//! [flat_map sample]\n"); 23 } 24 25 #include "main.hpp" 26 27 SCENARIO("threaded flat_map sample"){ 28 printf("//! [threaded flat_map sample]\n"); 29 printf("[thread %s] Start task\n", get_pid().c_str()); 30 auto values = rxcpp::observable<>::range(1, 3). 31 flat_map( __anonc1b991d30502(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 }, __anonc1b991d30602(int v_main, int v_sub)38 [](int v_main, int v_sub){ 39 printf("[thread %s] Call ResultSelector(v_main = %d, v_sub = %d)\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( __anonc1b991d30702(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));}, __anonc1b991d30802()47 [](){printf("[thread %s] OnCompleted\n", get_pid().c_str());}); 48 printf("[thread %s] Finish task\n", get_pid().c_str()); 49 printf("//! [threaded flat_map sample]\n"); 50 } 51