From 867f101cbbdcc3a423250e91c20939c81ad25be8 Mon Sep 17 00:00:00 2001
From: Aleksey Loginov <victimsnino@gmail.com>
Date: Fri, 13 Sep 2024 22:57:09 +0300
Subject: [PATCH 01/17] add test

---
 src/tests/utils/disposable_observable.hpp | 15 +++++++++++++++
 1 file changed, 15 insertions(+)

diff --git a/src/tests/utils/disposable_observable.hpp b/src/tests/utils/disposable_observable.hpp
index 49e451f13..c8124077f 100644
--- a/src/tests/utils/disposable_observable.hpp
+++ b/src/tests/utils/disposable_observable.hpp
@@ -99,6 +99,21 @@ void test_operator_over_observable_with_disposable(auto&& op)
         CHECK(observable_disposable.is_disposed());
     }
 
+    SECTION("operator doesn't disposes disposable too early")
+    {
+        auto observable_disposable = rpp::composite_disposable_wrapper::make();
+        auto observable            = rpp::source::create<T>([&observable_disposable](auto&& obs) {
+            obs.set_upstream(observable_disposable);
+        });
+
+        auto observer_disposable = rpp::composite_disposable_wrapper::make();
+        op(observable) | rpp::ops::subscribe(observer_disposable, [](const auto&) {});
+
+        CHECK(!observable_disposable.is_disposed());
+        observer_disposable.dispose();
+        CHECK(observable_disposable.is_disposed());
+    }
+
     SECTION("operator disposes disposable on_error")
     {
         op(rpp::source::create<T>([](auto&& obs) {

From bd836f84f8258cb95cdd79b2af49049dac58786b Mon Sep 17 00:00:00 2001
From: Aleksey Loginov <victimsnino@gmail.com>
Date: Sun, 15 Sep 2024 00:01:08 +0300
Subject: [PATCH 02/17] partial fix

---
 src/rpp/rpp/operators/combine_latest.hpp      | 22 ++++-----
 .../operators/details/combining_strategy.hpp  | 45 +++++++++----------
 src/rpp/rpp/operators/zip.hpp                 | 18 ++++----
 3 files changed, 40 insertions(+), 45 deletions(-)

diff --git a/src/rpp/rpp/operators/combine_latest.hpp b/src/rpp/rpp/operators/combine_latest.hpp
index 671232cdf..2beeffc94 100644
--- a/src/rpp/rpp/operators/combine_latest.hpp
+++ b/src/rpp/rpp/operators/combine_latest.hpp
@@ -19,11 +19,11 @@
 namespace rpp::operators::details
 {
     template<rpp::constraint::observer Observer, typename TSelector, rpp::constraint::decayed_type... Args>
-    class combine_latest_disposable final : public combining_disposable<Observer, Args...>
+    class combine_latest_state final : public combining_state<Observer>
     {
     public:
-        explicit combine_latest_disposable(Observer&& observer, const TSelector& selector)
-            : combining_disposable<Observer, Args...>(std::move(observer))
+        explicit combine_latest_state(Observer&& observer, const TSelector& selector)
+            : combining_state<Observer>(std::move(observer), sizeof...(Args))
             , m_selector(selector)
         {
         }
@@ -40,23 +40,23 @@ namespace rpp::operators::details
 
     template<size_t I, rpp::constraint::observer Observer, typename TSelector, rpp::constraint::decayed_type... Args>
     struct combine_latest_observer_strategy final
-        : public combining_observer_strategy<combine_latest_disposable<Observer, TSelector, Args...>>
+        : public combining_observer_strategy<combine_latest_state<Observer, TSelector, Args...>>
     {
-        using combining_observer_strategy<combine_latest_disposable<Observer, TSelector, Args...>>::disposable;
+        using combining_observer_strategy<combine_latest_state<Observer, TSelector, Args...>>::state;
 
         template<typename T>
         void on_next(T&& v) const
         {
             // mutex need to be locked during changing of values, generating new values and sending of new values due to we can't update value while we are sending old one
-            const auto observer = disposable->get_observer_under_lock();
-            disposable->get_values().template get<I>().emplace(std::forward<T>(v));
+            const auto observer = state->get_observer_under_lock();
+            state->get_values().template get<I>().emplace(std::forward<T>(v));
 
-            disposable->get_values().apply(&apply_impl<decltype(disposable)>, disposable, observer);
+            state->get_values().apply(&apply_impl<decltype(state)>, state, observer);
         }
 
     private:
-        template<typename TDisposable>
-        static void apply_impl(const TDisposable& disposable, const rpp::utils::pointer_under_lock<Observer>& observer, const std::optional<Args>&... vals)
+        template<typename TState>
+        static void apply_impl(const TState& disposable, const rpp::utils::pointer_under_lock<Observer>& observer, const std::optional<Args>&... vals)
         {
             if ((vals.has_value() && ...))
                 observer->on_next(disposable->get_selector()(vals.value()...));
@@ -64,7 +64,7 @@ namespace rpp::operators::details
     };
 
     template<typename TSelector, rpp::constraint::observable... TObservables>
-    struct combine_latest_t : public combining_operator_t<combine_latest_disposable, combine_latest_observer_strategy, TSelector, TObservables...>
+    struct combine_latest_t : public combining_operator_t<combine_latest_state, combine_latest_observer_strategy, TSelector, TObservables...>
     {
     };
 } // namespace rpp::operators::details
diff --git a/src/rpp/rpp/operators/details/combining_strategy.hpp b/src/rpp/rpp/operators/details/combining_strategy.hpp
index f7f5152c5..d3a0b1c92 100644
--- a/src/rpp/rpp/operators/details/combining_strategy.hpp
+++ b/src/rpp/rpp/operators/details/combining_strategy.hpp
@@ -22,12 +22,13 @@
 
 namespace rpp::operators::details
 {
-    template<rpp::constraint::observer Observer, rpp::constraint::decayed_type... Args>
-    class combining_disposable : public composite_disposable
+    template<rpp::constraint::observer Observer>
+    class combining_state
     {
     public:
-        explicit combining_disposable(Observer&& observer)
+        explicit combining_state(Observer&& observer, size_t on_completed_needed)
             : m_observer_with_mutex{std::move(observer)}
+            , m_on_completed_needed{on_completed_needed}
         {
         }
 
@@ -42,41 +43,37 @@ namespace rpp::operators::details
     private:
         rpp::utils::value_with_mutex<Observer> m_observer_with_mutex{};
 
-        std::atomic_size_t m_on_completed_needed{sizeof...(Args)};
+        std::atomic_size_t m_on_completed_needed;
     };
 
-    template<typename TDisposable>
+    template<typename TState>
     struct combining_observer_strategy
     {
-        std::shared_ptr<TDisposable> disposable{};
+        std::shared_ptr<TState> state{};
 
         void set_upstream(const rpp::disposable_wrapper& d) const
         {
-            disposable->add(d);
+            state->get_observer_under_lock()->set_upstream(d);
         }
 
         bool is_disposed() const
         {
-            return disposable->is_disposed();
+            return state->get_observer_under_lock()->is_disposed();
         }
 
         void on_error(const std::exception_ptr& err) const
         {
-            disposable->get_observer_under_lock()->on_error(err);
-            disposable->dispose();
+            state->get_observer_under_lock()->on_error(err);
         }
 
         void on_completed() const
         {
-            if (disposable->decrement_on_completed())
-            {
-                disposable->get_observer_under_lock()->on_completed();
-                disposable->dispose();
-            }
+            if (state->decrement_on_completed())
+                state->get_observer_under_lock()->on_completed();
         }
     };
 
-    template<template<typename...> typename TDisposable, template<auto, typename...> typename TStrategy, typename TSelector, rpp::constraint::observable... TObservables>
+    template<template<typename...> typename TState, template<auto, typename...> typename TStrategy, typename TSelector, rpp::constraint::observable... TObservables>
     struct combining_operator_t
     {
         RPP_NO_UNIQUE_ADDRESS rpp::utils::tuple<TObservables...> observables;
@@ -93,7 +90,7 @@ namespace rpp::operators::details
         };
 
         template<rpp::details::observables::constraint::disposable_strategy Prev>
-        using updated_disposable_strategy = rpp::details::observables::fixed_disposable_strategy_selector<1>;
+        using updated_disposable_strategy = ::rpp::details::observables::default_disposable_strategy_selector; // TODO: sum of Prev + TObservables
 
         template<rpp::constraint::decayed_type Type, rpp::constraint::observer Observer>
         auto lift(Observer&& observer) const
@@ -105,20 +102,18 @@ namespace rpp::operators::details
         template<rpp::constraint::decayed_type Type, rpp::constraint::observer Observer>
         static auto subscribe_impl(Observer&& observer, const TSelector& selector, const TObservables&... observables)
         {
-            using Disposable = TDisposable<Observer, TSelector, Type, rpp::utils::extract_observable_type_t<TObservables>...>;
+            using State = TState<Observer, TSelector, Type, rpp::utils::extract_observable_type_t<TObservables>...>;
 
-            const auto disposable = disposable_wrapper_impl<Disposable>::make(std::forward<Observer>(observer), selector);
-            auto       locked     = disposable.lock();
-            locked->get_observer_under_lock()->set_upstream(disposable.as_weak());
-            subscribe<std::decay_t<Type>>(locked, std::index_sequence_for<TObservables...>{}, observables...);
+            const auto state = std::make_shared<State>(std::forward<Observer>(observer), selector);
+            subscribe<std::decay_t<Type>>(state, std::index_sequence_for<TObservables...>{}, observables...);
 
-            return rpp::observer<Type, TStrategy<0, std::decay_t<Observer>, TSelector, Type, rpp::utils::extract_observable_type_t<TObservables>...>>{std::move(locked)};
+            return rpp::observer<Type, TStrategy<0, std::decay_t<Observer>, TSelector, Type, rpp::utils::extract_observable_type_t<TObservables>...>>{std::move(state)};
         }
 
         template<typename ExpectedValue, rpp::constraint::observer Observer, size_t... I>
-        static void subscribe(const std::shared_ptr<TDisposable<Observer, TSelector, ExpectedValue, rpp::utils::extract_observable_type_t<TObservables>...>>& disposable, std::index_sequence<I...>, const TObservables&... observables)
+        static void subscribe(const std::shared_ptr<TState<Observer, TSelector, ExpectedValue, rpp::utils::extract_observable_type_t<TObservables>...>>& state, std::index_sequence<I...>, const TObservables&... observables)
         {
-            (..., observables.subscribe(rpp::observer<rpp::utils::extract_observable_type_t<TObservables>, TStrategy<I + 1, Observer, TSelector, ExpectedValue, rpp::utils::extract_observable_type_t<TObservables>...>>{disposable}));
+            (..., observables.subscribe(rpp::observer<rpp::utils::extract_observable_type_t<TObservables>, TStrategy<I + 1, Observer, TSelector, ExpectedValue, rpp::utils::extract_observable_type_t<TObservables>...>>{state}));
         }
     };
 } // namespace rpp::operators::details
diff --git a/src/rpp/rpp/operators/zip.hpp b/src/rpp/rpp/operators/zip.hpp
index f4be887e3..f785dc7ad 100644
--- a/src/rpp/rpp/operators/zip.hpp
+++ b/src/rpp/rpp/operators/zip.hpp
@@ -21,11 +21,11 @@
 namespace rpp::operators::details
 {
     template<rpp::constraint::observer Observer, typename TSelector, rpp::constraint::decayed_type... Args>
-    class zip_disposable final : public combining_disposable<Observer, Args...>
+    class zip_state final : public combining_state<Observer>
     {
     public:
-        explicit zip_disposable(Observer&& observer, const TSelector& selector)
-            : combining_disposable<Observer, Args...>(std::move(observer))
+        explicit zip_state(Observer&& observer, const TSelector& selector)
+            : combining_state<Observer>(std::move(observer), sizeof...(Args))
             , m_selector(selector)
         {
         }
@@ -42,17 +42,17 @@ namespace rpp::operators::details
 
     template<size_t I, rpp::constraint::observer Observer, typename TSelector, rpp::constraint::decayed_type... Args>
     struct zip_observer_strategy final
-        : public combining_observer_strategy<zip_disposable<Observer, TSelector, Args...>>
+        : public combining_observer_strategy<zip_state<Observer, TSelector, Args...>>
     {
-        using combining_observer_strategy<zip_disposable<Observer, TSelector, Args...>>::disposable;
+        using combining_observer_strategy<zip_state<Observer, TSelector, Args...>>::state;
 
         template<typename T>
         void on_next(T&& v) const
         {
-            const auto observer = disposable->get_observer_under_lock();
-            disposable->get_pendings().template get<I>().push_back(std::forward<T>(v));
+            const auto observer = state->get_observer_under_lock();
+            state->get_pendings().template get<I>().push_back(std::forward<T>(v));
 
-            disposable->get_pendings().apply(&apply_impl<decltype(disposable)>, disposable, observer);
+            state->get_pendings().apply(&apply_impl<decltype(state)>, state, observer);
         }
 
     private:
@@ -68,7 +68,7 @@ namespace rpp::operators::details
     };
 
     template<typename TSelector, rpp::constraint::observable... TObservables>
-    struct zip_t : public combining_operator_t<zip_disposable, zip_observer_strategy, TSelector, TObservables...>
+    struct zip_t : public combining_operator_t<zip_state, zip_observer_strategy, TSelector, TObservables...>
     {
     };
 } // namespace rpp::operators::details

From 0719f84ba3ddca8d9c8755cfa19dfb8bdfe7c7f2 Mon Sep 17 00:00:00 2001
From: Aleksey Loginov <victimsnino@gmail.com>
Date: Sun, 15 Sep 2024 21:27:56 +0300
Subject: [PATCH 03/17] fix concat

---
 .../rpp/disposables/refcount_disposable.hpp   | 11 ++++--
 src/rpp/rpp/operators/concat.hpp              | 34 ++++++++-----------
 src/rpp/rpp/sources/concat.hpp                | 21 ++++++------
 3 files changed, 33 insertions(+), 33 deletions(-)

diff --git a/src/rpp/rpp/disposables/refcount_disposable.hpp b/src/rpp/rpp/disposables/refcount_disposable.hpp
index c41521cde..10e327886 100644
--- a/src/rpp/rpp/disposables/refcount_disposable.hpp
+++ b/src/rpp/rpp/disposables/refcount_disposable.hpp
@@ -54,7 +54,12 @@ namespace rpp
         friend class details::refocunt_disposable_inner;
         refcount_disposable() = default;
 
-        composite_disposable_wrapper add_ref();
+        enum class Mode : bool
+        {
+            Weak,
+            Strong
+        };
+        composite_disposable_wrapper add_ref(Mode mode = Mode::Weak);
 
     private:
         std::atomic<size_t>     m_refcount{0};
@@ -91,7 +96,7 @@ namespace rpp::details
 
 namespace rpp
 {
-    inline composite_disposable_wrapper refcount_disposable::add_ref()
+    inline composite_disposable_wrapper refcount_disposable::add_ref(refcount_disposable::Mode mode)
     {
         auto current_value = m_refcount.load(std::memory_order::seq_cst);
         while (true)
@@ -103,7 +108,7 @@ namespace rpp
             if (m_refcount.compare_exchange_strong(current_value, current_value + 1, std::memory_order::seq_cst))
             {
                 auto inner = composite_disposable_wrapper::make<details::refocunt_disposable_inner>(wrapper_from_this());
-                add(inner.as_weak());
+                add(mode == Mode::Weak ? inner.as_weak() : inner);
                 return inner;
             }
         }
diff --git a/src/rpp/rpp/operators/concat.hpp b/src/rpp/rpp/operators/concat.hpp
index b750b5e15..65e8ecde2 100644
--- a/src/rpp/rpp/operators/concat.hpp
+++ b/src/rpp/rpp/operators/concat.hpp
@@ -34,29 +34,33 @@ namespace rpp::operators::details
     };
 
     template<rpp::constraint::observable TObservable, rpp::constraint::observer TObserver>
-    class concat_state_t final : public rpp::refcount_disposable
+    class concat_state_t final : public std::enable_shared_from_this<concat_state_t<TObservable, TObserver>>
     {
     public:
         concat_state_t(TObserver&& observer)
             : m_observer{std::move(observer)}
         {
+            const auto d = disposable_wrapper_impl<refcount_disposable>::make();
+            m_disposable = d.lock();
+            get_observer()->set_upstream(d);
         }
 
         rpp::utils::pointer_under_lock<TObserver>               get_observer() { return m_observer; }
         rpp::utils::pointer_under_lock<std::queue<TObservable>> get_queue() { return m_queue; }
+        const std::shared_ptr<refcount_disposable>& get_disposable() const { return m_disposable; }
 
         std::atomic<ConcatStage>& stage() { return m_stage; }
 
         void drain(rpp::composite_disposable_wrapper refcounted)
         {
-            while (!is_disposed())
+            while (!m_disposable->is_disposed())
             {
                 const auto observable = get_observable();
                 if (!observable)
                 {
                     stage().store(ConcatStage::None, std::memory_order::relaxed);
                     refcounted.dispose();
-                    if (is_disposed())
+                    if (m_disposable->is_disposed())
                         get_observer()->on_completed();
                     return;
                 }
@@ -74,12 +78,13 @@ namespace rpp::operators::details
             drain(refcounted);
         }
 
+
     private:
         bool handle_observable_impl(const rpp::constraint::decayed_same_as<TObservable> auto& observable, rpp::composite_disposable_wrapper refcounted)
         {
             stage().store(ConcatStage::Draining, std::memory_order::relaxed);
             refcounted.clear();
-            observable.subscribe(concat_inner_observer_strategy<TObservable, TObserver>{disposable_wrapper_impl<concat_state_t>{wrapper_from_this()}.lock(), std::move(refcounted)});
+            observable.subscribe(concat_inner_observer_strategy<TObservable, TObserver>{this->shared_from_this(), std::move(refcounted)});
 
             ConcatStage current = ConcatStage::Draining;
             return stage().compare_exchange_strong(current, ConcatStage::Processing, std::memory_order::seq_cst);
@@ -97,6 +102,7 @@ namespace rpp::operators::details
         }
 
     private:
+        std::shared_ptr<refcount_disposable>                  m_disposable{};
         rpp::utils::value_with_mutex<TObserver>               m_observer;
         rpp::utils::value_with_mutex<std::queue<TObservable>> m_queue;
         std::atomic<ConcatStage>                              m_stage{};
@@ -112,7 +118,7 @@ namespace rpp::operators::details
         }
 
         concat_observer_strategy_base(std::shared_ptr<concat_state_t<TObservable, TObserver>> state)
-            : concat_observer_strategy_base{state, state->add_ref()}
+            : concat_observer_strategy_base{state, state->get_disposable()->add_ref(refcount_disposable::Mode::Strong)}
         {
         }
 
@@ -122,7 +128,7 @@ namespace rpp::operators::details
         void on_error(const std::exception_ptr& err) const
         {
             state->get_observer()->on_error(err);
-            state->dispose();
+            state->get_disposable()->dispose();
         }
 
         void set_upstream(const disposable_wrapper& d) const { refcounted.add(d); }
@@ -162,7 +168,7 @@ namespace rpp::operators::details
         using preferred_disposable_strategy = rpp::details::observers::none_disposable_strategy;
 
         concat_observer_strategy(TObserver&& observer)
-            : base{init_state(std::move(observer))}
+            : base{std::make_shared<concat_state_t<TObservable, TObserver>>(std::move(observer))}
         {
         }
 
@@ -171,7 +177,7 @@ namespace rpp::operators::details
         {
             ConcatStage current = ConcatStage::None;
             if (base::state->stage().compare_exchange_strong(current, ConcatStage::Draining, std::memory_order::seq_cst))
-                base::state->handle_observable(std::forward<T>(v), base::state->add_ref());
+                base::state->handle_observable(std::forward<T>(v), base::state->get_disposable()->add_ref(refcount_disposable::Mode::Strong));
             else
                 base::state->get_queue()->push(std::forward<T>(v));
         }
@@ -179,19 +185,9 @@ namespace rpp::operators::details
         void on_completed() const
         {
             base::refcounted.dispose();
-            if (base::state->is_disposed())
+            if (base::state->get_disposable()->is_disposed())
                 base::state->get_observer()->on_completed();
         }
-
-
-    private:
-        static std::shared_ptr<concat_state_t<TObservable, TObserver>> init_state(TObserver&& observer)
-        {
-            const auto d   = disposable_wrapper_impl<concat_state_t<TObservable, TObserver>>::make(std::move(observer));
-            auto       ptr = d.lock();
-            ptr->get_observer()->set_upstream(d.as_weak());
-            return ptr;
-        }
     };
 
     struct concat_t : lift_operator<concat_t>
diff --git a/src/rpp/rpp/sources/concat.hpp b/src/rpp/rpp/sources/concat.hpp
index ef0d6ab94..85b0dac97 100644
--- a/src/rpp/rpp/sources/concat.hpp
+++ b/src/rpp/rpp/sources/concat.hpp
@@ -22,12 +22,13 @@
 namespace rpp::details
 {
     template<rpp::constraint::observer TObserver, constraint::decayed_type PackedContainer>
-    struct concat_state_t : public rpp::composite_disposable
+    struct concat_state_t
     {
         concat_state_t(TObserver&& in_observer, const PackedContainer& in_container)
             : observer(std::move(in_observer))
             , container(in_container)
         {
+            observer.set_upstream(disposable);
             try
             {
                 itr = std::cbegin(container);
@@ -35,12 +36,13 @@ namespace rpp::details
             catch (...)
             {
                 this->observer.on_error(std::current_exception());
-                this->dispose();
+                this->disposable.dispose();
             }
         }
 
         RPP_NO_UNIQUE_ADDRESS TObserver                 observer;
         RPP_NO_UNIQUE_ADDRESS PackedContainer           container;
+        rpp::composite_disposable_wrapper               disposable = composite_disposable_wrapper::make();
         std::optional<decltype(std::cbegin(container))> itr{};
         std::atomic<bool>                               is_inside_drain{};
     };
@@ -68,14 +70,14 @@ namespace rpp::details
             state->observer.on_error(err);
         }
 
-        void set_upstream(const disposable_wrapper& d) { state->add(d); }
+        void set_upstream(const disposable_wrapper& d) { state->disposable.add(d); }
 
-        bool is_disposed() const { return locally_disposed || state->is_disposed(); }
+        bool is_disposed() const { return locally_disposed || state->disposable.is_disposed(); }
 
         void on_completed() const
         {
             locally_disposed = true;
-            state->clear();
+            state->disposable.clear();
 
             if (state->is_inside_drain.exchange(false, std::memory_order::seq_cst))
                 return;
@@ -87,12 +89,12 @@ namespace rpp::details
     template<rpp::constraint::observer TObserver, typename PackedContainer>
     void drain(const std::shared_ptr<concat_state_t<TObserver, PackedContainer>>& state)
     {
-        while (!state->is_disposed())
+        while (!state->disposable.is_disposed())
         {
             if (state->itr.value() == std::cend(state->container))
             {
                 state->observer.on_completed();
-                state->dispose();
+                state->disposable.dispose();
                 return;
             }
 
@@ -130,10 +132,7 @@ namespace rpp::details
         template<constraint::observer_strategy<value_type> Strategy>
         void subscribe(observer<value_type, Strategy>&& obs) const
         {
-            const auto d     = disposable_wrapper_impl<concat_state_t<observer<value_type, Strategy>, PackedContainer>>::make(std::move(obs), container);
-            const auto state = d.lock();
-            state->observer.set_upstream(d.as_weak());
-            drain(state);
+            drain(std::make_shared<concat_state_t<observer<value_type, Strategy>, PackedContainer>>(std::move(obs), container));
         }
     };
 

From 0be57d396408e049737220447e2c30653bcfa037 Mon Sep 17 00:00:00 2001
From: Aleksey Loginov <victimsnino@gmail.com>
Date: Sun, 15 Sep 2024 21:33:43 +0300
Subject: [PATCH 04/17] Debounce

---
 src/rpp/rpp/operators/debounce.hpp | 61 +++++++++++++-----------------
 1 file changed, 27 insertions(+), 34 deletions(-)

diff --git a/src/rpp/rpp/operators/debounce.hpp b/src/rpp/rpp/operators/debounce.hpp
index 1803c6db3..6d36b4dac 100644
--- a/src/rpp/rpp/operators/debounce.hpp
+++ b/src/rpp/rpp/operators/debounce.hpp
@@ -18,27 +18,26 @@
 
 namespace rpp::operators::details
 {
-    template<rpp::constraint::observer Observer, typename Worker, rpp::details::disposables::constraint::disposable_container Container>
-    class debounce_disposable;
+    template<rpp::constraint::observer Observer, typename Worker>
+    class debounce_state;
 
-    template<rpp::constraint::observer Observer, typename Worker, rpp::details::disposables::constraint::disposable_container Container>
-    struct debounce_disposable_wrapper
+    template<rpp::constraint::observer Observer, typename Worker>
+    struct debounce_state_wrapper
     {
-        std::shared_ptr<debounce_disposable<Observer, Worker, Container>> disposable{};
+        std::shared_ptr<debounce_state<Observer, Worker>> state{};
 
-        bool is_disposed() const { return disposable->is_disposed(); }
+        bool is_disposed() const { return state->get_observer_under_lock()->is_disposed(); }
 
-        void on_error(const std::exception_ptr& err) const { disposable->get_observer_under_lock()->on_error(err); }
+        void on_error(const std::exception_ptr& err) const { state->get_observer_under_lock()->on_error(err); }
     };
 
-    template<rpp::constraint::observer Observer, typename Worker, rpp::details::disposables::constraint::disposable_container Container>
-    class debounce_disposable final : public rpp::composite_disposable_impl<Container>
-        , public rpp::details::enable_wrapper_from_this<debounce_disposable<Observer, Worker, Container>>
+    template<rpp::constraint::observer Observer, typename Worker>
+    class debounce_state final : public std::enable_shared_from_this<debounce_state<Observer, Worker>>
     {
         using T = rpp::utils::extract_observer_type_t<Observer>;
 
     public:
-        debounce_disposable(Observer&& in_observer, Worker&& in_worker, rpp::schedulers::duration period)
+        debounce_state(Observer&& in_observer, Worker&& in_worker, rpp::schedulers::duration period)
             : m_observer(std::move(in_observer))
             , m_worker{std::move(in_worker)}
             , m_period{period}
@@ -71,17 +70,17 @@ namespace rpp::operators::details
         {
             m_worker.schedule(
                 m_time_when_value_should_be_emitted.value(),
-                [](const debounce_disposable_wrapper<Observer, Worker, Container>& handler) -> schedulers::optional_delay_to {
-                    auto value_or_duration = handler.disposable->extract_value_or_time();
+                [](const debounce_state_wrapper<Observer, Worker>& handler) -> schedulers::optional_delay_to {
+                    auto value_or_duration = handler.state->extract_value_or_time();
                     if (auto* timepoint = std::get_if<schedulers::time_point>(&value_or_duration))
                         return schedulers::optional_delay_to{*timepoint};
 
                     if (auto* value = std::get_if<T>(&value_or_duration))
-                        handler.disposable->get_observer_under_lock()->on_next(std::move(*value));
+                        handler.state->get_observer_under_lock()->on_next(std::move(*value));
 
                     return std::nullopt;
                 },
-                debounce_disposable_wrapper<Observer, Worker, Container>{this->wrapper_from_this().lock()});
+                debounce_state_wrapper<Observer, Worker>{this->shared_from_this()});
         }
 
         std::variant<std::monostate, T, schedulers::time_point> extract_value_or_time()
@@ -108,41 +107,38 @@ namespace rpp::operators::details
         std::optional<T>                      m_value_to_be_emitted{};
     };
 
-    template<rpp::constraint::observer Observer, typename Worker, rpp::details::disposables::constraint::disposable_container Container>
+    template<rpp::constraint::observer Observer, typename Worker>
     struct debounce_observer_strategy
     {
         using preferred_disposable_strategy = rpp::details::observers::none_disposable_strategy;
 
-        std::shared_ptr<debounce_disposable<Observer, Worker, Container>> disposable{};
+        std::shared_ptr<debounce_state<Observer, Worker>> state{};
 
         void set_upstream(const rpp::disposable_wrapper& d) const
         {
-            disposable->add(d);
+            state->get_observer_under_lock()->set_upstream(d);
         }
 
         bool is_disposed() const
         {
-            return disposable->is_disposed();
+            return state->get_observer_under_lock()->is_disposed();
         }
 
         template<typename T>
         void on_next(T&& v) const
         {
-            disposable->emplace_safe(std::forward<T>(v));
+            state->emplace_safe(std::forward<T>(v));
         }
 
         void on_error(const std::exception_ptr& err) const noexcept
         {
-            disposable->dispose();
-            disposable->get_observer_under_lock()->on_error(err);
+            state->get_observer_under_lock()->on_error(err);
         }
 
         void on_completed() const noexcept
         {
-            disposable->dispose();
-            const auto value    = disposable->extract_value();
-            const auto observer = disposable->get_observer_under_lock();
-            if (value)
+            const auto observer = state->get_observer_under_lock();
+            if (const auto value    = state->extract_value())
                 observer->on_next(std::move(value).value());
             observer->on_completed();
         }
@@ -158,21 +154,18 @@ namespace rpp::operators::details
         };
 
         template<rpp::details::observables::constraint::disposable_strategy Prev>
-        using updated_disposable_strategy = rpp::details::observables::fixed_disposable_strategy_selector<1>;
+        using updated_disposable_strategy = Prev;
 
         rpp::schedulers::duration       duration;
         RPP_NO_UNIQUE_ADDRESS Scheduler scheduler;
 
-        template<rpp::constraint::decayed_type Type, rpp::details::observables::constraint::disposable_strategy DisposableStrategy, rpp::constraint::observer Observer>
-        auto lift_with_disposable_strategy(Observer&& observer) const
+        template<rpp::constraint::decayed_type Type, rpp::constraint::observer Observer>
+        auto lift(Observer&& observer) const
         {
             using worker_t  = rpp::schedulers::utils::get_worker_t<Scheduler>;
-            using container = typename DisposableStrategy::disposable_container;
 
-            const auto disposable = disposable_wrapper_impl<debounce_disposable<std::decay_t<Observer>, worker_t, container>>::make(std::forward<Observer>(observer), scheduler.create_worker(), duration);
-            auto       ptr        = disposable.lock();
-            ptr->get_observer_under_lock()->set_upstream(disposable.as_weak());
-            return rpp::observer<Type, debounce_observer_strategy<std::decay_t<Observer>, worker_t, container>>{std::move(ptr)};
+            auto ptr = std::make_shared<debounce_state<std::decay_t<Observer>, worker_t>>(std::forward<Observer>(observer), scheduler.create_worker(), duration);
+            return rpp::observer<Type, debounce_observer_strategy<std::decay_t<Observer>, worker_t>>{std::move(ptr)};
         }
     };
 } // namespace rpp::operators::details

From 48fddad2744e60591d7767cad2d3b61ca4f95e73 Mon Sep 17 00:00:00 2001
From: Aleksey Loginov <victimsnino@gmail.com>
Date: Sun, 15 Sep 2024 21:39:24 +0300
Subject: [PATCH 05/17] fix delay

---
 src/rpp/rpp/operators/delay.hpp | 77 ++++++++++++++++-----------------
 1 file changed, 37 insertions(+), 40 deletions(-)

diff --git a/src/rpp/rpp/operators/delay.hpp b/src/rpp/rpp/operators/delay.hpp
index 649844ca0..83c0c3b58 100644
--- a/src/rpp/rpp/operators/delay.hpp
+++ b/src/rpp/rpp/operators/delay.hpp
@@ -35,12 +35,12 @@ namespace rpp::operators::details
         rpp::schedulers::time_point                           time_point{};
     };
 
-    template<rpp::constraint::observer Observer, typename Worker, rpp::details::disposables::constraint::disposable_container Container>
-    struct delay_disposable final : public rpp::composite_disposable_impl<Container>
+    template<rpp::constraint::observer Observer, typename Worker>
+    struct delay_state final
     {
         using T = rpp::utils::extract_observer_type_t<Observer>;
 
-        delay_disposable(Observer&& in_observer, Worker&& in_worker, rpp::schedulers::duration delay)
+        delay_state(Observer&& in_observer, Worker&& in_worker, rpp::schedulers::duration delay)
             : observer(std::move(in_observer))
             , worker{std::move(in_worker)}
             , delay{delay}
@@ -56,29 +56,29 @@ namespace rpp::operators::details
         bool                    is_active{};
     };
 
-    template<rpp::constraint::observer Observer, typename Worker, rpp::details::disposables::constraint::disposable_container Container>
-    struct delay_disposable_wrapper
+    template<rpp::constraint::observer Observer, typename Worker>
+    struct delay_state_wrapper
     {
-        std::shared_ptr<delay_disposable<Observer, Worker, Container>> disposable{};
+        std::shared_ptr<delay_state<Observer, Worker>> state{};
 
-        bool is_disposed() const { return disposable->is_disposed(); }
+        bool is_disposed() const { return state->observer.is_disposed(); }
 
-        void on_error(const std::exception_ptr& err) const { disposable->observer.on_error(err); }
+        void on_error(const std::exception_ptr& err) const { state->observer.on_error(err); }
     };
 
-    template<rpp::constraint::observer Observer, typename Worker, rpp::details::disposables::constraint::disposable_container Container, bool ClearOnError>
+    template<rpp::constraint::observer Observer, typename Worker, bool ClearOnError>
     struct delay_observer_strategy
     {
-        std::shared_ptr<delay_disposable<Observer, Worker, Container>> disposable{};
+        std::shared_ptr<delay_state<Observer, Worker>> state{};
 
         void set_upstream(const rpp::disposable_wrapper& d) const
         {
-            disposable->add(d);
+            state->observer.set_upstream(d);
         }
 
         bool is_disposed() const
         {
-            return disposable->is_disposed();
+            return state->observer.is_disposed();
         }
 
         template<typename T>
@@ -103,59 +103,59 @@ namespace rpp::operators::details
         {
             if (const auto tp = emplace_safe(std::forward<TT>(value)))
             {
-                disposable->worker.schedule(
+                state->worker.schedule(
                     tp.value(),
-                    [](const delay_disposable_wrapper<Observer, Worker, Container>& wrapper) { return drain_queue(wrapper.disposable); },
-                    delay_disposable_wrapper<Observer, Worker, Container>{disposable});
+                    [](const delay_state_wrapper<Observer, Worker>& wrapper) { return drain_queue(wrapper.state); },
+                    delay_state_wrapper<Observer, Worker>{state});
             }
         }
 
         template<typename TT>
         std::optional<rpp::schedulers::time_point> emplace_safe(TT&& item) const
         {
-            std::lock_guard lock{disposable->mutex};
+            std::lock_guard lock{state->mutex};
             if constexpr (ClearOnError && rpp::constraint::decayed_same_as<std::exception_ptr, TT>)
             {
-                disposable->queue = std::queue<emission<rpp::utils::extract_observer_type_t<Observer>>>{};
-                disposable->observer.on_error(std::forward<TT>(item));
+                state->queue = std::queue<emission<rpp::utils::extract_observer_type_t<Observer>>>{};
+                state->observer.on_error(std::forward<TT>(item));
                 return std::nullopt;
             }
             else
             {
-                const auto tp = disposable->worker.now() + disposable->delay;
-                disposable->queue.emplace(std::forward<TT>(item), tp);
-                if (!disposable->is_active)
+                const auto tp = state->worker.now() + state->delay;
+                state->queue.emplace(std::forward<TT>(item), tp);
+                if (!state->is_active)
                 {
-                    disposable->is_active = true;
+                    state->is_active = true;
                     return tp;
                 }
                 return std::nullopt;
             }
         }
 
-        static schedulers::optional_delay_to drain_queue(const std::shared_ptr<delay_disposable<Observer, Worker, Container>>& disposable)
+        static schedulers::optional_delay_to drain_queue(const std::shared_ptr<delay_state<Observer, Worker>>& state)
         {
             while (true)
             {
-                std::unique_lock lock{disposable->mutex};
-                if (disposable->queue.empty())
+                std::unique_lock lock{state->mutex};
+                if (state->queue.empty())
                 {
-                    disposable->is_active = false;
+                    state->is_active = false;
                     return std::nullopt;
                 }
 
-                auto& top = disposable->queue.front();
-                if (top.time_point > disposable->worker.now())
+                auto& top = state->queue.front();
+                if (top.time_point > state->worker.now())
                     return schedulers::optional_delay_to{top.time_point};
 
                 auto item = std::move(top.value);
-                disposable->queue.pop();
+                state->queue.pop();
                 lock.unlock();
 
-                std::visit(rpp::utils::overloaded{[&](rpp::utils::extract_observer_type_t<Observer>&& v) { disposable->observer.on_next(std::move(v)); },
-                                                  [&](const std::exception_ptr& err) { disposable->observer.on_error(err); },
+                std::visit(rpp::utils::overloaded{[&](rpp::utils::extract_observer_type_t<Observer>&& v) { state->observer.on_next(std::move(v)); },
+                                                  [&](const std::exception_ptr& err) { state->observer.on_error(err); },
                                                   [&](rpp::utils::none) {
-                                                      disposable->observer.on_completed();
+                                                      state->observer.on_completed();
                                                   }},
                            std::move(item));
             }
@@ -172,21 +172,18 @@ namespace rpp::operators::details
         };
 
         template<rpp::details::observables::constraint::disposable_strategy Prev>
-        using updated_disposable_strategy = rpp::details::observables::fixed_disposable_strategy_selector<1>;
+        using updated_disposable_strategy = Prev;
 
         rpp::schedulers::duration       duration;
         RPP_NO_UNIQUE_ADDRESS Scheduler scheduler;
 
-        template<rpp::constraint::decayed_type Type, rpp::details::observables::constraint::disposable_strategy DisposableStrategy, rpp::constraint::observer Observer>
-        auto lift_with_disposable_strategy(Observer&& observer) const
+        template<rpp::constraint::decayed_type Type, rpp::constraint::observer Observer>
+        auto lift(Observer&& observer) const
         {
             using worker_t  = rpp::schedulers::utils::get_worker_t<Scheduler>;
-            using container = typename DisposableStrategy::disposable_container;
 
-            const auto disposable = disposable_wrapper_impl<delay_disposable<std::decay_t<Observer>, worker_t, container>>::make(std::forward<Observer>(observer), scheduler.create_worker(), duration);
-            auto       ptr        = disposable.lock();
-            ptr->observer.set_upstream(disposable.as_weak());
-            return rpp::observer<Type, delay_observer_strategy<std::decay_t<Observer>, worker_t, container, ClearOnError>>{std::move(ptr)};
+            auto state = std::make_shared<delay_state<std::decay_t<Observer>, worker_t>>(std::forward<Observer>(observer), scheduler.create_worker(), duration);
+            return rpp::observer<Type, delay_observer_strategy<std::decay_t<Observer>, worker_t, ClearOnError>>{std::move(state)};
         }
     };
 } // namespace rpp::operators::details

From fd29804b9995fad24a056316fbf15dd3d18e2d74 Mon Sep 17 00:00:00 2001
From: "pre-commit-ci[bot]"
 <66853113+pre-commit-ci[bot]@users.noreply.github.com>
Date: Sun, 15 Sep 2024 18:43:21 +0000
Subject: [PATCH 06/17] [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci
---
 src/rpp/rpp/operators/concat.hpp   | 2 +-
 src/rpp/rpp/operators/debounce.hpp | 4 ++--
 src/rpp/rpp/operators/delay.hpp    | 2 +-
 3 files changed, 4 insertions(+), 4 deletions(-)

diff --git a/src/rpp/rpp/operators/concat.hpp b/src/rpp/rpp/operators/concat.hpp
index 65e8ecde2..70bda64e9 100644
--- a/src/rpp/rpp/operators/concat.hpp
+++ b/src/rpp/rpp/operators/concat.hpp
@@ -47,7 +47,7 @@ namespace rpp::operators::details
 
         rpp::utils::pointer_under_lock<TObserver>               get_observer() { return m_observer; }
         rpp::utils::pointer_under_lock<std::queue<TObservable>> get_queue() { return m_queue; }
-        const std::shared_ptr<refcount_disposable>& get_disposable() const { return m_disposable; }
+        const std::shared_ptr<refcount_disposable>&             get_disposable() const { return m_disposable; }
 
         std::atomic<ConcatStage>& stage() { return m_stage; }
 
diff --git a/src/rpp/rpp/operators/debounce.hpp b/src/rpp/rpp/operators/debounce.hpp
index 6d36b4dac..a9546a62e 100644
--- a/src/rpp/rpp/operators/debounce.hpp
+++ b/src/rpp/rpp/operators/debounce.hpp
@@ -138,7 +138,7 @@ namespace rpp::operators::details
         void on_completed() const noexcept
         {
             const auto observer = state->get_observer_under_lock();
-            if (const auto value    = state->extract_value())
+            if (const auto value = state->extract_value())
                 observer->on_next(std::move(value).value());
             observer->on_completed();
         }
@@ -162,7 +162,7 @@ namespace rpp::operators::details
         template<rpp::constraint::decayed_type Type, rpp::constraint::observer Observer>
         auto lift(Observer&& observer) const
         {
-            using worker_t  = rpp::schedulers::utils::get_worker_t<Scheduler>;
+            using worker_t = rpp::schedulers::utils::get_worker_t<Scheduler>;
 
             auto ptr = std::make_shared<debounce_state<std::decay_t<Observer>, worker_t>>(std::forward<Observer>(observer), scheduler.create_worker(), duration);
             return rpp::observer<Type, debounce_observer_strategy<std::decay_t<Observer>, worker_t>>{std::move(ptr)};
diff --git a/src/rpp/rpp/operators/delay.hpp b/src/rpp/rpp/operators/delay.hpp
index 83c0c3b58..4bfb9df22 100644
--- a/src/rpp/rpp/operators/delay.hpp
+++ b/src/rpp/rpp/operators/delay.hpp
@@ -180,7 +180,7 @@ namespace rpp::operators::details
         template<rpp::constraint::decayed_type Type, rpp::constraint::observer Observer>
         auto lift(Observer&& observer) const
         {
-            using worker_t  = rpp::schedulers::utils::get_worker_t<Scheduler>;
+            using worker_t = rpp::schedulers::utils::get_worker_t<Scheduler>;
 
             auto state = std::make_shared<delay_state<std::decay_t<Observer>, worker_t>>(std::forward<Observer>(observer), scheduler.create_worker(), duration);
             return rpp::observer<Type, delay_observer_strategy<std::decay_t<Observer>, worker_t, ClearOnError>>{std::move(state)};

From 30d7a1e72a02036c2df8beb4d189cf65adf6cfb1 Mon Sep 17 00:00:00 2001
From: Aleksey Loginov <victimsnino@gmail.com>
Date: Sun, 15 Sep 2024 21:57:35 +0300
Subject: [PATCH 07/17] fix merge

---
 src/rpp/rpp/operators/merge.hpp | 53 +++++++++++++++------------------
 1 file changed, 24 insertions(+), 29 deletions(-)

diff --git a/src/rpp/rpp/operators/merge.hpp b/src/rpp/rpp/operators/merge.hpp
index d7715098c..89ce4ab60 100644
--- a/src/rpp/rpp/operators/merge.hpp
+++ b/src/rpp/rpp/operators/merge.hpp
@@ -13,7 +13,6 @@
 #include <rpp/operators/fwd.hpp>
 
 #include <rpp/defs.hpp>
-#include <rpp/disposables/composite_disposable.hpp>
 #include <rpp/operators/details/strategy.hpp>
 #include <rpp/schedulers/current_thread.hpp>
 #include <rpp/utils/tuple.hpp>
@@ -24,12 +23,13 @@
 namespace rpp::operators::details
 {
     template<rpp::constraint::observer TObserver>
-    class merge_disposable final : public composite_disposable
+    class merge_state final
     {
     public:
-        merge_disposable(TObserver&& observer)
+        merge_state(TObserver&& observer)
             : m_observer(std::move(observer))
         {
+            get_observer_under_lock()->set_upstream(m_disposable);
         }
 
         // just need atomicity, not guarding anything
@@ -40,56 +40,60 @@ namespace rpp::operators::details
 
         rpp::utils::pointer_under_lock<TObserver> get_observer_under_lock() { return m_observer; }
 
+        const rpp::composite_disposable_wrapper& get_disposable() const { return m_disposable; }
+
     private:
         rpp::utils::value_with_mutex<TObserver> m_observer{};
+        rpp::composite_disposable_wrapper       m_disposable = composite_disposable_wrapper::make();
         std::atomic_size_t                      m_on_completed_needed{1};
     };
 
     template<rpp::constraint::observer TObserver>
     struct merge_observer_base_strategy
     {
-        merge_observer_base_strategy(std::shared_ptr<merge_disposable<TObserver>>&& disposable)
-            : m_disposable{std::move(disposable)}
+        merge_observer_base_strategy(std::shared_ptr<merge_state<TObserver>>&& state)
+            : m_state{std::move(state)}
         {
         }
 
-        merge_observer_base_strategy(const std::shared_ptr<merge_disposable<TObserver>>& disposable)
-            : m_disposable{disposable}
+        merge_observer_base_strategy(const std::shared_ptr<merge_state<TObserver>>& state)
+            : m_state{state}
         {
         }
 
         void set_upstream(const rpp::disposable_wrapper& d) const
         {
-            m_disposable->add(d);
+            m_state->get_disposable().add(d);
             m_disposables.push_back(d);
         }
 
         bool is_disposed() const
         {
-            return m_disposable->is_disposed();
+            return m_state->get_observer_under_lock()->is_disposed();
         }
 
         void on_error(const std::exception_ptr& err) const
         {
-            m_disposable->get_observer_under_lock()->on_error(err);
-            m_disposable->dispose();
+            m_state->get_observer_under_lock()->on_error(err);
         }
 
         void on_completed() const
         {
-            if (m_disposable->decrement_on_completed())
+            if (m_state->decrement_on_completed())
+            {
+                m_state->get_observer_under_lock()->on_completed();
+            }
+            else
             {
                 for (const auto& v : m_disposables)
                 {
-                    m_disposable->remove(v);
+                    m_state->get_disposable().remove(v);
                 }
-                m_disposable->get_observer_under_lock()->on_completed();
-                m_disposable->dispose();
             }
         }
 
     protected:
-        std::shared_ptr<merge_disposable<TObserver>> m_disposable;
+        std::shared_ptr<merge_state<TObserver>> m_state;
         mutable std::vector<rpp::disposable_wrapper> m_disposables{};
     };
 
@@ -101,7 +105,7 @@ namespace rpp::operators::details
         template<typename T>
         void on_next(T&& v) const
         {
-            merge_observer_base_strategy<TObserver>::m_disposable->get_observer_under_lock()->on_next(std::forward<T>(v));
+            merge_observer_base_strategy<TObserver>::m_state->get_observer_under_lock()->on_next(std::forward<T>(v));
         }
     };
 
@@ -110,24 +114,15 @@ namespace rpp::operators::details
     {
     public:
         explicit merge_observer_strategy(TObserver&& observer)
-            : merge_observer_base_strategy<TObserver>{init_state(std::move(observer))}
+            : merge_observer_base_strategy<TObserver>{std::make_shared<merge_state<TObserver>>(std::move(observer))}
         {
         }
 
         template<typename T>
         void on_next(T&& v) const
         {
-            merge_observer_base_strategy<TObserver>::m_disposable->increment_on_completed();
-            std::forward<T>(v).subscribe(rpp::observer<rpp::utils::extract_observer_type_t<TObserver>, merge_observer_inner_strategy<TObserver>>{merge_observer_inner_strategy<TObserver>{merge_observer_base_strategy<TObserver>::m_disposable}});
-        }
-
-    private:
-        static std::shared_ptr<merge_disposable<TObserver>> init_state(TObserver&& observer)
-        {
-            const auto d   = disposable_wrapper_impl<merge_disposable<TObserver>>::make(std::move(observer));
-            auto       ptr = d.lock();
-            ptr->get_observer_under_lock()->set_upstream(d.as_weak());
-            return ptr;
+            merge_observer_base_strategy<TObserver>::m_state->increment_on_completed();
+            std::forward<T>(v).subscribe(rpp::observer<rpp::utils::extract_observer_type_t<TObserver>, merge_observer_inner_strategy<TObserver>>{merge_observer_inner_strategy<TObserver>{merge_observer_base_strategy<TObserver>::m_state}});
         }
     };
 

From 48837bd2bb73e0d7c528b4f89fd36a13ce8c241c Mon Sep 17 00:00:00 2001
From: "pre-commit-ci[bot]"
 <66853113+pre-commit-ci[bot]@users.noreply.github.com>
Date: Sun, 15 Sep 2024 18:57:21 +0000
Subject: [PATCH 08/17] [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci
---
 src/rpp/rpp/operators/merge.hpp | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/src/rpp/rpp/operators/merge.hpp b/src/rpp/rpp/operators/merge.hpp
index 89ce4ab60..791a65f84 100644
--- a/src/rpp/rpp/operators/merge.hpp
+++ b/src/rpp/rpp/operators/merge.hpp
@@ -93,7 +93,7 @@ namespace rpp::operators::details
         }
 
     protected:
-        std::shared_ptr<merge_state<TObserver>> m_state;
+        std::shared_ptr<merge_state<TObserver>>      m_state;
         mutable std::vector<rpp::disposable_wrapper> m_disposables{};
     };
 

From c959bb25d7b4b2a4f62f498c6262665afe5a718e Mon Sep 17 00:00:00 2001
From: Aleksey Loginov <victimsnino@gmail.com>
Date: Sun, 15 Sep 2024 22:01:14 +0300
Subject: [PATCH 09/17] includes

---
 src/rpp/rpp/operators/debounce.hpp                   | 1 -
 src/rpp/rpp/operators/delay.hpp                      | 1 -
 src/rpp/rpp/operators/details/combining_strategy.hpp | 1 -
 3 files changed, 3 deletions(-)

diff --git a/src/rpp/rpp/operators/debounce.hpp b/src/rpp/rpp/operators/debounce.hpp
index a9546a62e..24c1a9d34 100644
--- a/src/rpp/rpp/operators/debounce.hpp
+++ b/src/rpp/rpp/operators/debounce.hpp
@@ -12,7 +12,6 @@
 
 #include <rpp/operators/fwd.hpp>
 
-#include <rpp/disposables/composite_disposable.hpp>
 #include <rpp/operators/details/strategy.hpp>
 #include <rpp/utils/utils.hpp>
 
diff --git a/src/rpp/rpp/operators/delay.hpp b/src/rpp/rpp/operators/delay.hpp
index 4bfb9df22..f92106414 100644
--- a/src/rpp/rpp/operators/delay.hpp
+++ b/src/rpp/rpp/operators/delay.hpp
@@ -13,7 +13,6 @@
 #include <rpp/operators/fwd.hpp>
 
 #include <rpp/defs.hpp>
-#include <rpp/disposables/composite_disposable.hpp>
 #include <rpp/operators/details/strategy.hpp>
 
 #include <mutex>
diff --git a/src/rpp/rpp/operators/details/combining_strategy.hpp b/src/rpp/rpp/operators/details/combining_strategy.hpp
index d3a0b1c92..4aa03b752 100644
--- a/src/rpp/rpp/operators/details/combining_strategy.hpp
+++ b/src/rpp/rpp/operators/details/combining_strategy.hpp
@@ -13,7 +13,6 @@
 #include <rpp/operators/fwd.hpp>
 
 #include <rpp/defs.hpp>
-#include <rpp/disposables/composite_disposable.hpp>
 #include <rpp/operators/details/strategy.hpp>
 #include <rpp/schedulers/current_thread.hpp>
 #include <rpp/utils/utils.hpp>

From 0d3643cf9ecebb02263118870687b64ba0b69ec9 Mon Sep 17 00:00:00 2001
From: Aleksey Loginov <victimsnino@gmail.com>
Date: Sun, 15 Sep 2024 22:06:51 +0300
Subject: [PATCH 10/17] on_error_resume

---
 .../rpp/operators/on_error_resume_next.hpp    | 40 +++++--------------
 1 file changed, 9 insertions(+), 31 deletions(-)

diff --git a/src/rpp/rpp/operators/on_error_resume_next.hpp b/src/rpp/rpp/operators/on_error_resume_next.hpp
index c010613ca..23d1c820b 100644
--- a/src/rpp/rpp/operators/on_error_resume_next.hpp
+++ b/src/rpp/rpp/operators/on_error_resume_next.hpp
@@ -17,18 +17,6 @@
 
 namespace rpp::operators::details
 {
-    template<rpp::constraint::observer TObserver>
-    struct on_error_resume_next_disposable final : public rpp::composite_disposable
-    {
-        on_error_resume_next_disposable(TObserver&& observer)
-            : rpp::composite_disposable{}
-            , observer(std::move(observer))
-        {
-        }
-
-        RPP_NO_UNIQUE_ADDRESS TObserver observer;
-    };
-
     template<rpp::constraint::observer TObserver>
     struct on_error_resume_next_inner_observer_strategy
     {
@@ -64,53 +52,43 @@ namespace rpp::operators::details
         using preferred_disposable_strategy = rpp::details::observers::none_disposable_strategy;
 
         on_error_resume_next_observer_strategy(TObserver&& observer, const Selector& selector)
-            : state{init_state(std::move(observer))}
+            : state{std::make_shared<TObserver>(std::move(observer))}
             , selector{selector}
         {
         }
 
-        std::shared_ptr<on_error_resume_next_disposable<TObserver>> state;
-        RPP_NO_UNIQUE_ADDRESS Selector                              selector;
+        std::shared_ptr<TObserver>     state;
+        RPP_NO_UNIQUE_ADDRESS Selector selector;
 
         template<typename T>
         void on_next(T&& v) const
         {
-            state->observer.on_next(std::forward<T>(v));
+            state->on_next(std::forward<T>(v));
         }
 
         void on_error(const std::exception_ptr& err) const
         {
             try
             {
-                selector(err).subscribe(on_error_resume_next_inner_observer_strategy<TObserver>{std::shared_ptr<TObserver>(state, &state->observer)});
+                selector(err).subscribe(on_error_resume_next_inner_observer_strategy<TObserver>{state});
             }
             catch (...)
             {
-                state->observer.on_error(std::current_exception());
+                state->on_error(std::current_exception());
             }
-            state->dispose();
         }
 
         void on_completed() const
         {
-            state->observer.on_completed();
-            state->dispose();
+            state->on_completed();
         }
 
         void set_upstream(const disposable_wrapper& d) const
         {
-            state->add(d);
+            state->set_upstream(d);
         }
 
         bool is_disposed() const { return state->is_disposed(); }
-
-        static std::shared_ptr<on_error_resume_next_disposable<TObserver>> init_state(TObserver&& observer)
-        {
-            const auto d   = disposable_wrapper_impl<on_error_resume_next_disposable<TObserver>>::make(std::move(observer));
-            auto       ptr = d.lock();
-            ptr->observer.set_upstream(d.as_weak());
-            return ptr;
-        }
     };
 
     template<rpp::constraint::decayed_type Selector>
@@ -135,7 +113,7 @@ namespace rpp::operators::details
         };
 
         template<rpp::details::observables::constraint::disposable_strategy Prev>
-        using updated_disposable_strategy = rpp::details::observables::atomic_dynamic_disposable_strategy_selector<1>;
+        using updated_disposable_strategy = rpp::details::observables::default_disposable_strategy_selector;
     };
 } // namespace rpp::operators::details
 

From ab2a111a0c27ecbbc0a3635af9580b928d58516b Mon Sep 17 00:00:00 2001
From: Aleksey Loginov <victimsnino@gmail.com>
Date: Sun, 15 Sep 2024 22:10:36 +0300
Subject: [PATCH 11/17] retry

---
 src/rpp/rpp/operators/retry.hpp | 19 +++++++++----------
 1 file changed, 9 insertions(+), 10 deletions(-)

diff --git a/src/rpp/rpp/operators/retry.hpp b/src/rpp/rpp/operators/retry.hpp
index 686daec58..89a34dca7 100644
--- a/src/rpp/rpp/operators/retry.hpp
+++ b/src/rpp/rpp/operators/retry.hpp
@@ -18,7 +18,7 @@
 namespace rpp::operators::details
 {
     template<rpp::constraint::observer TObserver, constraint::decayed_type Observable>
-    struct retry_state_t final : public rpp::composite_disposable
+    struct retry_state_t final
     {
         retry_state_t(TObserver&& in_observer, const Observable& observable, std::optional<size_t> count)
             : count{count}
@@ -26,6 +26,7 @@ namespace rpp::operators::details
             , observable(observable)
 
         {
+            observer.set_upstream(disposable);
         }
 
         std::optional<size_t> count;
@@ -33,6 +34,8 @@ namespace rpp::operators::details
 
         RPP_NO_UNIQUE_ADDRESS TObserver  observer;
         RPP_NO_UNIQUE_ADDRESS Observable observable;
+
+        rpp::composite_disposable_wrapper disposable = composite_disposable_wrapper::make();
     };
 
     template<rpp::constraint::observer TObserver, typename TObservable>
@@ -58,11 +61,10 @@ namespace rpp::operators::details
             if (state->count == 0)
             {
                 state->observer.on_error(err);
-                state->dispose();
                 return;
             }
 
-            state->clear();
+            state->disposable.clear();
 
             if (state->is_inside_drain.exchange(false, std::memory_order::seq_cst))
                 return;
@@ -78,16 +80,16 @@ namespace rpp::operators::details
 
         void set_upstream(const disposable_wrapper& d) const
         {
-            state->add(d);
+            state->disposable.add(d);
         }
 
-        bool is_disposed() const { return locally_disposed || state->is_disposed(); }
+        bool is_disposed() const { return locally_disposed || state->disposable.is_disposed(); }
     };
 
     template<rpp::constraint::observer TObserver, typename TObservable>
     void drain(const std::shared_ptr<retry_state_t<TObserver, TObservable>>& state)
     {
-        while (!state->is_disposed())
+        while (!state->disposable.is_disposed())
         {
             if (state->count)
                 --state->count.value();
@@ -124,10 +126,7 @@ namespace rpp::operators::details
         template<rpp::constraint::observer TObserver, typename TObservable>
         void subscribe(TObserver&& observer, TObservable&& observble) const
         {
-            const auto d   = disposable_wrapper_impl<retry_state_t<std::decay_t<TObserver>, std::decay_t<TObservable>>>::make(std::forward<TObserver>(observer), std::forward<TObservable>(observble), count ? count.value() + 1 : count);
-            auto       ptr = d.lock();
-
-            ptr->observer.set_upstream(d.as_weak());
+            const auto ptr   = std::make_shared<retry_state_t<std::decay_t<TObserver>, std::decay_t<TObservable>>>(std::forward<TObserver>(observer), std::forward<TObservable>(observble), count ? count.value() + 1 : count);
             drain(ptr);
         }
     };

From 826dce540539be0172c1cfb4f458d3dec941510a Mon Sep 17 00:00:00 2001
From: Aleksey Loginov <victimsnino@gmail.com>
Date: Sun, 15 Sep 2024 22:13:38 +0300
Subject: [PATCH 12/17] retry_when

---
 src/rpp/rpp/operators/retry_when.hpp | 27 ++++++++++++++-------------
 1 file changed, 14 insertions(+), 13 deletions(-)

diff --git a/src/rpp/rpp/operators/retry_when.hpp b/src/rpp/rpp/operators/retry_when.hpp
index bf481db07..c09633f51 100644
--- a/src/rpp/rpp/operators/retry_when.hpp
+++ b/src/rpp/rpp/operators/retry_when.hpp
@@ -23,13 +23,14 @@ namespace rpp::operators::details
     template<rpp::constraint::observer TObserver,
              typename TObservable,
              typename TNotifier>
-    struct retry_when_state final : public rpp::composite_disposable
+    struct retry_when_state final
     {
-        retry_when_state(TObserver&& observer, const TObservable& observable, const TNotifier& notifier)
-            : observer(std::move(observer))
+        retry_when_state(TObserver&& in_observer, const TObservable& observable, const TNotifier& notifier)
+            : observer(std::move(in_observer))
             , observable(observable)
             , notifier(notifier)
         {
+            observer.set_upstream(disposable);
         }
 
         std::atomic_bool is_inside_drain{};
@@ -37,6 +38,9 @@ namespace rpp::operators::details
         RPP_NO_UNIQUE_ADDRESS TObserver   observer;
         RPP_NO_UNIQUE_ADDRESS TObservable observable;
         RPP_NO_UNIQUE_ADDRESS TNotifier   notifier;
+
+        rpp::composite_disposable_wrapper disposable = composite_disposable_wrapper::make();
+
     };
 
     template<rpp::constraint::observer TObserver, typename TObservable, typename TNotifier>
@@ -56,7 +60,7 @@ namespace rpp::operators::details
         void on_next(T&&) const
         {
             locally_disposed = true;
-            state->clear();
+            state->disposable.clear();
 
             if (state->is_inside_drain.exchange(false, std::memory_order::seq_cst))
                 return;
@@ -76,9 +80,9 @@ namespace rpp::operators::details
             state->observer.on_completed();
         }
 
-        void set_upstream(const disposable_wrapper& d) const { state->add(d); }
+        void set_upstream(const disposable_wrapper& d) const { state->disposable.add(d); }
 
-        bool is_disposed() const { return locally_disposed || state->is_disposed(); }
+        bool is_disposed() const { return locally_disposed || state->disposable.is_disposed(); }
     };
 
     template<rpp::constraint::observer TObserver,
@@ -113,15 +117,15 @@ namespace rpp::operators::details
             state->observer.on_completed();
         }
 
-        void set_upstream(const disposable_wrapper& d) { state->add(d); }
+        void set_upstream(const disposable_wrapper& d) { state->disposable.add(d); }
 
-        bool is_disposed() const { return state->is_disposed(); }
+        bool is_disposed() const { return state->disposable.is_disposed(); }
     };
 
     template<rpp::constraint::observer TObserver, typename TObservable, typename TNotifier>
     void drain(const std::shared_ptr<retry_when_state<TObserver, TObservable, TNotifier>>& state)
     {
-        while (!state->is_disposed())
+        while (!state->disposable.is_disposed())
         {
             state->is_inside_drain.store(true, std::memory_order::seq_cst);
             try
@@ -157,10 +161,7 @@ namespace rpp::operators::details
         template<rpp::constraint::observer TObserver, typename TObservable>
         void subscribe(TObserver&& observer, TObservable&& observable) const
         {
-            const auto d   = disposable_wrapper_impl<retry_when_state<std::decay_t<TObserver>, std::decay_t<TObservable>, std::decay_t<TNotifier>>>::make(std::forward<TObserver>(observer), std::forward<TObservable>(observable), notifier);
-            auto       ptr = d.lock();
-
-            ptr->observer.set_upstream(d.as_weak());
+            const auto ptr   = std::make_shared<retry_when_state<std::decay_t<TObserver>, std::decay_t<TObservable>, std::decay_t<TNotifier>>>(std::forward<TObserver>(observer), std::forward<TObservable>(observable), notifier);
             drain(ptr);
         }
     };

From dc2e3edaac1780fe50885425f60410563567d5c2 Mon Sep 17 00:00:00 2001
From: Aleksey Loginov <victimsnino@gmail.com>
Date: Sun, 15 Sep 2024 22:28:58 +0300
Subject: [PATCH 13/17] take_until

---
 src/rpp/rpp/operators/retry_when.hpp |  1 -
 src/rpp/rpp/operators/take_until.hpp | 19 ++++++++-----------
 src/tests/rpp/test_take_until.cpp    |  6 ++++--
 3 files changed, 12 insertions(+), 14 deletions(-)

diff --git a/src/rpp/rpp/operators/retry_when.hpp b/src/rpp/rpp/operators/retry_when.hpp
index c09633f51..62c2ae595 100644
--- a/src/rpp/rpp/operators/retry_when.hpp
+++ b/src/rpp/rpp/operators/retry_when.hpp
@@ -40,7 +40,6 @@ namespace rpp::operators::details
         RPP_NO_UNIQUE_ADDRESS TNotifier   notifier;
 
         rpp::composite_disposable_wrapper disposable = composite_disposable_wrapper::make();
-
     };
 
     template<rpp::constraint::observer TObserver, typename TObservable, typename TNotifier>
diff --git a/src/rpp/rpp/operators/take_until.hpp b/src/rpp/rpp/operators/take_until.hpp
index ed2aca2c2..c8c20e8e9 100644
--- a/src/rpp/rpp/operators/take_until.hpp
+++ b/src/rpp/rpp/operators/take_until.hpp
@@ -13,22 +13,21 @@
 #include <rpp/operators/fwd.hpp>
 
 #include <rpp/defs.hpp>
-#include <rpp/disposables/composite_disposable.hpp>
 #include <rpp/schedulers/current_thread.hpp>
 #include <rpp/utils/utils.hpp>
 
 namespace rpp::operators::details
 {
     template<rpp::constraint::observer TObserver>
-    class take_until_disposable final : public rpp::composite_disposable
+    class take_until_state final
     {
     public:
-        take_until_disposable(TObserver&& observer)
+        take_until_state(TObserver&& observer)
             : m_observer_with_mutex(std::move(observer))
         {
         }
 
-        take_until_disposable(const TObserver& observer)
+        take_until_state(const TObserver& observer)
             : m_observer_with_mutex(observer)
         {
         }
@@ -49,7 +48,7 @@ namespace rpp::operators::details
     {
         using preferred_disposable_strategy = rpp::details::observers::none_disposable_strategy;
 
-        std::shared_ptr<take_until_disposable<TObserver>> state;
+        std::shared_ptr<take_until_state<TObserver>> state;
 
         void on_error(const std::exception_ptr& err) const
         {
@@ -63,9 +62,9 @@ namespace rpp::operators::details
                 state->get_observer()->on_completed();
         }
 
-        void set_upstream(const disposable_wrapper& d) { state->add(d); }
+        void set_upstream(const disposable_wrapper& d) { state->get_observer()->set_upstream(d); }
 
-        bool is_disposed() const { return state->is_disposed(); }
+        bool is_disposed() const { return state->get_observer()->is_disposed(); }
     };
 
     template<rpp::constraint::observer TObserver>
@@ -104,14 +103,12 @@ namespace rpp::operators::details
         };
 
         template<rpp::details::observables::constraint::disposable_strategy Prev>
-        using updated_disposable_strategy = rpp::details::observables::fixed_disposable_strategy_selector<1>;
+        using updated_disposable_strategy = rpp::details::observables::default_disposable_strategy_selector;
 
         template<rpp::constraint::decayed_type Type, rpp::constraint::observer Observer>
         auto lift(Observer&& observer) const
         {
-            const auto d   = disposable_wrapper_impl<take_until_disposable<std::decay_t<Observer>>>::make(std::forward<Observer>(observer));
-            auto       ptr = d.lock();
-            ptr->get_observer()->set_upstream(d.as_weak());
+            auto ptr = std::make_shared<take_until_state<std::decay_t<Observer>>>(std::forward<Observer>(observer));
 
             observable.subscribe(take_until_throttle_observer_strategy<std::decay_t<Observer>>{ptr});
             return rpp::observer<Type, take_until_observer_strategy<std::decay_t<Observer>>>(std::move(ptr));
diff --git a/src/tests/rpp/test_take_until.cpp b/src/tests/rpp/test_take_until.cpp
index 7d65a96a8..46509393d 100644
--- a/src/tests/rpp/test_take_until.cpp
+++ b/src/tests/rpp/test_take_until.cpp
@@ -14,10 +14,12 @@
 #include <rpp/observers/mock_observer.hpp>
 #include <rpp/operators/as_blocking.hpp>
 #include <rpp/operators/finally.hpp>
+#include <rpp/operators/take.hpp>
 #include <rpp/operators/take_until.hpp>
 #include <rpp/schedulers/current_thread.hpp>
 #include <rpp/sources/empty.hpp>
 #include <rpp/sources/error.hpp>
+#include <rpp/sources/concat.hpp>
 #include <rpp/sources/interval.hpp>
 #include <rpp/sources/just.hpp>
 #include <rpp/sources/never.hpp>
@@ -146,7 +148,7 @@ TEST_CASE("take_until can handle race condition")
 
         SECTION("on_completed shall not interleave with on_next")
         {
-            rpp::source::interval(std::chrono::milliseconds{200}, rpp::schedulers::current_thread{})
+                rpp::source::concat(rpp::source::just(1) | rpp::ops::take(1), rpp::source::never<int>())
                 | rpp::ops::take_until(subject.get_observable())
                 | rpp::ops::as_blocking()
                 | rpp::ops::subscribe([&](auto&&) {
@@ -173,7 +175,7 @@ TEST_CASE("take_until can handle race condition")
 
         SECTION("on_error shall not interleave with on_next")
         {
-            rpp::source::interval(std::chrono::milliseconds{200}, rpp::schedulers::current_thread{})
+            rpp::source::concat(rpp::source::just(1) | rpp::ops::take(1), rpp::source::never<int>())
                 | rpp::ops::take_until(subject.get_observable())
                 | rpp::ops::as_blocking()
                 | rpp::ops::subscribe([&](auto&&) {

From 5d6fafd7a0fae4baef479ffb4beadd0720d8b839 Mon Sep 17 00:00:00 2001
From: Aleksey Loginov <victimsnino@gmail.com>
Date: Sun, 15 Sep 2024 22:32:15 +0300
Subject: [PATCH 14/17] with_latest_from

---
 src/rpp/rpp/operators/with_latest_from.hpp | 48 +++++++++++-----------
 1 file changed, 23 insertions(+), 25 deletions(-)

diff --git a/src/rpp/rpp/operators/with_latest_from.hpp b/src/rpp/rpp/operators/with_latest_from.hpp
index 78ae5bafe..5c553d6c6 100644
--- a/src/rpp/rpp/operators/with_latest_from.hpp
+++ b/src/rpp/rpp/operators/with_latest_from.hpp
@@ -23,13 +23,14 @@
 namespace rpp::operators::details
 {
     template<rpp::constraint::observer Observer, typename TSelector, rpp::constraint::decayed_type... RestArgs>
-    class with_latest_from_disposable final : public composite_disposable
+    class with_latest_from_state final
     {
     public:
-        explicit with_latest_from_disposable(Observer&& observer, const TSelector& selector)
+        explicit with_latest_from_state(Observer&& observer, const TSelector& selector)
             : m_observer_with_mutex{std::move(observer)}
             , m_selector{selector}
         {
+            get_observer_under_lock()->set_upstream(m_disposable);
         }
 
         rpp::utils::pointer_under_lock<Observer> get_observer_under_lock() { return m_observer_with_mutex; }
@@ -37,39 +38,40 @@ namespace rpp::operators::details
         rpp::utils::tuple<rpp::utils::value_with_mutex<std::optional<RestArgs>>...>& get_values() { return m_values; }
 
         const TSelector& get_selector() const { return m_selector; }
+        const composite_disposable_wrapper& get_disposable() const { return m_disposable; }
 
     private:
         rpp::utils::value_with_mutex<Observer>                                      m_observer_with_mutex{};
         rpp::utils::tuple<rpp::utils::value_with_mutex<std::optional<RestArgs>>...> m_values{};
+        composite_disposable_wrapper                                                m_disposable = composite_disposable_wrapper::make();
         RPP_NO_UNIQUE_ADDRESS TSelector                                             m_selector;
     };
 
     template<size_t I, rpp::constraint::observer Observer, typename TSelector, rpp::constraint::decayed_type... RestArgs>
     struct with_latest_from_inner_observer_strategy
     {
-        std::shared_ptr<with_latest_from_disposable<Observer, TSelector, RestArgs...>> disposable{};
+        std::shared_ptr<with_latest_from_state<Observer, TSelector, RestArgs...>> state{};
 
         void set_upstream(const rpp::disposable_wrapper& d) const
         {
-            disposable->add(d);
+            state->get_disposable().add(d);
         }
 
         bool is_disposed() const
         {
-            return disposable->is_disposed();
+            return state->get_disposable().is_disposed();
         }
 
         template<typename T>
         void on_next(T&& v) const
         {
-            auto locked_value = disposable->get_values().template get<I>().lock();
+            auto locked_value = state->get_values().template get<I>().lock();
             locked_value->emplace(std::forward<T>(v));
         }
 
         void on_error(const std::exception_ptr& err) const
         {
-            disposable->get_observer_under_lock()->on_error(err);
-            disposable->dispose();
+            state->get_observer_under_lock()->on_error(err);
         }
 
         static constexpr rpp::utils::empty_function_t<> on_completed{};
@@ -79,26 +81,26 @@ namespace rpp::operators::details
         requires std::invocable<TSelector, OriginalValue, RestArgs...>
     struct with_latest_from_observer_strategy
     {
-        using Disposable                    = with_latest_from_disposable<Observer, TSelector, RestArgs...>;
+        using Disposable                    = with_latest_from_state<Observer, TSelector, RestArgs...>;
         using Result                        = std::invoke_result_t<TSelector, OriginalValue, RestArgs...>;
         using preferred_disposable_strategy = rpp::details::observers::none_disposable_strategy;
 
-        std::shared_ptr<Disposable> disposable{};
+        std::shared_ptr<Disposable> state{};
 
         void set_upstream(const rpp::disposable_wrapper& d) const
         {
-            disposable->add(d);
+            state->get_disposable().add(d);
         }
 
         bool is_disposed() const
         {
-            return disposable->is_disposed();
+            return state->get_disposable().is_disposed();
         }
 
         template<typename T>
         void on_next(T&& v) const
         {
-            auto result = disposable->get_values().apply([&d = this->disposable, &v](rpp::utils::value_with_mutex<std::optional<RestArgs>>&... vals) -> std::optional<Result> {
+            auto result = state->get_values().apply([&d = this->state, &v](rpp::utils::value_with_mutex<std::optional<RestArgs>>&... vals) -> std::optional<Result> {
                 auto lock = std::scoped_lock{vals.get_mutex()...};
 
                 if ((vals.get_value_unsafe().has_value() && ...))
@@ -107,19 +109,17 @@ namespace rpp::operators::details
             });
 
             if (result.has_value())
-                disposable->get_observer_under_lock()->on_next(std::move(result).value());
+                state->get_observer_under_lock()->on_next(std::move(result).value());
         }
 
         void on_error(const std::exception_ptr& err) const
         {
-            disposable->get_observer_under_lock()->on_error(err);
-            disposable->dispose();
+            state->get_observer_under_lock()->on_error(err);
         }
 
         void on_completed() const
         {
-            disposable->get_observer_under_lock()->on_completed();
-            disposable->dispose();
+            state->get_observer_under_lock()->on_completed();
         }
     };
 
@@ -140,7 +140,7 @@ namespace rpp::operators::details
         };
 
         template<rpp::details::observables::constraint::disposable_strategy Prev>
-        using updated_disposable_strategy = rpp::details::observables::fixed_disposable_strategy_selector<1>;
+        using updated_disposable_strategy = rpp::details::observables::default_disposable_strategy_selector;
 
         template<rpp::constraint::decayed_type Type, rpp::constraint::observer Observer>
         auto lift(Observer&& observer) const
@@ -152,20 +152,18 @@ namespace rpp::operators::details
         template<rpp::constraint::decayed_type Type, rpp::constraint::observer Observer>
         static auto subscribe_impl(Observer&& observer, const TSelector& selector, const TObservables&... observables)
         {
-            using Disposable = with_latest_from_disposable<Observer, TSelector, rpp::utils::extract_observable_type_t<TObservables>...>;
+            using State = with_latest_from_state<Observer, TSelector, rpp::utils::extract_observable_type_t<TObservables>...>;
 
-            const auto disposable = disposable_wrapper_impl<Disposable>::make(std::forward<Observer>(observer), selector);
-            auto       ptr        = disposable.lock();
-            ptr->get_observer_under_lock()->set_upstream(disposable.as_weak());
+            auto ptr = std::make_shared<State>(std::forward<Observer>(observer), selector);
             subscribe(ptr, std::index_sequence_for<TObservables...>{}, observables...);
 
             return rpp::observer<Type, with_latest_from_observer_strategy<std::decay_t<Observer>, TSelector, Type, rpp::utils::extract_observable_type_t<TObservables>...>>{std::move(ptr)};
         }
 
         template<rpp::constraint::observer Observer, size_t... I>
-        static void subscribe(const std::shared_ptr<with_latest_from_disposable<Observer, TSelector, rpp::utils::extract_observable_type_t<TObservables>...>>& disposable, std::index_sequence<I...>, const TObservables&... observables)
+        static void subscribe(const std::shared_ptr<with_latest_from_state<Observer, TSelector, rpp::utils::extract_observable_type_t<TObservables>...>>& state, std::index_sequence<I...>, const TObservables&... observables)
         {
-            (..., observables.subscribe(rpp::observer<rpp::utils::extract_observable_type_t<TObservables>, with_latest_from_inner_observer_strategy<I, Observer, TSelector, rpp::utils::extract_observable_type_t<TObservables>...>>{disposable}));
+            (..., observables.subscribe(rpp::observer<rpp::utils::extract_observable_type_t<TObservables>, with_latest_from_inner_observer_strategy<I, Observer, TSelector, rpp::utils::extract_observable_type_t<TObservables>...>>{state}));
         }
     };
 } // namespace rpp::operators::details

From 4b80c5f0623b4761031f6a1852f85303c59fab2f Mon Sep 17 00:00:00 2001
From: "pre-commit-ci[bot]"
 <66853113+pre-commit-ci[bot]@users.noreply.github.com>
Date: Sun, 15 Sep 2024 19:31:45 +0000
Subject: [PATCH 15/17] [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci
---
 src/rpp/rpp/operators/retry.hpp            | 2 +-
 src/rpp/rpp/operators/retry_when.hpp       | 2 +-
 src/rpp/rpp/operators/with_latest_from.hpp | 2 +-
 src/tests/rpp/test_take_until.cpp          | 4 ++--
 4 files changed, 5 insertions(+), 5 deletions(-)

diff --git a/src/rpp/rpp/operators/retry.hpp b/src/rpp/rpp/operators/retry.hpp
index 89a34dca7..037ac7fb7 100644
--- a/src/rpp/rpp/operators/retry.hpp
+++ b/src/rpp/rpp/operators/retry.hpp
@@ -126,7 +126,7 @@ namespace rpp::operators::details
         template<rpp::constraint::observer TObserver, typename TObservable>
         void subscribe(TObserver&& observer, TObservable&& observble) const
         {
-            const auto ptr   = std::make_shared<retry_state_t<std::decay_t<TObserver>, std::decay_t<TObservable>>>(std::forward<TObserver>(observer), std::forward<TObservable>(observble), count ? count.value() + 1 : count);
+            const auto ptr = std::make_shared<retry_state_t<std::decay_t<TObserver>, std::decay_t<TObservable>>>(std::forward<TObserver>(observer), std::forward<TObservable>(observble), count ? count.value() + 1 : count);
             drain(ptr);
         }
     };
diff --git a/src/rpp/rpp/operators/retry_when.hpp b/src/rpp/rpp/operators/retry_when.hpp
index 62c2ae595..d5ad56a69 100644
--- a/src/rpp/rpp/operators/retry_when.hpp
+++ b/src/rpp/rpp/operators/retry_when.hpp
@@ -160,7 +160,7 @@ namespace rpp::operators::details
         template<rpp::constraint::observer TObserver, typename TObservable>
         void subscribe(TObserver&& observer, TObservable&& observable) const
         {
-            const auto ptr   = std::make_shared<retry_when_state<std::decay_t<TObserver>, std::decay_t<TObservable>, std::decay_t<TNotifier>>>(std::forward<TObserver>(observer), std::forward<TObservable>(observable), notifier);
+            const auto ptr = std::make_shared<retry_when_state<std::decay_t<TObserver>, std::decay_t<TObservable>, std::decay_t<TNotifier>>>(std::forward<TObserver>(observer), std::forward<TObservable>(observable), notifier);
             drain(ptr);
         }
     };
diff --git a/src/rpp/rpp/operators/with_latest_from.hpp b/src/rpp/rpp/operators/with_latest_from.hpp
index 5c553d6c6..1e0eff555 100644
--- a/src/rpp/rpp/operators/with_latest_from.hpp
+++ b/src/rpp/rpp/operators/with_latest_from.hpp
@@ -37,7 +37,7 @@ namespace rpp::operators::details
 
         rpp::utils::tuple<rpp::utils::value_with_mutex<std::optional<RestArgs>>...>& get_values() { return m_values; }
 
-        const TSelector& get_selector() const { return m_selector; }
+        const TSelector&                    get_selector() const { return m_selector; }
         const composite_disposable_wrapper& get_disposable() const { return m_disposable; }
 
     private:
diff --git a/src/tests/rpp/test_take_until.cpp b/src/tests/rpp/test_take_until.cpp
index 46509393d..5ab1b77cf 100644
--- a/src/tests/rpp/test_take_until.cpp
+++ b/src/tests/rpp/test_take_until.cpp
@@ -17,9 +17,9 @@
 #include <rpp/operators/take.hpp>
 #include <rpp/operators/take_until.hpp>
 #include <rpp/schedulers/current_thread.hpp>
+#include <rpp/sources/concat.hpp>
 #include <rpp/sources/empty.hpp>
 #include <rpp/sources/error.hpp>
-#include <rpp/sources/concat.hpp>
 #include <rpp/sources/interval.hpp>
 #include <rpp/sources/just.hpp>
 #include <rpp/sources/never.hpp>
@@ -148,7 +148,7 @@ TEST_CASE("take_until can handle race condition")
 
         SECTION("on_completed shall not interleave with on_next")
         {
-                rpp::source::concat(rpp::source::just(1) | rpp::ops::take(1), rpp::source::never<int>())
+            rpp::source::concat(rpp::source::just(1) | rpp::ops::take(1), rpp::source::never<int>())
                 | rpp::ops::take_until(subject.get_observable())
                 | rpp::ops::as_blocking()
                 | rpp::ops::subscribe([&](auto&&) {

From 15a664e591146e143fa4c4edb2439afec152293e Mon Sep 17 00:00:00 2001
From: Aleksey Loginov <victimsnino@gmail.com>
Date: Mon, 16 Sep 2024 21:54:52 +0300
Subject: [PATCH 16/17] try fix

---
 src/rpp/rpp/disposables/refcount_disposable.hpp | 10 +++++-----
 1 file changed, 5 insertions(+), 5 deletions(-)

diff --git a/src/rpp/rpp/disposables/refcount_disposable.hpp b/src/rpp/rpp/disposables/refcount_disposable.hpp
index 10e327886..90396fc14 100644
--- a/src/rpp/rpp/disposables/refcount_disposable.hpp
+++ b/src/rpp/rpp/disposables/refcount_disposable.hpp
@@ -56,10 +56,10 @@ namespace rpp
 
         enum class Mode : bool
         {
-            Weak,
-            Strong
+            WeakRefStrongSource,
+            StrongRefRefSource
         };
-        composite_disposable_wrapper add_ref(Mode mode = Mode::Weak);
+        composite_disposable_wrapper add_ref(Mode mode = Mode::WeakRefStrongSource);
 
     private:
         std::atomic<size_t>     m_refcount{0};
@@ -107,8 +107,8 @@ namespace rpp
             // just need atomicity, not guarding anything
             if (m_refcount.compare_exchange_strong(current_value, current_value + 1, std::memory_order::seq_cst))
             {
-                auto inner = composite_disposable_wrapper::make<details::refocunt_disposable_inner>(wrapper_from_this());
-                add(mode == Mode::Weak ? inner.as_weak() : inner);
+                auto inner = composite_disposable_wrapper::make<details::refocunt_disposable_inner>(mode == Mode::WeakRefStrongSource ? wrapper_from_this() : wrapper_from_this().as_weak());
+                add(mode == Mode::WeakRefStrongSource ? inner.as_weak() : inner);
                 return inner;
             }
         }

From 499ba4f293864bd526373777f0626eec50b5a501 Mon Sep 17 00:00:00 2001
From: Aleksey Loginov <victimsnino@gmail.com>
Date: Mon, 16 Sep 2024 22:02:26 +0300
Subject: [PATCH 17/17] one more

---
 src/rpp/rpp/operators/concat.hpp | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/src/rpp/rpp/operators/concat.hpp b/src/rpp/rpp/operators/concat.hpp
index 70bda64e9..c4cc12823 100644
--- a/src/rpp/rpp/operators/concat.hpp
+++ b/src/rpp/rpp/operators/concat.hpp
@@ -118,7 +118,7 @@ namespace rpp::operators::details
         }
 
         concat_observer_strategy_base(std::shared_ptr<concat_state_t<TObservable, TObserver>> state)
-            : concat_observer_strategy_base{state, state->get_disposable()->add_ref(refcount_disposable::Mode::Strong)}
+            : concat_observer_strategy_base{state, state->get_disposable()->add_ref(refcount_disposable::Mode::StrongRefRefSource)}
         {
         }
 
@@ -177,7 +177,7 @@ namespace rpp::operators::details
         {
             ConcatStage current = ConcatStage::None;
             if (base::state->stage().compare_exchange_strong(current, ConcatStage::Draining, std::memory_order::seq_cst))
-                base::state->handle_observable(std::forward<T>(v), base::state->get_disposable()->add_ref(refcount_disposable::Mode::Strong));
+                base::state->handle_observable(std::forward<T>(v), base::state->get_disposable()->add_ref(refcount_disposable::Mode::StrongRefRefSource));
             else
                 base::state->get_queue()->push(std::forward<T>(v));
         }