Dispose before on_error/on_completed

src/rpp/rpp/operators/concat.hpp

@@ -121,8 +121,8 @@ namespace rpp::operators::details
 
         void on_error(const std::exception_ptr& err) const
         {
-            state->dispose();
             state->get_observer()->on_error(err);
+            state->dispose();
         }
 
         void set_upstream(const disposable_wrapper& d) const { refcounted.add(d); }

src/rpp/rpp/operators/details/combining_strategy.hpp

@@ -62,16 +62,16 @@ namespace rpp::operators::details
 
         void on_error(const std::exception_ptr& err) const
         {
-            disposable->dispose();
             disposable->get_observer_under_lock()->on_error(err);
+            disposable->dispose();
         }
 
         void on_completed() const
         {
             if (disposable->decrement_on_completed())
             {
-                disposable->dispose();
                 disposable->get_observer_under_lock()->on_completed();
+                disposable->dispose();
             }
         }
     };

src/rpp/rpp/operators/merge.hpp

@@ -71,8 +71,8 @@ namespace rpp::operators::details
 
         void on_error(const std::exception_ptr& err) const
         {
-            m_disposable->dispose();
             m_disposable->get_observer_under_lock()->on_error(err);
+            m_disposable->dispose();
         }
 
         void on_completed() const
@@ -83,8 +83,8 @@ namespace rpp::operators::details
                 {
                     m_disposable->remove(v);
                 }
-                m_disposable->dispose();
                 m_disposable->get_observer_under_lock()->on_completed();
+                m_disposable->dispose();
             }
         }
 

src/rpp/rpp/operators/on_error_resume_next.hpp

@@ -38,8 +38,6 @@ namespace rpp::operators::details
 
         void on_error(const std::exception_ptr& err) const
         {
-            disposable.dispose();
-
             std::optional<std::invoke_result_t<Selector, std::exception_ptr>> selector_obs;
             try
             {
@@ -53,12 +51,13 @@ namespace rpp::operators::details
             {
                 std::move(selector_obs).value().subscribe(std::move(observer));
             }
+            disposable.dispose();
         }
 
         void on_completed() const
         {
-            disposable.dispose();
             observer.on_completed();
+            disposable.dispose();
         }
 
         void set_upstream(const disposable_wrapper& d)

src/rpp/rpp/operators/switch_on_next.hpp

@@ -62,8 +62,8 @@ namespace rpp::operators::details
 
         void on_error(const std::exception_ptr& err) const
         {
-            m_state->dispose();
             m_state->get_observer()->on_error(err);
+            m_state->dispose();
         }
 
         void on_completed() const
@@ -106,8 +106,8 @@ namespace rpp::operators::details
 
         void on_error(const std::exception_ptr& err) const
         {
-            m_state->dispose();
             m_state->get_observer()->on_error(err);
+            m_state->dispose();
         }
 
         void on_completed() const

src/rpp/rpp/operators/with_latest_from.hpp

@@ -68,8 +68,8 @@ namespace rpp::operators::details
 
         void on_error(const std::exception_ptr& err) const
         {
-            disposable->dispose();
             disposable->get_observer_under_lock()->on_error(err);
+            disposable->dispose();
         }
 
         static constexpr rpp::utils::empty_function_t<> on_completed{};
@@ -112,14 +112,14 @@ namespace rpp::operators::details
 
         void on_error(const std::exception_ptr& err) const
         {
-            disposable->dispose();
             disposable->get_observer_under_lock()->on_error(err);
+            disposable->dispose();
         }
 
         void on_completed() const
         {
-            disposable->dispose();
             disposable->get_observer_under_lock()->on_completed();
+            disposable->dispose();
         }
     };
 

src/rpp/rpp/sources/concat.hpp

@@ -90,6 +90,7 @@ namespace rpp::details
             if (state->itr.value() == std::cend(state->container))
             {
                 state->observer.on_completed();
+                state->dispose();
                 return;
             }
 

src/rpp/rpp/utils/functors.hpp

@@ -10,6 +10,7 @@
 
 #pragma once
 
+#include <exception>
 #include <tuple>
 
 namespace rpp::utils

src/tests/rpp/test_combine_latest.cpp

@@ -159,8 +159,10 @@ TEST_CASE("combine_latest satisfies disposable contracts")
     auto observable_disposable = rpp::composite_disposable_wrapper::make();
     {
         auto observable = observable_with_disposable<int>(observable_disposable);
+        auto op         = rpp::ops::combine_latest(observable);
 
-        test_operator_with_disposable<int>(rpp::ops::combine_latest(observable));
+        test_operator_with_disposable<int>(op);
+        test_operator_finish_before_dispose<int>(op);
     }
 
     CHECK(observable_disposable.is_disposed() || observable_disposable.lock().use_count() == 2);

