diff --git a/src/rpp/rpp/disposables/refcount_disposable.hpp b/src/rpp/rpp/disposables/refcount_disposable.hpp
index 90396fc14..c41521cde 100644
--- a/src/rpp/rpp/disposables/refcount_disposable.hpp
+++ b/src/rpp/rpp/disposables/refcount_disposable.hpp
@@ -54,12 +54,7 @@ namespace rpp
         friend class details::refocunt_disposable_inner;
         refcount_disposable() = default;
 
-        enum class Mode : bool
-        {
-            WeakRefStrongSource,
-            StrongRefRefSource
-        };
-        composite_disposable_wrapper add_ref(Mode mode = Mode::WeakRefStrongSource);
+        composite_disposable_wrapper add_ref();
 
     private:
         std::atomic<size_t>     m_refcount{0};
@@ -96,7 +91,7 @@ namespace rpp::details
 
 namespace rpp
 {
-    inline composite_disposable_wrapper refcount_disposable::add_ref(refcount_disposable::Mode mode)
+    inline composite_disposable_wrapper refcount_disposable::add_ref()
     {
         auto current_value = m_refcount.load(std::memory_order::seq_cst);
         while (true)
@@ -107,8 +102,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>(mode == Mode::WeakRefStrongSource ? wrapper_from_this() : wrapper_from_this().as_weak());
-                add(mode == Mode::WeakRefStrongSource ? inner.as_weak() : inner);
+                auto inner = composite_disposable_wrapper::make<details::refocunt_disposable_inner>(wrapper_from_this());
+                add(inner.as_weak());
                 return inner;
             }
         }
diff --git a/src/rpp/rpp/operators/combine_latest.hpp b/src/rpp/rpp/operators/combine_latest.hpp
index a843647bb..2be977ac2 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_state final : public combining_state<Observer>
+    class combine_latest_disposable final : public combining_disposable<Observer>
     {
     public:
-        explicit combine_latest_state(Observer&& observer, const TSelector& selector)
-            : combining_state<Observer>(std::move(observer), sizeof...(Args))
+        explicit combine_latest_disposable(Observer&& observer, const TSelector& selector)
+            : combining_disposable<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_state<Observer, TSelector, Args...>>
+        : public combining_observer_strategy<combine_latest_disposable<Observer, TSelector, Args...>>
     {
-        using combining_observer_strategy<combine_latest_state<Observer, TSelector, Args...>>::state;
+        using combining_observer_strategy<combine_latest_disposable<Observer, TSelector, Args...>>::disposable;
 
         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 = state->get_observer_under_lock();
-            state->get_values().template get<I>().emplace(std::forward<T>(v));
+            const auto observer = disposable->get_observer_under_lock();
+            disposable->get_values().template get<I>().emplace(std::forward<T>(v));
 
-            state->get_values().apply(&apply_impl<decltype(state)>, state, observer);
+            disposable->get_values().apply(&apply_impl<decltype(disposable)>, disposable, observer);
         }
 
     private:
-        template<typename TState>
-        static void apply_impl(const TState& disposable, const rpp::utils::pointer_under_lock<Observer>& observer, const std::optional<Args>&... vals)
+        template<typename TDisposable>
+        static void apply_impl(const TDisposable& 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_state, combine_latest_observer_strategy, TSelector, TObservables...>
+    struct combine_latest_t : public combining_operator_t<combine_latest_disposable, 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 fbe29fced..68534f80d 100644
--- a/src/rpp/rpp/operators/concat.hpp
+++ b/src/rpp/rpp/operators/concat.hpp
@@ -34,33 +34,29 @@ namespace rpp::operators::details
     };
 
     template<rpp::constraint::observable TObservable, rpp::constraint::observer TObserver>
-    class concat_state_t final : public std::enable_shared_from_this<concat_state_t<TObservable, TObserver>>
+    class concat_disposable final : public rpp::refcount_disposable
     {
     public:
-        concat_state_t(TObserver&& observer)
+        concat_disposable(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 (!m_disposable->is_disposed())
+            while (!is_disposed())
             {
                 const auto observable = get_observable();
                 if (!observable)
                 {
                     stage().store(ConcatStage::None, std::memory_order::relaxed);
                     refcounted.dispose();
-                    if (m_disposable->is_disposed())
+                    if (is_disposed())
                         get_observer()->on_completed();
                     return;
                 }
@@ -78,13 +74,12 @@ 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>{this->shared_from_this(), std::move(refcounted)});
+            observable.subscribe(concat_inner_observer_strategy<TObservable, TObserver>{disposable_wrapper_impl<concat_disposable>{wrapper_from_this()}.lock(), std::move(refcounted)});
 
             ConcatStage current = ConcatStage::Draining;
             return stage().compare_exchange_strong(current, ConcatStage::Processing, std::memory_order::seq_cst);
@@ -102,7 +97,6 @@ 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{};
@@ -111,23 +105,23 @@ namespace rpp::operators::details
     template<rpp::constraint::observable TObservable, rpp::constraint::observer TObserver>
     struct concat_observer_strategy_base
     {
-        concat_observer_strategy_base(std::shared_ptr<concat_state_t<TObservable, TObserver>> state, rpp::composite_disposable_wrapper refcounted)
-            : state{std::move(state)}
+        concat_observer_strategy_base(std::shared_ptr<concat_disposable<TObservable, TObserver>> disposable, rpp::composite_disposable_wrapper refcounted)
+            : disposable{std::move(disposable)}
             , refcounted{std::move(refcounted)}
         {
         }
 
-        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::StrongRefRefSource)}
+        concat_observer_strategy_base(std::shared_ptr<concat_disposable<TObservable, TObserver>> disposable)
+            : concat_observer_strategy_base{disposable, disposable->add_ref()}
         {
         }
 
-        std::shared_ptr<concat_state_t<TObservable, TObserver>> state;
-        rpp::composite_disposable_wrapper                       refcounted;
+        std::shared_ptr<concat_disposable<TObservable, TObserver>> disposable;
+        rpp::composite_disposable_wrapper                          refcounted;
 
         void on_error(const std::exception_ptr& err) const
         {
-            state->get_observer()->on_error(err);
+            disposable->get_observer()->on_error(err);
         }
 
         void set_upstream(const disposable_wrapper& d) const { refcounted.add(d); }
@@ -146,18 +140,18 @@ namespace rpp::operators::details
         template<typename T>
         void on_next(T&& v) const
         {
-            base::state->get_observer()->on_next(std::forward<T>(v));
+            base::disposable->get_observer()->on_next(std::forward<T>(v));
         }
 
         void on_completed() const
         {
             ConcatStage current{ConcatStage::Draining};
-            if (base::state->stage().compare_exchange_strong(current, ConcatStage::CompletedWhileDraining, std::memory_order::seq_cst))
+            if (base::disposable->stage().compare_exchange_strong(current, ConcatStage::CompletedWhileDraining, std::memory_order::seq_cst))
                 return;
 
             assert(current == ConcatStage::Processing);
 
-            base::state->drain(base::refcounted);
+            base::disposable->drain(base::refcounted);
         }
     };
 
@@ -168,7 +162,7 @@ namespace rpp::operators::details
         static constexpr auto preferred_disposables_mode = rpp::details::observers::disposables_mode::None;
 
         concat_observer_strategy(TObserver&& observer)
-            : base{std::make_shared<concat_state_t<TObservable, TObserver>>(std::move(observer))}
+            : base{init_state(std::move(observer))}
         {
         }
 
@@ -176,17 +170,27 @@ namespace rpp::operators::details
         void on_next(T&& v) const
         {
             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::StrongRefRefSource));
