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-zip.hpp File Reference

Bring by one item from all given observables and select a value to emit from the new observable that is returned. More...

#include "../rx-includes.hpp"
Include dependency graph for rx-zip.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< zip_tag >
 

Namespaces

 rxcpp
 
 rxcpp::operators
 

Macros

#define RXCPP_OPERATORS_RX_ZIP_HPP
 

Functions

template<class... AN>
auto rxcpp::operators::zip (AN &&... an) -> operator_factory< zip_tag, AN... >
 

Detailed Description

Bring by one item from all given observables and select a value to emit from the new observable that is returned.

Template Parameters
ANtypes of scheduler (optional), aggregate function (optional), and source observables
Parameters
anscheduler (optional), aggregation function (optional), and source observables
Returns
Observable that emits the result of combining the items emitted and brought by one from each of 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\n

Neither scheduler nor aggregation function are present:

auto o1 = rxcpp::observable<>::interval(std::chrono::milliseconds(1));
auto o2 = rxcpp::observable<>::interval(std::chrono::milliseconds(2));
auto o3 = rxcpp::observable<>::interval(std::chrono::milliseconds(3));
auto values = o1.zip(o2, o3);
values.
take(3).
[](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: 1, 1, 1
OnNext: 2, 2, 2
OnNext: 3, 3, 3
OnCompleted

Only scheduler is present:

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

Only aggregation function is present:

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

Both scheduler and aggregation function are present:

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

Macro Definition Documentation

◆ RXCPP_OPERATORS_RX_ZIP_HPP

#define RXCPP_OPERATORS_RX_ZIP_HPP
rxcpp::operators::zip
auto zip(AN &&... an) -> operator_factory< zip_tag, AN... >
Definition: rx-zip.hpp:274
rxcpp::observe_on_new_thread
observe_on_one_worker observe_on_new_thread()
Definition: rx-observe_on.hpp:328
rxcpp::synchronize_event_loop
synchronize_in_one_worker synchronize_event_loop()
Definition: rx-synchronize.hpp:250
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