diff --git a/src/rpp/rpp/disposables/refcount_disposable.hpp b/src/rpp/rpp/disposables/refcount_disposable.hpp
index c41521cde..90396fc14 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
+        {
+            WeakRefStrongSource,
+            StrongRefRefSource
+        };
+        composite_disposable_wrapper add_ref(Mode mode = Mode::WeakRefStrongSource);
 
     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)
@@ -102,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(inner.as_weak());
+                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;
             }
         }
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/concat.hpp b/src/rpp/rpp/operators/concat.hpp
index b750b5e15..c4cc12823 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::StrongRefRefSource)}
         {
         }
 
@@ -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::StrongRefRefSource));
             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/operators/debounce.hpp b/src/rpp/rpp/operators/debounce.hpp
index 1803c6db3..24c1a9d34 100644
--- a/src/rpp/rpp/operators/debounce.hpp
+++ b/src/rpp/rpp/operators/debounce.hpp
@@ -12,33 +12,31 @@
 
 #include <rpp/operators/fwd.hpp>
 
-#include <rpp/disposables/composite_disposable.hpp>
 #include <rpp/operators/details/strategy.hpp>
 #include <rpp/utils/utils.hpp>
 
 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 +69,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 +106,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 +153,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;
+            using worker_t = rpp::schedulers::utils::get_worker_t<Scheduler>;
 
-            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
diff --git a/src/rpp/rpp/operators/delay.hpp b/src/rpp/rpp/operators/delay.hpp
index 649844ca0..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>
@@ -35,12 +34,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 +55,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 +102,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 +171,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;
+            using worker_t = rpp::schedulers::utils::get_worker_t<Scheduler>;
 
-            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
diff --git a/src/rpp/rpp/operators/details/combining_strategy.hpp b/src/rpp/rpp/operators/details/combining_strategy.hpp
index f7f5152c5..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>
@@ -22,12 +21,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 +42,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 +89,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 +101,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/merge.hpp b/src/rpp/rpp/operators/merge.hpp
index d7715098c..791a65f84 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}});
         }
     };
 
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
 
diff --git a/src/rpp/rpp/operators/retry.hpp b/src/rpp/rpp/operators/retry.hpp
index 686daec58..037ac7fb7 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);
         }
     };
diff --git a/src/rpp/rpp/operators/retry_when.hpp b/src/rpp/rpp/operators/retry_when.hpp
index bf481db07..d5ad56a69 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,8 @@ 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 +59,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 +79,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 +116,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 +160,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);
         }
     };
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/rpp/rpp/operators/with_latest_from.hpp b/src/rpp/rpp/operators/with_latest_from.hpp
index 78ae5bafe..1e0eff555 100644
--- a/src/rpp/rpp/operators/with_latest_from.hpp
+++ b/src/rpp/rpp/operators/with_latest_from.hpp
@@ -23,53 +23,55 @@
 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; }
 
         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:
         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
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
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));
         }
     };
 
diff --git a/src/tests/rpp/test_take_until.cpp b/src/tests/rpp/test_take_until.cpp
index 7d65a96a8..5ab1b77cf 100644
--- a/src/tests/rpp/test_take_until.cpp
+++ b/src/tests/rpp/test_take_until.cpp
@@ -14,8 +14,10 @@
 #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/concat.hpp>
 #include <rpp/sources/empty.hpp>
 #include <rpp/sources/error.hpp>
 #include <rpp/sources/interval.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&&) {
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) {