+            if (base::disposable->stage().compare_exchange_strong(current, ConcatStage::Draining, std::memory_order::seq_cst))
+                base::disposable->handle_observable(std::forward<T>(v), base::disposable->add_ref());
             else
-                base::state->get_queue()->push(std::forward<T>(v));
+                base::disposable->get_queue()->push(std::forward<T>(v));
         }
 
         void on_completed() const
         {
             base::refcounted.dispose();
-            if (base::state->get_disposable()->is_disposed())
-                base::state->get_observer()->on_completed();
+            if (base::disposable->is_disposed())
+                base::disposable->get_observer()->on_completed();
+        }
+
+
+    private:
+        static std::shared_ptr<concat_disposable<TObservable, TObserver>> init_state(TObserver&& observer)
+        {
+            const auto d   = disposable_wrapper_impl<concat_disposable<TObservable, TObserver>>::make(std::move(observer));
+            auto       ptr = d.lock();
+            ptr->get_observer()->set_upstream(d.as_weak());
+            return ptr;
         }
     };
 
diff --git a/src/rpp/rpp/operators/debounce.hpp b/src/rpp/rpp/operators/debounce.hpp
index 3130fe9e8..8e0d3e916 100644
--- a/src/rpp/rpp/operators/debounce.hpp
+++ b/src/rpp/rpp/operators/debounce.hpp
@@ -12,32 +12,33 @@
 
 #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>
-    class debounce_state;
+    template<rpp::constraint::observer Observer, typename Worker, rpp::details::disposables::constraint::disposables_container Container>
+    class debounce_disposable;
 
-    template<rpp::constraint::observer Observer, typename Worker>
-    struct debounce_state_wrapper
+    template<rpp::constraint::observer Observer, typename Worker, rpp::details::disposables::constraint::disposables_container Container>
+    struct debounce_disposable_wrapper
     {
-        std::shared_ptr<debounce_state<Observer, Worker>> state{};
+        std::shared_ptr<debounce_disposable<Observer, Worker, Container>> disposable{};
 
-        bool is_disposed() const { return state->is_disposed(); }
+        bool is_disposed() const { return disposable->is_disposed(); }
 
-        void on_error(const std::exception_ptr& err) const { state->get_observer_under_lock()->on_error(err); }
+        void on_error(const std::exception_ptr& err) const { disposable->get_observer_under_lock()->on_error(err); }
     };
 
-    template<rpp::constraint::observer Observer, typename Worker>
-    class debounce_state final : public rpp::details::enable_wrapper_from_this<debounce_state<Observer, Worker>>
-        , public rpp::details::base_disposable
+    template<rpp::constraint::observer Observer, typename Worker, rpp::details::disposables::constraint::disposables_container Container>
+    class debounce_disposable final : public rpp::composite_disposable_impl<Container>
+        , public rpp::details::enable_wrapper_from_this<debounce_disposable<Observer, Worker, Container>>
     {
         using T = rpp::utils::extract_observer_type_t<Observer>;
 
     public:
-        debounce_state(Observer&& in_observer, Worker&& in_worker, rpp::schedulers::duration period)
+        debounce_disposable(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}
@@ -70,17 +71,17 @@ namespace rpp::operators::details
         {
             m_worker.schedule(
                 m_time_when_value_should_be_emitted.value(),
-                [](const debounce_state_wrapper<Observer, Worker>& handler) -> schedulers::optional_delay_to {
-                    auto value_or_duration = handler.state->extract_value_or_time();
+                [](const debounce_disposable_wrapper<Observer, Worker, Container>& handler) -> schedulers::optional_delay_to {
+                    auto value_or_duration = handler.disposable->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.state->get_observer_under_lock()->on_next(std::move(*value));
+                        handler.disposable->get_observer_under_lock()->on_next(std::move(*value));
 
                     return std::nullopt;
                 },
-                debounce_state_wrapper<Observer, Worker>{this->wrapper_from_this().lock()});
+                debounce_disposable_wrapper<Observer, Worker, Container>{this->wrapper_from_this().lock()});
         }
 
         std::variant<std::monostate, T, schedulers::time_point> extract_value_or_time()
@@ -107,38 +108,38 @@ namespace rpp::operators::details
         std::optional<T>                      m_value_to_be_emitted{};
     };
 
