share operator

src/examples/rpp/doxygen/share.cpp

@@ -0,0 +1,35 @@
+#include <rpp/rpp.hpp>
+
+#include <iostream>
+
+/**
+ * @example share.cpp
+ **/
+int main() // NOLINT(bugprone-exception-escape)
+{
+ {
+ //! [share]
+ 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
+
+ //! [share]
+ }
+ return 0;
+}

src/rpp/rpp/operators.hpp

@@ -189,6 +189,7 @@
 #include<rpp/operators/multicast.hpp>
 #include<rpp/operators/publish.hpp>
 #include<rpp/operators/ref_count.hpp>
+#include<rpp/operators/share.hpp>
 
 /**
  * @defgroup aggregate_operators Aggregate Operators

src/rpp/rpp/operators/fwd.hpp

@@ -113,6 +113,9 @@ namespace rpp::operators
 template<typename Fn>
 autoscan(Fn&& accumulator);
 
+template<template<typename> typename Subject = rpp::subjects::publish_subject>
+autoshare();
+
 autoskip(size_t count);
 
 template<rpp::constraint::observable TObservable, rpp::constraint::observable... TObservables>

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 <rpp/operators/fwd.hpp>
+
+#include <rpp/operators/multicast.hpp>
+#include <rpp/subjects/publish_subject.hpp>
+
+namespace rpp::operators::details
+{
+ template<template<typename> typename Subject>
+ struct share_t
+ {
+ template<rpp::constraint::observable TObservable>
+ auto operator()(TObservable&& observable) const
+ {
+ return template_multicast_t<Subject>{}(std::forward<TObservable>(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<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
+ *
+ * @tparam 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>`
+ *
+ * @par Example
+ * @snippet share.cpp share
+ *
+ * @ingroup connectable_operators
+ * @see https://reactivex.io/documentation/operators/refcount.html
+ */
+ template<template<typename> typename Subject>
+ auto share()
+ {
+ return details::share_t<Subject>{};
+ }
+} // namespace rpp::operators

src/tests/rpp/test_connectable_observable.cpp

@@ -17,6 +17,9 @@
 #include <rpp/operators/map.hpp>
 #include <rpp/operators/multicast.hpp>
 #include <rpp/operators/publish.hpp>
+#include <rpp/operators/ref_count.hpp>
+#include <rpp/operators/share.hpp>
+#include <rpp/sources/create.hpp>
 #include <rpp/sources/just.hpp>
 #include <rpp/subjects/publish_subject.hpp>
 
@@ -339,3 +342,94 @@ TEST_CASE("ref_count")
  }
  }
 }
+
+TEST_CASE("share")
+{
+ auto observer_1 = mock_observer_strategy<int>{};
+ auto observer_2 = mock_observer_strategy<int>{};
+
+ SUBCASE("share is multicast + ref_count")
+ {
+ auto source = rpp::source::just(1);
+ static_assert(rpp::constraint::decayed_same_as<decltype(source | rpp::ops::share()),
+ decltype(source | rpp::ops::publish() | rpp::ops::ref_count())>);
+ }
+
+ SUBCASE("shared observable over subject")
+ {
+ size_t subscriptions_count{};
+ auto subj = rpp::subjects::publish_subject<int>{};
+
+ auto observable = rpp::source::create<int>([&](auto&& obs) {
+ ++subscriptions_count;
+ subj.get_observable().subscribe(std::forward<decltype(obs)>(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});
+ }
+ }
+ }
+ }
+}