rx-concat_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-concat_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< concat_map_tag >

Namespaces
	rxcpp
	rxcpp::operators

Macros
#define	RXCPP_OPERATORS_RX_CONCATMAP_HPP

Functions
template<class... AN>
auto	rxcpp::operators::concat_map (AN &&...an) -> operator_factory< concat_map_tag, AN... >
	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...
template<class... AN>
auto	rxcpp::operators::concat_transform (AN &&...an) -> operator_factory< concat_map_tag, AN... >
	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...

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

CollectionSelector	the type of the observable producing function. CollectionSelector must be a function with the signature: observable(concat_map::source_value_type)
ResultSelector	the type of the aggregation function (optional). ResultSelector must be a function with the signature: concat_map::value_type(concat_map::source_value_type, concat_map::collection_value_type)
Coordination	the type of the scheduler (optional).

Parameters

s	a function that returns an observable for each item emitted by the source observable.
rs	a 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).
cn	the 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 concatenated. There is another operator rxcpp::observable<T,SourceType>::flat_map that works similar but merges the observables.

Sample Code

    auto values = rxcpp::observable<>::range(1, 3).
        concat_map(
            [](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.
        subscribe(
            [](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: 1 - 2
OnNext: 1 - 3
OnNext: 2 - 1
OnNext: 2 - 2
OnNext: 2 - 3
OnNext: 3 - 1
OnNext: 3 - 2
OnNext: 3 - 3
OnCompleted

Sample Code

    printf("[thread %s] Start task\n", get_pid().c_str());
    auto values = rxcpp::observable<>::range(1, 3).
        concat_map(
            [](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, long v_sub){
                printf("[thread %s] Call ResultSelector(v_main = %d, v_sub = %ld)\n", get_pid().c_str(), v_main, v_sub);
                return std::make_tuple(v_main, v_sub);
            },
            rxcpp::observe_on_new_thread());
    values.
        as_blocking().
        subscribe(
            [](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 47481267428736] Start task
[thread 47481303181056] Call CollectionSelector(v = 1)
[thread 47481303181056] Call ResultSelector(v_main = 1, v_sub = 1)
[thread 47481303181056] OnNext: 1 - 1
[thread 47481303181056] Call ResultSelector(v_main = 1, v_sub = 2)
[thread 47481303181056] OnNext: 1 - 2
[thread 47481303181056] Call ResultSelector(v_main = 1, v_sub = 3)
[thread 47481303181056] OnNext: 1 - 3
[thread 47481303181056] Call CollectionSelector(v = 2)
[thread 47481303181056] Call ResultSelector(v_main = 2, v_sub = 1)
[thread 47481303181056] OnNext: 2 - 1
[thread 47481303181056] Call ResultSelector(v_main = 2, v_sub = 2)
[thread 47481303181056] OnNext: 2 - 2
[thread 47481303181056] Call ResultSelector(v_main = 2, v_sub = 3)
[thread 47481303181056] OnNext: 2 - 3
[thread 47481303181056] Call CollectionSelector(v = 3)
[thread 47481303181056] Call ResultSelector(v_main = 3, v_sub = 1)
[thread 47481303181056] OnNext: 3 - 1
[thread 47481303181056] Call ResultSelector(v_main = 3, v_sub = 2)
[thread 47481303181056] OnNext: 3 - 2
[thread 47481303181056] Call ResultSelector(v_main = 3, v_sub = 3)
[thread 47481303181056] OnNext: 3 - 3
[thread 47481303181056] OnCompleted
[thread 47481267428736] Finish task

Macro Definition Documentation

#define RXCPP_OPERATORS_RX_CONCATMAP_HPP