-    template<rpp::constraint::observer Observer, typename Worker>
+    template<rpp::constraint::observer Observer, typename Worker, rpp::details::disposables::constraint::disposables_container Container>
     struct debounce_observer_strategy
     {
         static constexpr auto preferred_disposables_mode = rpp::details::observers::disposables_mode::None;
 
-        std::shared_ptr<debounce_state<Observer, Worker>> state{};
+        std::shared_ptr<debounce_disposable<Observer, Worker, Container>> disposable{};
 
         void set_upstream(const rpp::disposable_wrapper& d) const
         {
-            state->get_observer_under_lock()->set_upstream(d);
+            disposable->add(d);
         }
 
         bool is_disposed() const
         {
-            return state->is_disposed();
+            return disposable->is_disposed();
         }
 
         template<typename T>
         void on_next(T&& v) const
         {
-            state->emplace_safe(std::forward<T>(v));
+            disposable->emplace_safe(std::forward<T>(v));
         }
 
         void on_error(const std::exception_ptr& err) const noexcept
         {
-            state->get_observer_under_lock()->on_error(err);
+            disposable->get_observer_under_lock()->on_error(err);
         }
 
         void on_completed() const noexcept
         {
-            const auto observer = state->get_observer_under_lock();
-            if (const auto value = state->extract_value())
+            const auto observer = disposable->get_observer_under_lock();
+            if (const auto value = disposable->extract_value())
                 observer->on_next(std::move(value).value());
             observer->on_completed();
         }
@@ -154,20 +155,22 @@ namespace rpp::operators::details
         };
 
         template<rpp::details::observables::constraint::disposables_strategy Prev>
-        using updated_optimal_disposables_strategy = typename Prev::template add<1>;
+        using updated_optimal_disposables_strategy = rpp::details::observables::fixed_disposables_strategy<1>;
 
         rpp::schedulers::duration       duration;
         RPP_NO_UNIQUE_ADDRESS Scheduler scheduler;
 
-        template<rpp::constraint::decayed_type Type, rpp::constraint::observer Observer>
-        auto lift(Observer&& observer) const
+        template<rpp::constraint::decayed_type Type, rpp::details::observables::constraint::disposables_strategy DisposableStrategy, rpp::constraint::observer Observer>
+        auto lift_with_disposable_strategy(Observer&& observer) const
         {
-            using worker_t = rpp::schedulers::utils::get_worker_t<Scheduler>;
+            using worker_t  = rpp::schedulers::utils::get_worker_t<Scheduler>;
+            using container = typename DisposableStrategy::disposables_container;
 
-            auto d   = rpp::disposable_wrapper_impl<debounce_state<std::decay_t<Observer>, worker_t>>::make(std::forward<Observer>(observer), scheduler.create_worker(), duration);
-            auto ptr = d.lock();
-            ptr->get_observer_under_lock()->set_upstream(d.as_weak());
-            return rpp::observer<Type, debounce_observer_strategy<std::decay_t<Observer>, worker_t>>{std::move(ptr)};
+
+            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)};
         }
     };
 } // namespace rpp::operators::details
diff --git a/src/rpp/rpp/operators/delay.hpp b/src/rpp/rpp/operators/delay.hpp
index 8b924ee4b..010eb968a 100644
--- a/src/rpp/rpp/operators/delay.hpp
+++ b/src/rpp/rpp/operators/delay.hpp
@@ -13,6 +13,7 @@
 #include <rpp/operators/fwd.hpp>
 
 #include <rpp/defs.hpp>
+#include <rpp/disposables/composite_disposable.hpp>
 #include <rpp/operators/details/strategy.hpp>
 
 #include <mutex>
@@ -34,12 +35,12 @@ namespace rpp::operators::details
         rpp::schedulers::time_point                           time_point{};
     };
 
-    template<rpp::constraint::observer Observer, typename Worker>
-    struct delay_state final
+    template<rpp::constraint::observer Observer, typename Worker, rpp::details::disposables::constraint::disposables_container Container>
+    struct delay_disposable final : public rpp::composite_disposable_impl<Container>
     {
         using T = rpp::utils::extract_observer_type_t<Observer>;
 
-        delay_state(Observer&& in_observer, Worker&& in_worker, rpp::schedulers::duration delay)
+        delay_disposable(Observer&& in_observer, Worker&& in_worker, rpp::schedulers::duration delay)
             : observer(std::move(in_observer))
             , worker{std::move(in_worker)}
             , delay{delay}
@@ -55,31 +56,30 @@ namespace rpp::operators::details
         bool                    is_active{};
     };
 
-    template<rpp::constraint::observer Observer, typename Worker>
-    struct delay_state_wrapper
+    template<rpp::constraint::observer Observer, typename Worker, rpp::details::disposables::constraint::disposables_container Container>
+    struct delay_disposable_wrapper
     {
-        std::shared_ptr<delay_state<Observer, Worker>> state{};
+        std::shared_ptr<delay_disposable<Observer, Worker, Container>> disposable{};
 
-        bool is_disposed() const { return state->observer.is_disposed(); }
+        bool is_disposed() const { return disposable->is_disposed(); }
 
-        void on_error(const std::exception_ptr& err) const { state->observer.on_error(err); }
+        void on_error(const std::exception_ptr& err) const { disposable->observer.on_error(err); }
     };
 
