Transforming operators are operators that transform items provided by observable. More...

Functions
auto	rpp::operators::buffer (size_t count)
	Periodically gather emissions emitted by an original Observable into bundles and emit these bundles rather than emitting the items one at a time.

template<typename Fn > requires (!utils::is_not_template_callable<Fn> \|\| rpp::constraint::observable<std::invoke_result_t<Fn, rpp::utils::convertible_to_any>>)
auto	rpp::operators::flat_map (Fn &&callable)
	Transform the items emitted by an Observable into Observables, then flatten the emissions from those into a single Observable.

template<typename KeySelector , typename ValueSelector = std::identity, typename KeyComparator = rpp::utils::less> requires ( (!utils::is_not_template_callable<KeySelector> \|\| !std::same_as<void, std::invoke_result_t<KeySelector, rpp::utils::convertible_to_any>>) && (!utils::is_not_template_callable<ValueSelector> \|\| !std::same_as<void, std::invoke_result_t<ValueSelector, rpp::utils::convertible_to_any>>) && (!utils::is_not_template_callable<KeyComparator> \|\| std::strict_weak_order<KeyComparator, rpp::utils::convertible_to_any, rpp::utils::convertible_to_any>))
auto	rpp::operators::group_by (KeySelector &&key_selector, ValueSelector &&value_selector, KeyComparator &&comparator)
	Divide original observable into multiple observables where each new observable emits some group of values from original observable.

template<typename Fn > requires (!utils::is_not_template_callable<Fn> \|\| !std::same_as<void, std::invoke_result_t<Fn, rpp::utils::convertible_to_any>>)
auto	rpp::operators::map (Fn &&callable)
	Transforms the items emitted by an Observable via applying a function to each item and emitting result.

template<typename InitialValue , typename Fn > requires (!utils::is_not_template_callable<Fn> \|\| std::same_as<std::decay_t<InitialValue>, std::invoke_result_t<Fn, std::decay_t<InitialValue> &&, rpp::utils::convertible_to_any>>)
auto	rpp::operators::scan (InitialValue &&initial_value, Fn &&accumulator)
	Apply accumulator function for each emission from observable and result of accumulator from previous step and emit (and cache) resulting value.

template<typename Fn >
auto	rpp::operators::scan (Fn &&accumulator)
	Apply accumulator function for each emission from observable and result of accumulator from previous step and emit (and cache) resulting value.

auto	rpp::operators::window (size_t count)
	Subdivide original observable into sub-observables (window observables) and emit sub-observables of items instead of original items.

template<rpp::constraint::observable TOpeningsObservable, typename TClosingsSelectorFn > requires rpp::constraint::observable<std::invoke_result_t<TClosingsSelectorFn, rpp::utils::extract_observable_type_t<TOpeningsObservable>>>
auto	rpp::operators::window_toggle (TOpeningsObservable &&openings, TClosingsSelectorFn &&closings_selector)
	Subdivide original observable into sub-observables (window observables) and emit sub-observables of items instead of original items.

Detailed Description

Transforming operators are operators that transform items provided by observable.

See also: https://reactivex.io/documentation/operators.html#transforming

Function Documentation

◆ buffer()

auto rpp::operators::buffer ( size_t count )

inline

Periodically gather emissions emitted by an original Observable into bundles and emit these bundles rather than emitting the items one at a time.

The resulting bundle is std::vector<Type> of requested size. Actually it is similar to window() operator, but it emits vectors instead of observables.

Parameters

count number of items being bundled.

Warning: #include <rpp/operators/buffer.hpp>

Example:: // The stream that uses rvalue overloads for operators

rpp::source::just(1, 2, 3, 4, 5)

| rpp::ops::buffer(2)

| rpp::ops::subscribe(

[](const std::vector<int>& v) { std::cout << v << "-"; },

[](const std::exception_ptr&) {},

[]() { std::cout << "|" << std::endl; });

// Source: -1-2-3-4-5--|

// Output: {1,2}-{3,4}-{5}-|

See also: https://reactivex.io/documentation/operators/buffer.html

◆ flat_map()

template<typename Fn >
requires (!utils::is_not_template_callable<Fn> || rpp::constraint::observable<std::invoke_result_t<Fn, rpp::utils::convertible_to_any>>)

auto rpp::operators::flat_map ( Fn && callable )

Transform the items emitted by an Observable into Observables, then flatten the emissions from those into a single Observable.

Actually it makes map(callable) and then merge.

Note that flat_map merges the emissions of these Observables, so that they may interleave.

Parameters

callable function that returns an observable for each item emitted by the source observable.

Warning: #include <rpp/operators/flat_map.hpp>

See also: https://reactivex.io/documentation/operators/flatmap.html

Examples: debounce.cpp, thread_pool.cpp, throttle.cpp, and timeout.cpp.

◆ group_by()

