RxCpp
The Reactive Extensions for Native (RxCpp) is a library for composing asynchronous and event-based programs using observable sequences and LINQ-style query operators in both C and C++.
Classes | Namespaces | Macros | Functions
rx-flat_map.hpp File Reference

For each item from this observable use the CollectionSelector to produce an observable and subscribe to that observable. For each item from all of the produced observables use the ResultSelector to produce a value to emit from the new observable that is returned. More...

#include "../rx-includes.hpp"
Include dependency graph for rx-flat_map.hpp:
This graph shows which files directly or indirectly include this file:

Go to the source code of this file.

Classes

struct  rxcpp::member_overload< flat_map_tag >
 

Namespaces

 rxcpp
 
 rxcpp::operators
 

Macros

#define RXCPP_OPERATORS_RX_FLATMAP_HPP
 

Functions

template<class... AN>
auto rxcpp::operators::flat_map (AN &&... an) -> operator_factory< flat_map_tag, AN... >
 
template<class... AN>
auto rxcpp::operators::merge_transform (AN &&... an) -> operator_factory< flat_map_tag, AN... >
 

Detailed Description

For each item from this observable use the CollectionSelector to produce an observable and subscribe to that observable. For each item from all of the produced observables use the ResultSelector to produce a value to emit from the new observable that is returned.

Template Parameters
CollectionSelectorthe type of the observable producing function. CollectionSelector must be a function with the signature observable(flat_map::source_value_type)
ResultSelectorthe type of the aggregation function (optional). ResultSelector must be a function with the signature flat_map::value_type(flat_map::source_value_type, flat_map::collection_value_type).
Coordinationthe type of the scheduler (optional).
Parameters
sa function that returns an observable for each item emitted by the source observable.
rsa function that combines one item emitted by each of the source and collection observables and returns an item to be emitted by the resulting observable (optional).
cnthe scheduler to synchronize sources from different contexts (optional).
Returns
Observable that emits the results of applying a function to a pair of values emitted by the source observable and the collection observable.

Observables, produced by the CollectionSelector, are merged. There is another operator rxcpp::observable<T,SourceType>::flat_map that works similar but concatenates the observables.

Sample Code\n
auto values = rxcpp::observable<>::range(1, 3).
[](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.
[](std::tuple<int, long> v){printf("OnNext: %d - %ld\n", std::get<0>(v), std::get<1>(v));},
[](){printf("OnCompleted\n");});
OnNext: 1 - 1
OnNext: 2 - 1
OnNext: 3 - 1
OnNext: 1 - 2
OnNext: 2 - 2
OnNext: 3 - 2
OnNext: 1 - 3
OnNext: 2 - 3
OnNext: 3 - 3
OnCompleted
Sample Code\n
printf("[thread %s] Start task\n", get_pid().c_str());
auto values = rxcpp::observable<>::range(1, 3).
[](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);
},
values.
[](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());
[thread 3070107664] Start task
[thread 3015701568] Call CollectionSelector(v = 1)
[thread 3015701568] Call CollectionSelector(v = 2)
[thread 3015701568] Call CollectionSelector(v = 3)
[thread 3015701568] Call ResultSelector(v_main = 1, v_sub = 1)
[thread 3015701568] OnNext: 1 - 1
[thread 3015701568] Call ResultSelector(v_main = 2, v_sub = 1)
[thread 3015701568] OnNext: 2 - 1
[thread 3015701568] Call ResultSelector(v_main = 3, v_sub = 1)
[thread 3015701568] OnNext: 3 - 1
[thread 3015701568] Call ResultSelector(v_main = 1, v_sub = 2)
[thread 3015701568] OnNext: 1 - 2
[thread 3015701568] Call ResultSelector(v_main = 2, v_sub = 2)
[thread 3015701568] OnNext: 2 - 2
[thread 3015701568] Call ResultSelector(v_main = 3, v_sub = 2)
[thread 3015701568] OnNext: 3 - 2
[thread 3015701568] Call ResultSelector(v_main = 1, v_sub = 3)
[thread 3015701568] OnNext: 1 - 3
[thread 3015701568] Call ResultSelector(v_main = 2, v_sub = 3)
[thread 3015701568] OnNext: 2 - 3
[thread 3015701568] Call ResultSelector(v_main = 3, v_sub = 3)
[thread 3015701568] OnNext: 3 - 3
[thread 3015701568] OnCompleted
[thread 3070107664] Finish task

Macro Definition Documentation

◆ RXCPP_OPERATORS_RX_FLATMAP_HPP

#define RXCPP_OPERATORS_RX_FLATMAP_HPP
rxcpp::sources::range
auto range(T first=0, T last=std::numeric_limits< T >::max(), std::ptrdiff_t step=1) -> observable< T, detail::range< T, identity_one_worker >>
Definition: rx-range.hpp:119
rxcpp::observe_on_new_thread
observe_on_one_worker observe_on_new_thread()
Definition: rx-observe_on.hpp:328
rxcpp::operators::flat_map
auto flat_map(AN &&... an) -> operator_factory< flat_map_tag, AN... >
Definition: rx-flat_map.hpp:244
rxcpp::operators::as_blocking
auto as_blocking() -> detail::blocking_factory
Definition: rx-subscribe.hpp:144
rxcpp::operators::take
auto take(AN &&... an) -> operator_factory< take_tag, AN... >
Definition: rx-take.hpp:133
rxcpp::operators::subscribe
auto subscribe(ArgN &&... an) -> detail::subscribe_factory< decltype(make_subscriber< T >(std::forward< ArgN >(an)...))>
Definition: rx-subscribe.hpp:87
rxcpp::sources::interval
auto interval(Duration period) -> typename std::enable_if< detail::defer_interval< Duration, identity_one_worker >::value, typename detail::defer_interval< Duration, identity_one_worker >::observable_type >::type
Definition: rx-interval.hpp:113