-    template<rpp::constraint::observer Observer, typename Worker, bool ClearOnError>
+    template<rpp::constraint::observer Observer, typename Worker, rpp::details::disposables::constraint::disposables_container Container, bool ClearOnError>
     struct delay_observer_strategy
     {
-        static constexpr auto preferred_disposables_mode = rpp::details::observers::disposables_mode::Auto;
-
-        std::shared_ptr<delay_state<Observer, Worker>> state{};
+        static constexpr auto                                          preferred_disposables_mode = rpp::details::observers::disposables_mode::Auto;
+        std::shared_ptr<delay_disposable<Observer, Worker, Container>> disposable{};
 
         void set_upstream(const rpp::disposable_wrapper& d) const
         {
-            state->observer.set_upstream(d);
+            disposable->add(d);
         }
 
         bool is_disposed() const
         {
-            return state->observer.is_disposed();
+            return disposable->is_disposed();
         }
 
         template<typename T>
@@ -104,59 +104,59 @@ namespace rpp::operators::details
         {
             if (const auto tp = emplace_safe(std::forward<TT>(value)))
             {
-                state->worker.schedule(
+                disposable->worker.schedule(
                     tp.value(),
-                    [](const delay_state_wrapper<Observer, Worker>& wrapper) { return drain_queue(wrapper.state); },
-                    delay_state_wrapper<Observer, Worker>{state});
+                    [](const delay_disposable_wrapper<Observer, Worker, Container>& wrapper) { return drain_queue(wrapper.disposable); },
+                    delay_disposable_wrapper<Observer, Worker, Container>{disposable});
             }
         }
 
         template<typename TT>
         std::optional<rpp::schedulers::time_point> emplace_safe(TT&& item) const
         {
-            std::lock_guard lock{state->mutex};
+            std::lock_guard lock{disposable->mutex};
             if constexpr (ClearOnError && rpp::constraint::decayed_same_as<std::exception_ptr, TT>)
             {
-                state->queue = std::queue<emission<rpp::utils::extract_observer_type_t<Observer>>>{};
-                state->observer.on_error(std::forward<TT>(item));
+                disposable->queue = std::queue<emission<rpp::utils::extract_observer_type_t<Observer>>>{};
+                disposable->observer.on_error(std::forward<TT>(item));
                 return std::nullopt;
             }
             else
             {
-                const auto tp = state->worker.now() + state->delay;
-                state->queue.emplace(std::forward<TT>(item), tp);
-                if (!state->is_active)
+                const auto tp = disposable->worker.now() + disposable->delay;
+                disposable->queue.emplace(std::forward<TT>(item), tp);
+                if (!disposable->is_active)
                 {
-                    state->is_active = true;
+                    disposable->is_active = true;
                     return tp;
                 }
                 return std::nullopt;
             }
         }
 
-        static schedulers::optional_delay_to drain_queue(const std::shared_ptr<delay_state<Observer, Worker>>& state)
+        static schedulers::optional_delay_to drain_queue(const std::shared_ptr<delay_disposable<Observer, Worker, Container>>& disposable)
         {
             while (true)
             {
-                std::unique_lock lock{state->mutex};
-                if (state->queue.empty())
+                std::unique_lock lock{disposable->mutex};
+                if (disposable->queue.empty())
                 {
-                    state->is_active = false;
+                    disposable->is_active = false;
                     return std::nullopt;
                 }
 
-                auto& top = state->queue.front();
-                if (top.time_point > state->worker.now())
+                auto& top = disposable->queue.front();
+                if (top.time_point > disposable->worker.now())
                     return schedulers::optional_delay_to{top.time_point};
 
                 auto item = std::move(top.value);
-                state->queue.pop();
+                disposable->queue.pop();
                 lock.unlock();
 
-                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); },
+                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); },
                                                   [&](rpp::utils::none) {
-                                                      state->observer.on_completed();
+                                                      disposable->observer.on_completed();
                                                   }},
                            std::move(item));
             }
@@ -173,18 +173,21 @@ namespace rpp::operators::details
         };
 
         template<rpp::details::observables::constraint::disposables_strategy Prev>
-        using updated_optimal_disposables_strategy = Prev;
+        using updated_optimal_disposables_strategy = rpp::details::observables::fixed_disposables_strategy<1>;
 
         rpp::schedulers::duration       duration;
         RPP_NO_UNIQUE_ADDRESS Scheduler scheduler;
 
-        template<rpp::constraint::decayed_type Type, rpp::constraint::observer Observer>
-        auto lift(Observer&& observer) const
+        template<rpp::constraint::decayed_type Type, rpp::details::observables::constraint::disposables_strategy DisposableStrategy, rpp::constraint::observer Observer>
+        auto lift_with_disposable_strategy(Observer&& observer) const
         {
-            using worker_t = rpp::schedulers::utils::get_worker_t<Scheduler>;
+            using worker_t  = rpp::schedulers::utils::get_worker_t<Scheduler>;
+            using container = typename DisposableStrategy::disposables_container;
 
-            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)};
+            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)};
         }
     };
 } // 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 89fbe5f60..af3ca98ee 100644