template<typename KeySelector , typename ValueSelector = std::identity, typename KeyComparator = rpp::utils::less>
requires ( (!utils::is_not_template_callable<KeySelector> || !std::same_as<void, std::invoke_result_t<KeySelector, rpp::utils::convertible_to_any>>) && (!utils::is_not_template_callable<ValueSelector> || !std::same_as<void, std::invoke_result_t<ValueSelector, rpp::utils::convertible_to_any>>) && (!utils::is_not_template_callable<KeyComparator> || std::strict_weak_order<KeyComparator, rpp::utils::convertible_to_any, rpp::utils::convertible_to_any>))

auto rpp::operators::group_by	(	KeySelector &&	key_selector,
		ValueSelector &&	value_selector,
		KeyComparator &&	comparator )

Divide original observable into multiple observables where each new observable emits some group of values from original observable.

Actually this operator applies key_selector to emission to obtain key, place rpp::grouped_observable to map with corresponding map and then send observable with this key (if not yet). Original values emitted via this grouped_observables

Parameters

key_selector	Function which determines key for provided item
value_selector	Function which determines value to be emitted to grouped observable
comparator	Function to provide strict_weak_order between key types

Warning: #include <rpp/operators/group_by.hpp>

Example:: rpp::source::just(1, 2, 3, 4, 5, 6, 7, 8)

| rpp::operators::group_by([](int v) { return v % 2 == 0; })

| rpp::operators::subscribe([](auto grouped_observable) {

auto key = grouped_observable.get_key();

std::cout << "new grouped observable " << key << std::endl;

grouped_observable.subscribe([key](int val) {

std::cout << "key [" << key << "] Val: " << val << std::endl;

});

});

// Output: new grouped observable 0

// key [0] Val: 1

// new grouped observable 1

// key [1] Val: 2

// key [0] Val: 3

// key [1] Val: 4

// key [0] Val: 5

// key [1] Val: 6

// key [0] Val: 7

// key [1] Val: 8

struct Person

{

std::string name;

int age;

};

rpp::source::just(Person{"Kate", 18},

Person{"Alex", 25},

Person{"Nick", 18},

Person{"Jack", 25},

Person{"Tom", 30},

Person{"Vanda", 18})

| rpp::operators::group_by([](const Person& v) { return v.age; }, [](const Person& v) { return v.name; })

| rpp::operators::subscribe([](auto grouped_observable) {

grouped_observable.subscribe([age = grouped_observable.get_key()](const std::string& name) {

std::cout << "Age [" << age << "] Name: " << name << std::endl;

});

});

// Output: Age [18] Name: Kate

// Age [25] Name: Alex

// Age [18] Name: Nick

// Age [25] Name: Jack

// Age [30] Name: Tom

// Age [18] Name: Vanda

See also: https://reactivex.io/documentation/operators/groupby.html

Examples: group_by.cpp.

◆ map()

template<typename Fn >
requires (!utils::is_not_template_callable<Fn> || !std::same_as<void, std::invoke_result_t<Fn, rpp::utils::convertible_to_any>>)

auto rpp::operators::map ( Fn && callable )

Transforms the items emitted by an Observable via applying a function to each item and emitting result.

Note: The Map operator can keep same type of value or change it to some another type.

Actually this operator just applies callable to each obtained emission and emit resulting value

Performance notes:

No any heap allocations at all
No any copies/moves of emissions, just forwarding to callable

Parameters

callable is callable used to provide this transformation. Should accept Type of original observable and return type for new observable

Warning: #include <rpp/operators/map.hpp>

Example with same type:: rpp::source::just(42)

| rpp::operators::map([](int value) { return value + 10; })

| rpp::operators::subscribe([](int v) { std::cout << v << std::endl; });

// Output: 52

Example with changed type:: rpp::source::just(42)

| rpp::operators::map([](int value) { return std::to_string(value) + " VAL"; })

| rpp::operators::subscribe([](const std::string& v) { std::cout << v << std::endl; });

// Output: 42 VAL

See also: https://reactivex.io/documentation/operators/map.html

Examples: map.cpp.

◆ scan() [1/2]

template<typename Fn >

auto rpp::operators::scan ( Fn && accumulator )

Apply accumulator function for each emission from observable and result of accumulator from previous step and emit (and cache) resulting value.

Actually this operator applies provided accumulator function to seed and new emission, emits resulting value and updates seed value for next emission

Warning: There is no initial value for seed, so, first value would be used as seed value and forwarded as is.

Performance notes:

No any heap allocations at all
Keep actual seed/cache inside observable and updating it every emission

Parameters

accumulator function which accepts seed value and new value from observable and return new value of seed. Can accept seed by move-reference.

Warning: #include <rpp/operators/scan.hpp>

Example: rpp::source::just(1, 2, 3)

| rpp::operators::scan(std::plus<int>{})

| rpp::operators::subscribe([](int v) { std::cout << v << std::endl; });

// Output: 1 3 6

