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.
template<template< typename > typename Subject = rpp::subjects::publish_subject>
auto	rpp::operators::share ()
	Shares single subscription to original observable between multiple observers.

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

Examples

ref_count.cpp.

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

connect.cpp, 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

ref_count.cpp, and window_toggle.cpp.

share()

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

auto rpp::operators::share

(

)

Shares single subscription to original observable between multiple observers.

This is a shortcut for multicast<Subject>() | ref_count(): observable is converted to rpp::connectable_observable with help of inline instantiated subject and immediately forced to behave like common observable. As a result original observable is subscribed on the first subscription and disposed on the last unsubscription, while emissions are multicasted to all observers.

Warning

This operator creates fresh Subject<Type> everytime new observable passed to it, but same subject is reused for all observers of that particular observable. In case of rpp::subjects::publish_subject observers subscribed after some emissions would obtain only upcoming ones

Template Parameters

Subject

is template template typename over Subject to be created to multicast emissions, rpp::subjects::publish_subject by default

Note

#include <rpp/operators/share.hpp>

Example

        auto subject = rpp::subjects::publish_subject<int>{};
 
        auto observable = rpp::source::create<int>([&](auto&& observer) {
                              std::cout << "SUBSCRIBE" << std::endl;
                              subject.get_observable().subscribe(std::forward<decltype(observer)>(observer));
                          })
                        | rpp::ops::share();
 
        std::cout << "before subscriptions" << std::endl;
        observable.subscribe([](int v) { std::cout << "#1 " << v << std::endl; });
        observable.subscribe([](int v) { std::cout << "#2 " << v << std::endl; });
 
        subject.get_observer().on_next(1);
        subject.get_observer().on_completed();
        // Output:
        // before subscriptions
        // SUBSCRIBE
        // #1 1
        // #2 1
 

See also

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

Examples

share.cpp.