--- a/src/rpp/rpp/operators/details/combining_strategy.hpp
+++ b/src/rpp/rpp/operators/details/combining_strategy.hpp
@@ -13,6 +13,7 @@
 #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,11 +23,10 @@
 namespace rpp::operators::details
 {
     template<rpp::constraint::observer Observer>
-    class combining_state : public rpp::details::enable_wrapper_from_this<combining_state<Observer>>
-        , public rpp::details::base_disposable
+    class combining_disposable : public composite_disposable
     {
     public:
-        explicit combining_state(Observer&& observer, size_t on_completed_needed)
+        explicit combining_disposable(Observer&& observer, size_t on_completed_needed)
             : m_observer_with_mutex{std::move(observer)}
             , m_on_completed_needed{on_completed_needed}
         {
@@ -46,36 +46,36 @@ namespace rpp::operators::details
         std::atomic_size_t m_on_completed_needed;
     };
 
-    template<typename TState>
+    template<typename TDisposable>
     struct combining_observer_strategy
     {
         static constexpr auto preferred_disposables_mode = rpp::details::observers::disposables_mode::Auto;
 
-        std::shared_ptr<TState> state{};
+        std::shared_ptr<TDisposable> disposable{};
 
         void set_upstream(const rpp::disposable_wrapper& d) const
         {
-            state->get_observer_under_lock()->set_upstream(d);
+            disposable->add(d);
         }
 
         bool is_disposed() const
         {
-            return state->is_disposed();
+            return disposable->is_disposed();
         }
 
         void on_error(const std::exception_ptr& err) const
         {
-            state->get_observer_under_lock()->on_error(err);
+            disposable->get_observer_under_lock()->on_error(err);
         }
 
         void on_completed() const
         {
-            if (state->decrement_on_completed())
-                state->get_observer_under_lock()->on_completed();
+            if (disposable->decrement_on_completed())
+                disposable->get_observer_under_lock()->on_completed();
         }
     };
 
-    template<template<typename...> typename TState, template<auto, typename...> typename TStrategy, typename TSelector, rpp::constraint::observable... TObservables>
+    template<template<typename...> typename TDisposable, template<auto, typename...> typename TStrategy, typename TSelector, rpp::constraint::observable... TObservables>
     struct combining_operator_t
     {
         RPP_NO_UNIQUE_ADDRESS rpp::utils::tuple<TObservables...> observables;
@@ -92,7 +92,7 @@ namespace rpp::operators::details
         };
 
         template<rpp::details::observables::constraint::disposables_strategy Prev>
-        using updated_optimal_disposables_strategy = ::rpp::details::observables::default_disposables_strategy; // TODO: sum of Prev + TObservables
+        using updated_optimal_disposables_strategy = ::rpp::details::observables::fixed_disposables_strategy<0>;
 
         template<rpp::constraint::decayed_type Type, rpp::constraint::observer Observer>
         auto lift(Observer&& observer) const
@@ -104,21 +104,21 @@ 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 State = TState<Observer, TSelector, Type, rpp::utils::extract_observable_type_t<TObservables>...>;
+            using Disposable = TDisposable<Observer, TSelector, Type, rpp::utils::extract_observable_type_t<TObservables>...>;
 
-            const auto d     = rpp::disposable_wrapper_impl<State>::make(std::forward<Observer>(observer), selector);
-            auto       state = d.lock();
-            state->get_observer_under_lock()->set_upstream(d.as_weak());
+            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>>(state, std::index_sequence_for<TObservables...>{}, observables...);
+            subscribe<std::decay_t<Type>>(locked, 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(state)};
+            return rpp::observer<Type, TStrategy<0, std::decay_t<Observer>, TSelector, Type, rpp::utils::extract_observable_type_t<TObservables>...>>{std::move(locked)};
         }
 
         template<typename ExpectedValue, rpp::constraint::observer Observer, size_t... I>
-        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)
+        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)
         {
-            (..., 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}));
+            (..., 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}));
         }
     };
 } // namespace rpp::operators::details
diff --git a/src/rpp/rpp/operators/merge.hpp b/src/rpp/rpp/operators/merge.hpp
index ed5a4ae61..ee480a924 100644
--- a/src/rpp/rpp/operators/merge.hpp
+++ b/src/rpp/rpp/operators/merge.hpp
@@ -13,6 +13,7 @@
 #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>
@@ -23,13 +24,12 @@
 namespace rpp::operators::details
 {
     template<rpp::constraint::observer TObserver>
-    class merge_state final
+    class merge_disposable final : public composite_disposable
     {
     public:
-        merge_state(TObserver&& observer)
+        merge_disposable(TObserver&& observer)
             : m_observer(std::move(observer))
         {
-            get_observer_under_lock()->set_upstream(m_disposable);
         }
 
         // just need atomicity, not guarding anything
@@ -40,11 +40,8 @@ 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};
     };
 
@@ -52,50 +49,49 @@ namespace rpp::operators::details
     struct merge_observer_base_strategy
     {
         static constexpr auto preferred_disposables_mode = rpp::details::observers::disposables_mode::Auto;
-
-        merge_observer_base_strategy(std::shared_ptr<merge_state<TObserver>>&& state)
-            : m_state{std::move(state)}
+        merge_observer_base_strategy(std::shared_ptr<merge_disposable<TObserver>>&& disposable)
+            : m_disposable{std::move(disposable)}
         {
         }
 
-        merge_observer_base_strategy(const std::shared_ptr<merge_state<TObserver>>& state)
-            : m_state{state}
+        merge_observer_base_strategy(const std::shared_ptr<merge_disposable<TObserver>>& disposable)
+            : m_disposable{disposable}
         {
         }
 
         void set_upstream(const rpp::disposable_wrapper& d) const
         {
-            m_state->get_disposable().add(d);
+            m_disposable->add(d);
             m_disposables.push_back(d);
         }
 
         bool is_disposed() const
         {
-            return m_state->get_disposable().is_disposed();
+            return m_disposable->is_disposed();
         }
 
         void on_error(const std::exception_ptr& err) const
         {
-            m_state->get_observer_under_lock()->on_error(err);
+            m_disposable->get_observer_under_lock()->on_error(err);
         }
 
         void on_completed() const
         {
-            if (m_state->decrement_on_completed())
+            if (m_disposable->decrement_on_completed())
             {
-                m_state->get_observer_under_lock()->on_completed();
+                m_disposable->get_observer_under_lock()->on_completed();
             }
             else
             {
                 for (const auto& v : m_disposables)
                 {
-                    m_state->get_disposable().remove(v);
+                    m_disposable->remove(v);
                 }
             }
         }
 
     protected:
-        std::shared_ptr<merge_state<TObserver>>      m_state;
+        std::shared_ptr<merge_disposable<TObserver>> m_disposable;
         mutable std::vector<rpp::disposable_wrapper> m_disposables{};
     };
 