See also: https://reactivex.io/documentation/operators/scan.html

◆ scan() [2/2]

template<typename InitialValue , typename Fn >
requires (!utils::is_not_template_callable<Fn> || std::same_as<std::decay_t<InitialValue>, std::invoke_result_t<Fn, std::decay_t<InitialValue> &&, rpp::utils::convertible_to_any>>)

auto rpp::operators::scan	(	InitialValue &&	initial_value,
		Fn &&	accumulator )

Apply accumulator function for each emission from observable and result of accumulator from previous step and emit (and cache) resulting value.

Actually this operator applies provided accumulator function to seed and new emission, emits resulting value and updates seed value for next emission

Warning: Initial value would be used as first value from this observable (would be sent during subscription) and initial value for cache

Performance notes:

No any heap allocations at all
Keep actual seed/cache inside observable and updating it every emission

Parameters

initial_value	initial value for seed which would be sent during subscription and applied for first value from observable. Then it will be replaced with result and etc.
accumulator	function which accepts seed value and new value from observable and return new value of seed. Can accept seed by move-reference.

Warning: #include <rpp/operators/scan.hpp>

Example: rpp::source::just(1, 2, 3)

| rpp::operators::scan(10, std::plus<int>{})

| rpp::operators::subscribe([](int v) { std::cout << v << std::endl; });

// Output: 10 11 13 16

rpp::source::just(1, 2, 3)

| rpp::operators::scan(std::vector<int>{}, [](std::vector<int>&& seed, int new_value) {

seed.push_back(new_value);

return std::move(seed);

})

| rpp::operators::subscribe([](const std::vector<int>& v) {

std::cout << "vector: ";

for (int val : v)

std::cout << val << " ";

std::cout << std::endl;

});

// Output: vector:

// vector: 1

// vector: 1 2

// vector: 1 2 3

See also: https://reactivex.io/documentation/operators/scan.html

Examples: scan.cpp.

◆ window()

auto rpp::operators::window ( size_t count )

inline

Subdivide original observable into sub-observables (window observables) and emit sub-observables of items instead of original items.

Actually it is similar to buffer but it emits observable instead of container.

Parameters

window_size amount of items which every observable would have

Warning: #include <rpp/operators/window.hpp>

Example: rpp::source::just(1, 2, 3, 4, 5)

| rpp::operators::window(3)

| rpp::operators::subscribe([](const rpp::window_observable<int>& v) {

std::cout << "\nNew observable " << std::endl;

v.subscribe([](int v) { std::cout << v << " "; });

});

// Output: New observable

// 1 2 3

// New observable

// 4 5

See also: https://reactivex.io/documentation/operators/window.html

Examples: window.cpp.

◆ window_toggle()

template<rpp::constraint::observable TOpeningsObservable, typename TClosingsSelectorFn >
requires rpp::constraint::observable<std::invoke_result_t<TClosingsSelectorFn, rpp::utils::extract_observable_type_t<TOpeningsObservable>>>

auto rpp::operators::window_toggle	(	TOpeningsObservable &&	openings,
		TClosingsSelectorFn &&	closings_selector )

Subdivide original observable into sub-observables (window observables) and emit sub-observables of items instead of original items.

Values from openings observable used to specify moment when new window will be opened. closings_selector is used to obtain observable to specify moment when new window will be closed.

Actually it is similar to buffer but it emits observable instead of container.

Parameters

openings	is observable which emissions used to start new window
closings_selector	is function which returns observable which emission/completion means closing of opened window

Warning: #include <rpp/operators/window.hpp>

Example: size_t counter{};

auto source = rpp::source::just(rpp::schedulers::current_thread{}, 1, 2, 3, 4, 5) | rpp::operators::publish() | rpp::operators::ref_count();

source

| rpp::operators::window_toggle(source, [source](int) { return source | rpp::ops::filter([](int v) { return v % 2 == 0; }); })

| rpp::operators::subscribe([&counter](const rpp::window_toggle_observable<int>& obs) {

std::cout << "New observable " << ++counter << std::endl;

obs.subscribe([counter](int v) { std::cout << counter << ": " << v << " " << std::endl; }, [counter]() { std::cout << "closing " << counter << std::endl; });

});

// Output:

// New observable 1

// 1: 1

// New observable 2

// 1: 2

// 2: 2

// closing 1

// New observable 3

// 2: 3

// 3: 3

// New observable 4

// 2: 4

// 3: 4

// 4: 4

// closing 2

// closing 3

// New observable 5

// 4: 5

// 5: 5

// closing 4

// closing 5

See also: https://reactivex.io/documentation/operators/window.html

Examples: window_toggle.cpp.

Functions

Detailed Description

Function Documentation

◆ buffer()

◆ flat_map()

◆ group_by()

◆ map()

◆ scan() [1/2]

◆ scan() [2/2]

◆ window()

◆ window_toggle()