diff --git a/src/examples/rpp/doxygen/share.cpp b/src/examples/rpp/doxygen/share.cpp new file mode 100644 index 000000000..0bdb14f3e --- /dev/null +++ b/src/examples/rpp/doxygen/share.cpp @@ -0,0 +1,35 @@ +#include + +#include + +/** + * @example share.cpp + **/ +int main() // NOLINT(bugprone-exception-escape) +{ + { + //! [share] + auto subject = rpp::subjects::publish_subject{}; + + auto observable = rpp::source::create([&](auto&& observer) { + std::cout << "SUBSCRIBE" << std::endl; + subject.get_observable().subscribe(std::forward(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 + + //! [share] + } + return 0; +} diff --git a/src/rpp/rpp/operators.hpp b/src/rpp/rpp/operators.hpp index 6b0911d66..f35310e3e 100644 --- a/src/rpp/rpp/operators.hpp +++ b/src/rpp/rpp/operators.hpp @@ -189,6 +189,7 @@ #include #include #include +#include /** * @defgroup aggregate_operators Aggregate Operators diff --git a/src/rpp/rpp/operators/fwd.hpp b/src/rpp/rpp/operators/fwd.hpp index 0fed6f13d..f660a50e6 100644 --- a/src/rpp/rpp/operators/fwd.hpp +++ b/src/rpp/rpp/operators/fwd.hpp @@ -113,6 +113,9 @@ namespace rpp::operators template auto scan(Fn&& accumulator); + template typename Subject = rpp::subjects::publish_subject> + auto share(); + auto skip(size_t count); template diff --git a/src/rpp/rpp/operators/share.hpp b/src/rpp/rpp/operators/share.hpp new file mode 100644 index 000000000..9c2a90b03 --- /dev/null +++ b/src/rpp/rpp/operators/share.hpp @@ -0,0 +1,53 @@ +// ReactivePlusPlus library +// +// Copyright Aleksey Loginov 2023 - present. +// Distributed under the Boost Software License, Version 1.0. +// (See accompanying file LICENSE_1_0.txt or copy at +// https://www.boost.org/LICENSE_1_0.txt) +// +// Project home: https://github.com/AlexInLog/ReactivePlusPlus +// + +#pragma once + +#include + +#include +#include + +namespace rpp::operators::details +{ + template typename Subject> + struct share_t + { + template + auto operator()(TObservable&& observable) const + { + return template_multicast_t{}(std::forward(observable)).ref_count(); + } + }; +} // namespace rpp::operators::details + +namespace rpp::operators +{ + /** + * @brief Shares single subscription to original observable between multiple observers + * @details This is a shortcut for `multicast() | 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` 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 + * + * @tparam Subject is template template typename over Subject to be created to multicast emissions, rpp::subjects::publish_subject by default + * @note `#include ` + * + * @par Example + * @snippet share.cpp share + * + * @ingroup connectable_operators + * @see https://reactivex.io/documentation/operators/refcount.html + */ + template typename Subject> + auto share() + { + return details::share_t{}; + } +} // namespace rpp::operators diff --git a/src/tests/rpp/test_connectable_observable.cpp b/src/tests/rpp/test_connectable_observable.cpp index 088067ccb..863a6e52a 100644 --- a/src/tests/rpp/test_connectable_observable.cpp +++ b/src/tests/rpp/test_connectable_observable.cpp @@ -17,6 +17,9 @@ #include #include #include +#include +#include +#include #include #include @@ -339,3 +342,94 @@ TEST_CASE("ref_count") } } } + +TEST_CASE("share") +{ + auto observer_1 = mock_observer_strategy{}; + auto observer_2 = mock_observer_strategy{}; + + SUBCASE("share is multicast + ref_count") + { + auto source = rpp::source::just(1); + static_assert(rpp::constraint::decayed_same_as); + } + + SUBCASE("shared observable over subject") + { + size_t subscriptions_count{}; + auto subj = rpp::subjects::publish_subject{}; + + auto observable = rpp::source::create([&](auto&& obs) { + ++subscriptions_count; + subj.get_observable().subscribe(std::forward(obs)); + }) + | rpp::ops::share(); + + SUBCASE("original observable is not subscribed before any subscription") + { + CHECK(subscriptions_count == 0); + } + SUBCASE("subscribe both observers") + { + observable.subscribe(observer_1); + auto sub = rpp::composite_disposable_wrapper::make(); + observable.subscribe(rpp::composite_disposable_wrapper{sub}, observer_2); + + SUBCASE("original observable is subscribed only once") + { + CHECK(subscriptions_count == 1); + } + SUBCASE("send value") + { + subj.get_observer().on_next(1); + SUBCASE("both observers obtain values") + { + auto validate = [](auto observer) { + CHECK(observer.get_received_values() == std::vector{1}); + CHECK(observer.get_total_on_next_count() == 1); + CHECK(observer.get_on_error_count() == 0); + CHECK(observer.get_on_completed_count() == 0); + }; + validate(observer_1); + validate(observer_2); + } + } + SUBCASE("unsubscribe second observer and send value") + { + sub.dispose(); + subj.get_observer().on_next(1); + SUBCASE("only first observer obtains values") + { + CHECK(observer_1.get_received_values() == std::vector{1}); + CHECK(observer_2.get_total_on_next_count() == 0); + } + SUBCASE("original observable is still subscribed") + { + CHECK(subscriptions_count == 1); + } + } + } + SUBCASE("subscribe, unsubscribe everything and subscribe again") + { + auto sub = rpp::composite_disposable_wrapper::make(); + observable.subscribe(rpp::composite_disposable_wrapper{sub}, observer_1); + sub.dispose(); + + observable.subscribe(observer_2); + SUBCASE("original observable is subscribed again") + { + CHECK(subscriptions_count == 2); + } + SUBCASE("send value") + { + subj.get_observer().on_next(1); + SUBCASE("only second observer obtains values") + { + CHECK(observer_1.get_total_on_next_count() == 0); + CHECK(observer_2.get_received_values() == std::vector{1}); + } + } + } + } +}