@@ -107,7 +103,7 @@ namespace rpp::operators::details
         template<typename T>
         void on_next(T&& v) const
         {
-            merge_observer_base_strategy<TObserver>::m_state->get_observer_under_lock()->on_next(std::forward<T>(v));
+            merge_observer_base_strategy<TObserver>::m_disposable->get_observer_under_lock()->on_next(std::forward<T>(v));
         }
     };
 
@@ -116,15 +112,24 @@ namespace rpp::operators::details
     {
     public:
         explicit merge_observer_strategy(TObserver&& observer)
-            : merge_observer_base_strategy<TObserver>{std::make_shared<merge_state<TObserver>>(std::move(observer))}
+            : merge_observer_base_strategy<TObserver>{init_state(std::move(observer))}
         {
         }
 
         template<typename T>
         void on_next(T&& v) const
         {
-            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}});
+            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;
         }
     };
 
diff --git a/src/rpp/rpp/operators/on_error_resume_next.hpp b/src/rpp/rpp/operators/on_error_resume_next.hpp
index 87ce69773..aa5930969 100644
--- a/src/rpp/rpp/operators/on_error_resume_next.hpp
+++ b/src/rpp/rpp/operators/on_error_resume_next.hpp
@@ -17,6 +17,18 @@
 
 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
     {
@@ -52,43 +64,52 @@ namespace rpp::operators::details
         static constexpr auto preferred_disposables_mode = rpp::details::observers::disposables_mode::None;
 
         on_error_resume_next_observer_strategy(TObserver&& observer, const Selector& selector)
-            : state{std::make_shared<TObserver>(std::move(observer))}
+            : state{init_state(std::move(observer))}
             , selector{selector}
         {
         }
 
-        std::shared_ptr<TObserver>     state;
-        RPP_NO_UNIQUE_ADDRESS Selector selector;
+        std::shared_ptr<on_error_resume_next_disposable<TObserver>> state;
+        RPP_NO_UNIQUE_ADDRESS Selector                              selector;
 
         template<typename T>
         void on_next(T&& v) const
         {
-            state->on_next(std::forward<T>(v));
+            state->observer.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>{state});
+                selector(err).subscribe(on_error_resume_next_inner_observer_strategy<TObserver>{std::shared_ptr<TObserver>(state, &state->observer)});
             }
             catch (...)
             {
-                state->on_error(std::current_exception());
+                state->observer.on_error(std::current_exception());
             }
+            state->dispose();
         }
 
         void on_completed() const
         {
-            state->on_completed();
+            state->observer.on_completed();
         }
 
         void set_upstream(const disposable_wrapper& d) const
         {
-            state->set_upstream(d);
+            state->add(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>
diff --git a/src/rpp/rpp/operators/retry.hpp b/src/rpp/rpp/operators/retry.hpp
index 73797acdc..cb6700cde 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
+    struct retry_state_t final : public rpp::composite_disposable
     {
         retry_state_t(TObserver&& in_observer, const Observable& observable, std::optional<size_t> count)
             : count{count}
@@ -26,7 +26,6 @@ namespace rpp::operators::details
             , observable(observable)
 
         {
-            observer.set_upstream(disposable);
         }
 
         std::optional<size_t> count;
@@ -34,8 +33,6 @@ 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>
@@ -64,7 +61,7 @@ namespace rpp::operators::details
                 return;
             }
 
-            state->disposable.clear();
+            state->clear();
 
             if (state->is_inside_drain.exchange(false, std::memory_order::seq_cst))
                 return;
@@ -80,16 +77,16 @@ namespace rpp::operators::details
 
         void set_upstream(const disposable_wrapper& d) const
         {
-            state->disposable.add(d);
+            state->add(d);
         }
 
-        bool is_disposed() const { return locally_disposed || state->disposable.is_disposed(); }
+        bool is_disposed() const { return locally_disposed || state->is_disposed(); }
     };
 
     template<rpp::constraint::observer TObserver, typename TObservable>
     void drain(const std::shared_ptr<retry_state_t<TObserver, TObservable>>& state)
     {
-        while (!state->disposable.is_disposed())
+        while (!state->is_disposed())
         {
             if (state->count)
                 --state->count.value();
@@ -126,7 +123,10 @@ 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 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());
             drain(ptr);
         }
     };
diff --git a/src/rpp/rpp/operators/retry_when.hpp b/src/rpp/rpp/operators/retry_when.hpp
index 79155d8aa..b1bea7e8e 100644
--- a/src/rpp/rpp/operators/retry_when.hpp
+++ b/src/rpp/rpp/operators/retry_when.hpp
@@ -23,14 +23,13 @@ namespace rpp::operators::details
     template<rpp::constraint::observer TObserver,
              typename TObservable,
              typename TNotifier>
-    struct retry_when_state final
+    struct retry_when_state final : public rpp::composite_disposable
     {
-        retry_when_state(TObserver&& in_observer, const TObservable& observable, const TNotifier& notifier)
-            : observer(std::move(in_observer))
+        retry_when_state(TObserver&& observer, const TObservable& observable, const TNotifier& notifier)
+            : observer(std::move(observer))
             , observable(observable)
             , notifier(notifier)
         {
-            observer.set_upstream(disposable);
         }
 
         std::atomic_bool is_inside_drain{};
@@ -38,8 +37,6 @@ 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>
@@ -59,7 +56,7 @@ namespace rpp::operators::details
         void on_next(T&&) const
         {
             locally_disposed = true;
-            state->disposable.clear();
+            state->clear();
 
             if (state->is_inside_drain.exchange(false, std::memory_order::seq_cst))
                 return;
@@ -79,9 +76,9 @@ namespace rpp::operators::details
             state->observer.on_completed();
         }
 
-        void set_upstream(const disposable_wrapper& d) const { state->disposable.add(d); }
+        void set_upstream(const disposable_wrapper& d) const { state->add(d); }
 
-        bool is_disposed() const { return locally_disposed || state->disposable.is_disposed(); }
+        bool is_disposed() const { return locally_disposed || state->is_disposed(); }
     };
 
     template<rpp::constraint::observer TObserver,