src/tests/rpp/test_merge.cpp

@@ -285,8 +285,10 @@ TEST_CASE("merge satisfies disposable contracts")
     auto observable_disposable = rpp::composite_disposable_wrapper::make();
     {
         auto observable = observable_with_disposable<int>(observable_disposable);
+        auto op         = rpp::ops::merge_with(observable);
 
-        test_operator_with_disposable<int>(rpp::ops::merge_with(observable));
+        test_operator_with_disposable<int>(op);
+        test_operator_finish_before_dispose<int>(op);
     }
     CHECK(observable_disposable.is_disposed() || observable_disposable.lock().use_count() == 2);
 }

src/tests/rpp/test_on_error_resume_next.cpp

@@ -168,9 +168,10 @@ TEST_CASE("on_error_resume_next satisfies disposable contracts")
     auto observable_disposable = rpp::composite_disposable_wrapper::make();
     {
         auto observable = observable_with_disposable<int>(observable_disposable);
+        auto op         = rpp::ops::on_error_resume_next([](const std::exception_ptr&) { return rpp::source::empty<int>(); });
 
-        test_operator_with_disposable<int>(
-            rpp::ops::on_error_resume_next([](const std::exception_ptr&) { return rpp::source::empty<int>(); }));
+        test_operator_with_disposable<int>(op);
+        test_operator_finish_before_dispose<int>(op);
     }
 
     CHECK(observable_disposable.is_disposed() || observable_disposable.lock().use_count() == 2);

src/tests/rpp/test_with_lastest_from.cpp → src/tests/rpp/test_with_latest_from.cpp

@@ -198,8 +198,10 @@ TEST_CASE("with_latest_from satisfies disposable contracts")
     auto observable_disposable = rpp::composite_disposable_wrapper::make();
     {
         auto observable = observable_with_disposable<int>(observable_disposable);
+        auto op         = rpp::ops::with_latest_from(observable);
 
-        test_operator_with_disposable<int>(rpp::ops::with_latest_from(observable));
+        test_operator_with_disposable<int>(op);
+        test_operator_finish_before_dispose<int>(op);
     }
 
     CHECK(observable_disposable.is_disposed() || observable_disposable.lock().use_count() == 2);

src/tests/rpp/test_zip.cpp

@@ -180,8 +180,10 @@ TEST_CASE("zip satisfies disposable contracts")
     auto observable_disposable = rpp::composite_disposable_wrapper::make();
     {
         auto observable = observable_with_disposable<int>(observable_disposable);
+        auto op         = rpp::ops::zip(observable);
 
-        test_operator_with_disposable<int>(rpp::ops::zip(observable));
+        test_operator_with_disposable<int>(op);
+        test_operator_finish_before_dispose<int>(op);
     }
 
     CHECK(observable_disposable.is_disposed() || observable_disposable.lock().use_count() == 2);

src/tests/utils/disposable_observable.hpp

@@ -45,6 +45,38 @@ struct wrapped_observable_strategy_no_set_upstream
     auto subscribe(auto&&) const {}
 };
 
+template<typename T>
+void test_operator_over_observable_finish_before_dispose(auto&& op)
+{
+    SECTION("operator calls on_error before dispose")
+    {
+        bool callback_called       = false;
+        auto observable_disposable = rpp::make_callback_disposable([&callback_called]() noexcept {
+            callback_called = true;
+        });
+        auto observable            = rpp::source::create<T>([&observable_disposable](auto&& obs) {
+            obs.set_upstream(observable_disposable);
+            obs.on_error(std::make_exception_ptr(std::runtime_error{""}));
+        });
+
+        op(observable) | rpp::ops::subscribe([](const auto&) {}, [&callback_called](const std::exception_ptr&) { CHECK(!callback_called); });
+    }
+
+    SECTION("operator calls on_completed before dispose")
+    {
+        bool callback_called       = false;
+        auto observable_disposable = rpp::make_callback_disposable([&callback_called]() noexcept {
+            callback_called = true;
+        });
+        auto observable            = rpp::source::create<T>([&observable_disposable](auto&& obs) {
+            obs.set_upstream(observable_disposable);
+            obs.on_completed();
+        });
+
+        op(observable) | rpp::ops::subscribe([](const auto&) {}, [&callback_called]() { CHECK(!callback_called); });
+    }
+}
+
 template<typename T>
 void test_operator_over_observable_with_disposable(auto&& op)
 {
@@ -112,3 +144,9 @@ void test_operator_with_disposable(auto&& op)
 {
     test_operator_over_observable_with_disposable<T>([op](auto&& observable) { return observable | op; });
 }
+
+template<typename T>
+void test_operator_finish_before_dispose(auto&& op)
+{
+    test_operator_over_observable_finish_before_dispose<T>([op](auto&& observable) { return observable | op; });
+}