rx-with_latest_from.hpp File Reference

For each item from the first observable select the latest value from all the observables to emit from the new observable that is returned. More...

#include "../rx-includes.hpp"

Include dependency graph for rx-with_latest_from.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< with_latest_from_tag >

Namespaces
	rxcpp
	rxcpp::operators

Macros
#define	RXCPP_OPERATORS_RX_WITH_LATEST_FROM_HPP

Functions
template<class... AN>
auto	rxcpp::operators::with_latest_from (AN &&...an) -> operator_factory< with_latest_from_tag, AN... >
	For each item from the first observable select the latest value from all the observables to emit from the new observable that is returned. More...

Detailed Description

For each item from the first observable select the latest value from all the observables to emit from the new observable that is returned.

Template Parameters

AN	types of scheduler (optional), aggregate function (optional), and source observables

Parameters

an	scheduler (optional), aggregation function (optional), and source observables

Returns

Observable that emits items that are the result of combining the items emitted by the source observables.

If scheduler is omitted, identity_current_thread is used.

If aggregation function is omitted, the resulting observable returns tuples of emitted items.

Sample Code

Neither scheduler nor aggregation function are present:

    auto o1 = rxcpp::observable<>::interval(std::chrono::milliseconds(2));
    auto o2 = rxcpp::observable<>::interval(std::chrono::milliseconds(3));
    auto o3 = rxcpp::observable<>::interval(std::chrono::milliseconds(5));
    auto values = o1.with_latest_from(o2, o3);
    values.
        take(5).
        subscribe(
            [](std::tuple<int, int, int> v){printf("OnNext: %d, %d, %d\n", std::get<0>(v), std::get<1>(v), std::get<2>(v));},
            [](){printf("OnCompleted\n");});

OnNext: 2, 1, 1
OnNext: 3, 2, 1
OnNext: 4, 2, 2
OnNext: 5, 3, 2
OnNext: 6, 4, 2
OnCompleted

Only scheduler is present:

    printf("[thread %s] Start task\n", get_pid().c_str());
    auto thr = rxcpp::synchronize_event_loop();
    auto o1 = rxcpp::observable<>::interval(std::chrono::milliseconds(2)).map([](int v) {
        printf("[thread %s] Source1 OnNext: %d\n", get_pid().c_str(), v);
        return v;
    });
    auto o2 = rxcpp::observable<>::interval(std::chrono::milliseconds(3)).map([](int v) {
        printf("[thread %s] Source2 OnNext: %d\n", get_pid().c_str(), v);
        return v;
    });
    auto o3 = rxcpp::observable<>::interval(std::chrono::milliseconds(5)).map([](int v) {
        printf("[thread %s] Source3 OnNext: %d\n", get_pid().c_str(), v);
        return v;
    });
    auto values = o1.with_latest_from(thr, o2, o3);
    values.
        take(5).
        as_blocking().
        subscribe(
            [](std::tuple<int, int, int> v){printf("[thread %s] OnNext: %d, %d, %d\n", get_pid().c_str(), std::get<0>(v), std::get<1>(v), std::get<2>(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 47481267428736] Source1 OnNext: 1
[thread 47481267428736] Source2 OnNext: 1
[thread 47481267428736] Source3 OnNext: 1
[thread 47481267428736] Source1 OnNext: 2
[thread 47481294776064] OnNext: 2, 1, 1
[thread 47481267428736] Source2 OnNext: 2
[thread 47481267428736] Source1 OnNext: 3
[thread 47481294776064] OnNext: 3, 2, 1
[thread 47481267428736] Source3 OnNext: 2
[thread 47481267428736] Source1 OnNext: 4
[thread 47481267428736] Source2 OnNext: 3
[thread 47481294776064] OnNext: 4, 2, 2
[thread 47481267428736] Source1 OnNext: 5
[thread 47481294776064] OnNext: 5, 3, 2
[thread 47481267428736] Source2 OnNext: 4
[thread 47481267428736] Source1 OnNext: 6
[thread 47481267428736] Source3 OnNext: 3
[thread 47481294776064] OnNext: 6, 4, 2
[thread 47481294776064] OnCompleted
[thread 47481267428736] Finish task

Only aggregation function is present:

    auto o1 = rxcpp::observable<>::interval(std::chrono::milliseconds(2));
    auto o2 = rxcpp::observable<>::interval(std::chrono::milliseconds(3));
    auto o3 = rxcpp::observable<>::interval(std::chrono::milliseconds(5));
    auto values = o1.with_latest_from(
        [](int v1, int v2, int v3) {
            return 100 * v1 + 10 * v2 + v3;
        },
        o2, o3);
    values.
        take(5).
        subscribe(
            [](int v){printf("OnNext: %d\n", v);},
            [](){printf("OnCompleted\n");});

OnNext: 211
OnNext: 321
OnNext: 422
OnNext: 532
OnNext: 642
OnCompleted

Both scheduler and aggregation function are present:

    auto o1 = rxcpp::observable<>::interval(std::chrono::milliseconds(2));
    auto o2 = rxcpp::observable<>::interval(std::chrono::milliseconds(3));
    auto o3 = rxcpp::observable<>::interval(std::chrono::milliseconds(5));
    auto values = o1.with_latest_from(
        rxcpp::observe_on_new_thread(),
        [](int v1, int v2, int v3) {
            return 100 * v1 + 10 * v2 + v3;
        },
        o2, o3);
    values.
        take(5).
        as_blocking().
        subscribe(
            [](int v){printf("OnNext: %d\n", v);},
            [](){printf("OnCompleted\n");});

OnNext: 211
OnNext: 321
OnNext: 422
OnNext: 532
OnNext: 642
OnCompleted

Macro Definition Documentation

#define RXCPP_OPERATORS_RX_WITH_LATEST_FROM_HPP