@@ -116,15 +113,15 @@ namespace rpp::operators::details
             state->observer.on_completed();
         }
 
-        void set_upstream(const disposable_wrapper& d) { state->disposable.add(d); }
+        void set_upstream(const disposable_wrapper& d) { state->add(d); }
 
-        bool is_disposed() const { return state->disposable.is_disposed(); }
+        bool is_disposed() const { return state->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->disposable.is_disposed())
+        while (!state->is_disposed())
         {
             state->is_inside_drain.store(true, std::memory_order::seq_cst);
             try
@@ -160,7 +157,10 @@ 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 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());
             drain(ptr);
         }
     };
diff --git a/src/rpp/rpp/operators/take_until.hpp b/src/rpp/rpp/operators/take_until.hpp
index 6a9933953..a9b6d03bb 100644
--- a/src/rpp/rpp/operators/take_until.hpp
+++ b/src/rpp/rpp/operators/take_until.hpp
@@ -13,21 +13,22 @@
 #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_state final
+    class take_until_disposable final : public rpp::composite_disposable
     {
     public:
-        take_until_state(TObserver&& observer)
+        take_until_disposable(TObserver&& observer)
             : m_observer_with_mutex(std::move(observer))
         {
         }
 
-        take_until_state(const TObserver& observer)
+        take_until_disposable(const TObserver& observer)
             : m_observer_with_mutex(observer)
         {
         }
@@ -48,7 +49,7 @@ namespace rpp::operators::details
     {
         static constexpr auto preferred_disposables_mode = rpp::details::observers::disposables_mode::None;
 
-        std::shared_ptr<take_until_state<TObserver>> state;
+        std::shared_ptr<take_until_disposable<TObserver>> state;
 
         void on_error(const std::exception_ptr& err) const
         {
@@ -62,9 +63,9 @@ namespace rpp::operators::details
                 state->get_observer()->on_completed();
         }
 
-        void set_upstream(const disposable_wrapper& d) { state->get_observer()->set_upstream(d); }
+        void set_upstream(const disposable_wrapper& d) { state->add(d); }
 
-        bool is_disposed() const { return state->get_observer()->is_disposed(); }
+        bool is_disposed() const { return state->is_disposed(); }
     };
 
     template<rpp::constraint::observer TObserver>
@@ -103,12 +104,14 @@ namespace rpp::operators::details
         };
 
         template<rpp::details::observables::constraint::disposables_strategy Prev>
-        using updated_optimal_disposables_strategy = rpp::details::observables::default_disposables_strategy;
+        using updated_optimal_disposables_strategy = rpp::details::observables::fixed_disposables_strategy<1>;
 
         template<rpp::constraint::decayed_type Type, rpp::constraint::observer Observer>
         auto lift(Observer&& observer) const
         {
-            auto ptr = std::make_shared<take_until_state<std::decay_t<Observer>>>(std::forward<Observer>(observer));
+            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());
 
             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 f952d43f8..f55bfa3c9 100644
--- a/src/rpp/rpp/operators/with_latest_from.hpp
+++ b/src/rpp/rpp/operators/with_latest_from.hpp
@@ -23,57 +23,53 @@
 namespace rpp::operators::details
 {
     template<rpp::constraint::observer Observer, typename TSelector, rpp::constraint::decayed_type... RestArgs>
-    class with_latest_from_state final
+    class with_latest_from_disposable final : public composite_disposable
     {
     public:
-        explicit with_latest_from_state(Observer&& observer, const TSelector& selector)
+        explicit with_latest_from_disposable(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 composite_disposable_wrapper& get_disposable() const { return m_disposable; }
+        const TSelector& get_selector() const { return m_selector; }
 
     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
     {
-        static constexpr auto preferred_disposables_mode = rpp::details::observers::disposables_mode::Auto;
-
-        std::shared_ptr<with_latest_from_state<Observer, TSelector, RestArgs...>> state{};
+        static constexpr auto                                                          preferred_disposables_mode = rpp::details::observers::disposables_mode::Auto;
+        std::shared_ptr<with_latest_from_disposable<Observer, TSelector, RestArgs...>> disposable{};
 
         void set_upstream(const rpp::disposable_wrapper& d) const
         {
-            state->get_disposable().add(d);
+            disposable->add(d);
         }
 
         bool is_disposed() const
         {
-            return state->get_disposable().is_disposed();
+            return disposable->is_disposed();
         }
 
         template<typename T>
         void on_next(T&& v) const
         {
-            auto locked_value = state->get_values().template get<I>().lock();
+            auto locked_value = disposable->get_values().template get<I>().lock();
             locked_value->emplace(std::forward<T>(v));
         }
 
         void on_error(const std::exception_ptr& err) const
         {
-            state->get_observer_under_lock()->on_error(err);
+            disposable->get_observer_under_lock()->on_error(err);
         }
 
         static constexpr rpp::utils::empty_function_t<> on_completed{};
@@ -83,26 +79,26 @@ namespace rpp::operators::details
         requires std::invocable<TSelector, OriginalValue, RestArgs...>
     struct with_latest_from_observer_strategy
     {
-        using Disposable                                 = with_latest_from_state<Observer, TSelector, RestArgs...>;
+        using Disposable                                 = with_latest_from_disposable<Observer, TSelector, RestArgs...>;
         using Result                                     = std::invoke_result_t<TSelector, OriginalValue, RestArgs...>;
         static constexpr auto preferred_disposables_mode = rpp::details::observers::disposables_mode::None;
 
-        std::shared_ptr<Disposable> state{};
+        std::shared_ptr<Disposable> disposable{};
 
         void set_upstream(const rpp::disposable_wrapper& d) const
         {
-            state->get_disposable().add(d);
+            disposable->add(d);
         }
 
         bool is_disposed() const
         {
-            return state->get_disposable().is_disposed();
+            return disposable->is_disposed();
         }
 
         template<typename T>
         void on_next(T&& v) const
         {
-            auto result = state->get_values().apply([&d = this->state, &v](rpp::utils::value_with_mutex<std::optional<RestArgs>>&... vals) -> std::optional<Result> {
+            auto result = disposable->get_values().apply([&d = this->disposable, &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() && ...))
@@ -111,17 +107,17 @@ namespace rpp::operators::details
             });
 
             if (result.has_value())
-                state->get_observer_under_lock()->on_next(std::move(result).value());
+                disposable->get_observer_under_lock()->on_next(std::move(result).value());
         }
 
         void on_error(const std::exception_ptr& err) const
         {
-            state->get_observer_under_lock()->on_error(err);
+            disposable->get_observer_under_lock()->on_error(err);
         }
 
         void on_completed() const
         {
-            state->get_observer_under_lock()->on_completed();
+            disposable->get_observer_under_lock()->on_completed();
         }
     };
 
@@ -142,7 +138,7 @@ namespace rpp::operators::details
         };
 
         template<rpp::details::observables::constraint::disposables_strategy Prev>
-        using updated_optimal_disposables_strategy = rpp::details::observables::default_disposables_strategy;
+        using updated_optimal_disposables_strategy = rpp::details::observables::fixed_disposables_strategy<1>;
 
         template<rpp::constraint::decayed_type Type, rpp::constraint::observer Observer>
         auto lift(Observer&& observer) const
@@ -154,18 +150,20 @@ 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 State = with_latest_from_state<Observer, TSelector, rpp::utils::extract_observable_type_t<TObservables>...>;
+            using Disposable = with_latest_from_disposable<Observer, TSelector, rpp::utils::extract_observable_type_t<TObservables>...>;
 
-            auto ptr = std::make_shared<State>(std::forward<Observer>(observer), selector);
+            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());
             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_state<Observer, TSelector, rpp::utils::extract_observable_type_t<TObservables>...>>& state, std::index_sequence<I...>, const TObservables&... observables)
+        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)
         {
-            (..., 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}));
+            (..., 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}));
         }
     };
 } // namespace rpp::operators::details
