Connectable Operators

Connectable operators are operators that provide extra functionality for multicasting of controlling of subscription. More...

Functions
template<subjects::constraint::subject_of_type< Type > TSubject> requires is_header_included<multicast_tag, TSubject>
auto	observable::multicast (TSubject &&subject) const &
	Converts ordinary observable to rpp::connectable_observable with help of provided subject.
template<typename ... Args> requires is_header_included<publish_tag, Args...>
auto	observable::publish () const &
	Converts ordinary observable to rpp::connectable_observable with help of rpp::subjects::publish_subject.
template<typename ... Args> requires is_header_included<ref_count_tag, Args...>
auto	observable::ref_count () const &
	Forces rpp::connectable_observable to behave like common observable.

Detailed Description

Connectable operators are operators that provide extra functionality for multicasting of controlling of subscription.

See also

https://reactivex.io/documentation/operators.html#connectable

Function Documentation

multicast()

template<constraint::decayed_type Type, typename SpecificObservable >

template<subjects::constraint::subject_of_type< Type > TSubject>
requires is_header_included<multicast_tag, TSubject>

auto observable::multicast ( TSubject &&  subject ) const &

inline

Converts ordinary observable to rpp::connectable_observable with help of provided subject.

Connectable observable is common observable, but actually it starts emissions of items only after call "connect", "ref_count" or any other available way. Also it uses subject to multicast values to subscribers

Parameters

subject

is subject used to create rpp::connectable_observable

Returns

new specific_observable with the multicast operator as most recent operator.

Warning

#include <rpp/operators/multicast.hpp>

Example

    auto subject = rpp::subjects::publish_subject<int>{};
    auto observable = rpp::source::just(1, 2, 3).multicast(subject);
    observable.subscribe([](int v) {std::cout << "#1 " << v << std::endl; });
    observable.subscribe([](int v) {std::cout << "#2 " << v << std::endl; });
    observable.connect();
    // Output:
    // #1 1
    // #2 1
    // #1 2
    // #2 2
    // #1 3
    // #2 3

See also

https://reactivex.io/documentation/operators/publish.html

publish()

template<constraint::decayed_type Type, typename SpecificObservable >

template<typename ... Args>
requires is_header_included<publish_tag, Args...>

auto observable::publish ( ) const &

inline

Converts ordinary observable to rpp::connectable_observable with help of rpp::subjects::publish_subject.

Connectable observable is common observable, but actually it starts emissions of items only after call "connect", "ref_count" or any other available way. Also it uses subject to multicast values to subscribers

Returns

new specific_observable with the publish operator as most recent operator.

Warning

#include <rpp/operators/publish.hpp>

Example

    auto observable = rpp::source::just(1,2,3).publish();
    observable.subscribe([](int v) {std::cout << "#1 " << v << std::endl; });
    observable.subscribe([](int v) {std::cout << "#2 " << v << std::endl; });
    observable.connect();
    // Output:
    // #1 1
    // #2 1
    // #1 2
    // #2 2
    // #1 3
    // #2 3

See also

https://reactivex.io/documentation/operators/publish.html

ref_count()

template<constraint::decayed_type Type, typename SpecificObservable >

template<typename ... Args>
requires is_header_included<ref_count_tag, Args...>

auto observable::ref_count ( ) const &

inline

Forces rpp::connectable_observable to behave like common observable.

Connects rpp::connectable_observable on the first subscription and unsubscribes on last unsubscription

Returns

new specific_observable with the ref_count operator as most recent operator.

Warning

#include <rpp/operators/ref_count.hpp>

Example

    auto observable = rpp::source::just(1, 2, 3).publish();
    observable.subscribe([](int v) {std::cout << "#1 " << v << std::endl; });
    // No Output
 
    observable.ref_count().subscribe([](int v) {std::cout << "#2 " << v << std::endl; });
    // Output:
    // #1 1
    // #2 1
    // #1 2
    // #2 2
    // #1 3
    // #2 3

See also

https://reactivex.io/documentation/operators/refcount.html