Connectable Operators

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

Functions
auto	rpp::connectable_observable< OriginalObservable, Subject >::ref_count () const
	Forces rpp::connectable_observable to behave like common observable.
template<rpp::constraint::subject Subject>
auto	rpp::operators::multicast (Subject &&subject)
	Converts ordinary observable to rpp::connectable_observable with help of provided subject.
template<template< typename > typename Subject = rpp::subjects::publish_subject>
auto	rpp::operators::multicast ()
	Converts ordinary observable to rpp::connectable_observable with help of inline instsantiated subject of provided type.
auto	rpp::operators::publish ()
	Converts ordinary observable to rpp::connectable_observable with help of inline instsantiated publish subject.
auto	rpp::operators::ref_count ()
	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() [1/2]

template<template< typename > typename Subject = rpp::subjects::publish_subject>

auto rpp::operators::multicast

(

)

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

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

Warning

This overloading creates fresh Subject<Type> everytime new observable passed to this operator

Template Parameters

Subject

is template teamplate typename over Subject to be created to create corresponding connectable_observable for provided observable

Note

#include <rpp/operators/multicast.hpp>

Example

        auto observable = rpp::source::just(1, 2, 3) | rpp::operators::multicast<rpp::subjects::publish_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

Examples

multicast.cpp, and ref_count.cpp.

multicast() [2/2]

template<rpp::constraint::subject Subject>

auto rpp::operators::multicast

(

Subject &&

subject

)

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

Warning

Same subject would be used to all observables lifted via this operator. To have fresh subject everytime use another overloading

Parameters

subject

is subject used to create rpp::connectable_observable

Note

#include <rpp/operators/multicast.hpp>

Example

        auto subject    = rpp::subjects::publish_subject<int>{};
        auto observable = rpp::source::just(1, 2, 3) | rpp::operators::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()

auto rpp::operators::publish ( )

inline

Converts ordinary observable to rpp::connectable_observable with help of inline instsantiated 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

Warning

This overloading creates fresh Subject<Type> everytime new observable passed to this operator

Template Parameters

Subject

is template teamplate typename over Subject to be created to create corresponding connectable_observable for provided observable

Note

#include <rpp/operators/publish.hpp>

Example

        auto observable = rpp::source::just(1, 2, 3) | rpp::operators::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

Examples

multicast.cpp, and window_toggle.cpp.

ref_count() [1/2]

template<rpp::constraint::observable OriginalObservable, typename Subject >

auto rpp::connectable_observable< OriginalObservable, Subject >::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

Example

        auto observable = rpp::source::create<int>([](const auto& observer) {
                              std::cout << "SUBSCRIBE" << std::endl;
                              for (int i = 0; i < 3; ++i)
                              {
                                  observer.on_next(i);
                              }
                              observer.on_completed();
                          })
                        | rpp::operators::multicast();
 
        std::cout << "subscribe first" << std::endl;
        observable.subscribe([](int v) { std::cout << "#1 " << v << std::endl; });
        // No Output
 
        std::cout << "subscribe with ref_count" << std::endl;
        observable.ref_count().subscribe([](int v) { std::cout << "#2 " << v << std::endl; });
        // Output:
        // subscribe first
        // subscribe with ref_count
        // SUBSCRIBE
        // #1 0
        // #2 0
        // #1 1
        // #2 1
        // #1 2
        // #2 2
 

See also

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

ref_count() [2/2]

auto rpp::operators::ref_count ( )

inline

Forces rpp::connectable_observable to behave like common observable.

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

Example

        auto observable = rpp::source::just(1, 2, 3) | rpp::operators::multicast();
        observable.subscribe([](int v) { std::cout << "#1 " << v << std::endl; });
        // No Output
 
        observable | rpp::ops::ref_count() | rpp::ops::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

Examples

window_toggle.cpp.