diff --git a/src/rpp/rpp/operators/zip.hpp b/src/rpp/rpp/operators/zip.hpp
index ec29fd04e..71d0369d0 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_state final : public combining_state<Observer>
+    class zip_disposable final : public combining_disposable<Observer>
     {
     public:
-        explicit zip_state(Observer&& observer, const TSelector& selector)
-            : combining_state<Observer>(std::move(observer), sizeof...(Args))
+        explicit zip_disposable(Observer&& observer, const TSelector& selector)
+            : combining_disposable<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_state<Observer, TSelector, Args...>>
+        : public combining_observer_strategy<zip_disposable<Observer, TSelector, Args...>>
     {
-        using combining_observer_strategy<zip_state<Observer, TSelector, Args...>>::state;
+        using combining_observer_strategy<zip_disposable<Observer, TSelector, Args...>>::disposable;
 
         template<typename T>
         void on_next(T&& v) const
         {
-            const auto observer = state->get_observer_under_lock();
-            state->get_pendings().template get<I>().push_back(std::forward<T>(v));
+            const auto observer = disposable->get_observer_under_lock();
+            disposable->get_pendings().template get<I>().push_back(std::forward<T>(v));
 
-            state->get_pendings().apply(&apply_impl<decltype(state)>, state, observer);
+            disposable->get_pendings().apply(&apply_impl<decltype(disposable)>, disposable, 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_state, zip_observer_strategy, TSelector, TObservables...>
+    struct zip_t : public combining_operator_t<zip_disposable, 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 d8e74722d..b22f86c85 100644
--- a/src/rpp/rpp/sources/concat.hpp
+++ b/src/rpp/rpp/sources/concat.hpp
@@ -22,13 +22,12 @@
 namespace rpp::details
 {
     template<rpp::constraint::observer TObserver, constraint::decayed_type PackedContainer>
-    struct concat_state_t
+    struct concat_state_t : public rpp::composite_disposable
     {
         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);
@@ -41,7 +40,6 @@ namespace rpp::details
 
         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{};
     };
@@ -69,14 +67,14 @@ namespace rpp::details
             state->observer.on_error(err);
         }
 
-        void set_upstream(const disposable_wrapper& d) { state->disposable.add(d); }
+        void set_upstream(const disposable_wrapper& d) { state->add(d); }
 
-        bool is_disposed() const { return locally_disposed || state->disposable.is_disposed(); }
+        bool is_disposed() const { return locally_disposed || state->is_disposed(); }
 
         void on_completed() const
         {
             locally_disposed = true;
-            state->disposable.clear();
+            state->clear();
 
             if (state->is_inside_drain.exchange(false, std::memory_order::seq_cst))
                 return;
@@ -88,7 +86,7 @@ namespace rpp::details
     template<rpp::constraint::observer TObserver, typename PackedContainer>
     void drain(const std::shared_ptr<concat_state_t<TObserver, PackedContainer>>& state)
     {
-        while (!state->disposable.is_disposed())
+        while (!state->is_disposed())
         {
             if (state->itr.value() == std::cend(state->container))
             {
@@ -133,7 +131,10 @@ namespace rpp::details
         template<constraint::observer_strategy<value_type> Strategy>
         void subscribe(observer<value_type, Strategy>&& obs) const
         {
-            drain(std::make_shared<concat_state_t<observer<value_type, Strategy>, PackedContainer>>(std::move(obs), container));
+            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);
         }
     };