Disposable tests (#409)

victimsnino · Aug 20, 2023 · 6eecb02 · 6eecb02 · github-actions · Aug 20, 2023
1 parent 0a0e7e8
commit 6eecb02
Show file tree

Hide file tree

Showing 22 changed files with 175 additions and 9 deletions.
diff --git a/src/rpp/rpp/disposables/refcount_disposable.hpp b/src/rpp/rpp/disposables/refcount_disposable.hpp
@@ -26,7 +26,7 @@ namespace rpp
  * 
  * @ingroup disposables
  */
-class refcount_disposable final : public interface_composite_disposable, public std::enable_shared_from_this<refcount_disposable>
+class refcount_disposable : public interface_composite_disposable, public std::enable_shared_from_this<refcount_disposable>
 {
 public:
     refcount_disposable() = default;
@@ -35,9 +35,9 @@ class refcount_disposable final : public interface_composite_disposable, public
     refcount_disposable(const refcount_disposable&)     = delete;
     refcount_disposable(refcount_disposable&&) noexcept = delete;
 
-    bool is_disposed() const noexcept override { return m_refcount.load(std::memory_order_acquire) == 0; }
+    bool is_disposed() const noexcept final { return m_refcount.load(std::memory_order_acquire) == 0; }
 
-    void dispose() override
+    void dispose() final
     {
         while (auto current_value = m_refcount.load(std::memory_order_acquire))
         {
@@ -66,7 +66,7 @@ class refcount_disposable final : public interface_composite_disposable, public
 
     using interface_composite_disposable::add;
 
-    void add(disposable_wrapper disposable) override
+    void add(disposable_wrapper disposable) final
     {
         m_underlying.add(std::move(disposable));
     }

diff --git a/src/rpp/rpp/operators/details/utils.hpp b/src/rpp/rpp/operators/details/utils.hpp
@@ -43,5 +43,12 @@ class pointer_under_lock
     std::scoped_lock<std::mutex> m_lock;
 };
 
-
+template<typename T>
+struct with_auto_dispose : public T
+{
+    ~with_auto_dispose() override
+    {
+        T::dispose();
+    }
+};
 } // namespace rpp::operators::details
diff --git a/src/rpp/rpp/operators/group_by.hpp b/src/rpp/rpp/operators/group_by.hpp
@@ -12,6 +12,7 @@
 
 #include <rpp/operators/fwd.hpp>
 #include <rpp/operators/details/strategy.hpp>
+#include <rpp/operators/details/utils.hpp>
 #include <rpp/observables/grouped_observable.hpp>
 #include <rpp/subjects/publish_subject.hpp>
 #include <rpp/disposables/refcount_disposable.hpp>
@@ -67,7 +68,7 @@ struct group_by_observer_strategy
     RPP_NO_UNIQUE_ADDRESS KeyComparator comparator;
 
     mutable std::map<TKey, subjects::publish_subject<Type>, KeyComparator> key_to_subject{};
-    std::shared_ptr<refcount_disposable> disposable = std::make_shared<refcount_disposable>();
+    std::shared_ptr<details::with_auto_dispose<refcount_disposable>> disposable = std::make_shared<details::with_auto_dispose<refcount_disposable>>();
 
     void on_subscribe(rpp::constraint::observer auto& obs) const
     {

diff --git a/src/rpp/rpp/operators/merge.hpp b/src/rpp/rpp/operators/merge.hpp
@@ -26,7 +26,7 @@
 namespace rpp::operators::details
 {
 template<rpp::constraint::observer TObserver>
-class merge_disposable final : public composite_disposable
+class merge_disposable final : public with_auto_dispose<composite_disposable>
 {
 public:
     merge_disposable(TObserver&& observer) : m_observer(std::move(observer)) {}

diff --git a/src/rpp/rpp/operators/with_latest_from.hpp b/src/rpp/rpp/operators/with_latest_from.hpp
@@ -25,7 +25,7 @@
 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_disposable final : public with_auto_dispose<composite_disposable>
 {
 public:
     explicit with_latest_from_disposable(Observer&& observer, const TSelector& selector)

diff --git a/src/tests/rpp/test_concat.cpp b/src/tests/rpp/test_concat.cpp
@@ -17,6 +17,8 @@
 
 #include "mock_observer.hpp"
 #include "copy_count_tracker.hpp"
+#include "disposable_observable.hpp"
+
 
 #include <rpp/disposables/composite_disposable.hpp>
 #include <rpp/disposables/disposable_wrapper.hpp>
@@ -237,4 +239,9 @@ TEST_CASE("concat of iterable doesn't produce extra copies")
         CHECK(tracker.get_copy_count() - initial_copy == 2); // 1 copy to observable + 1 copy to observer
         CHECK(tracker.get_move_count() - initial_move == 0);
     }
+}
+
+TEST_CASE("concat disposes original disposable on disposing")
+{
+    test_operator_over_observable_with_disposable<int>([](auto&& observable){return rpp::source::concat(observable);});
 }
diff --git a/src/tests/rpp/test_distinct_until_changed.cpp b/src/tests/rpp/test_distinct_until_changed.cpp
@@ -15,6 +15,8 @@
 
 #include "mock_observer.hpp"
 #include "copy_count_tracker.hpp"
+#include "disposable_observable.hpp"
+
 
 TEMPLATE_TEST_CASE("distinct_until_changed filters out consecutive duplicates and send first value from duplicates", "", rpp::memory_model::use_stack, rpp::memory_model::use_shared)
 {
@@ -46,4 +48,9 @@ TEST_CASE("distinct_until_changed doesn't produce extra copies")
                                           .send_by_move = {.copy_count = 1, // 1 copy to internal state
                                                            .move_count = 1} // 1 move to final subscriber
                                       });
+}
+
+TEST_CASE("distinct_until_changed disposes original disposable on disposing")
+{
+    test_operator_with_disposable<int>(rpp::ops::distinct_until_changed());
 }
diff --git a/src/tests/rpp/test_filter.cpp b/src/tests/rpp/test_filter.cpp
@@ -15,6 +15,7 @@
 
 #include "mock_observer.hpp"
 #include "copy_count_tracker.hpp"
+#include "disposable_observable.hpp"
 
 #include <stdexcept>
 #include <string>
@@ -69,4 +70,9 @@ TEST_CASE("filter doesn't produce extra copies")
                                                             .move_count = 0}
                                         });
     }
+}
+
+TEST_CASE("filter disposes original disposable on disposing")
+{
+    test_operator_with_disposable<int>(rpp::ops::filter([](const int&){return false;}));
 }
diff --git a/src/tests/rpp/test_first.cpp b/src/tests/rpp/test_first.cpp
@@ -18,6 +18,8 @@
 
 #include "mock_observer.hpp"
 #include "copy_count_tracker.hpp"
+#include "disposable_observable.hpp"
+
 
 TEST_CASE("first only emits once")
 {
@@ -88,4 +90,9 @@ TEST_CASE("first doesn't produce extra copies")
                                                             .move_count = 1} // 1 move to final subscriber
                                         }, 2);
     }
+}
+
+TEST_CASE("first disposes original disposable on disposing")
+{
+    test_operator_with_disposable<int>(rpp::ops::first());
 }
diff --git a/src/tests/rpp/test_flat_map.cpp b/src/tests/rpp/test_flat_map.cpp
@@ -21,6 +21,8 @@
 
 #include "copy_count_tracker.hpp"
 #include "mock_observer.hpp"
+#include "disposable_observable.hpp"
+
 
 #include <stdexcept>
 #include <string>
@@ -134,4 +136,9 @@ TEMPLATE_TEST_CASE("flat_map", "", rpp::memory_model::use_stack, rpp::memory_mod
             }
         }
     }
+}
+
+TEST_CASE("flat_map disposes original disposable on disposing")
+{
+    test_operator_with_disposable<int>(rpp::ops::flat_map([](const auto& v){return rpp::source::just(v); }));
 }
diff --git a/src/tests/rpp/test_group_by.cpp b/src/tests/rpp/test_group_by.cpp
@@ -261,4 +261,4 @@ TEST_CASE("group_by selectors affects types", "[group_by]")
             REQUIRE(mock.get_on_completed_count() == 0);
         }
     }
-}
+}
diff --git a/src/tests/rpp/test_last.cpp b/src/tests/rpp/test_last.cpp
@@ -18,6 +18,8 @@
 
 #include "mock_observer.hpp"
 #include "copy_count_tracker.hpp"
+#include "disposable_observable.hpp"
+
 
 
 TEST_CASE("last only emits once")
@@ -92,4 +94,9 @@ TEST_CASE("last doesn't produce extra copies")
                                                             .move_count = 3} // 2 move to std::optional + 1 move to final subscriber
                                         }, 2);
     }
+}
+
+TEST_CASE("last disposes original disposable on disposing")
+{
+    test_operator_with_disposable<int>(rpp::ops::last());
 }
diff --git a/src/tests/rpp/test_map.cpp b/src/tests/rpp/test_map.cpp
@@ -15,6 +15,8 @@
 
 #include "mock_observer.hpp"
 #include "copy_count_tracker.hpp"
+#include "disposable_observable.hpp"
+
 
 #include <stdexcept>
 #include <string>
@@ -63,4 +65,9 @@ TEST_CASE("map doesn't produce extra copies")
                                                             .move_count = 2} // 1 move on return from map + 1 move to final subscriber
                                         });
     }
+}
+
+TEST_CASE("map disposes original disposable on disposing")
+{
+    test_operator_with_disposable<int>(rpp::ops::map([](auto&& v){return std::forward<decltype(v)>(v);}));
 }
diff --git a/src/tests/rpp/test_merge.cpp b/src/tests/rpp/test_merge.cpp
@@ -22,6 +22,8 @@
 
 #include "mock_observer.hpp"
 #include "copy_count_tracker.hpp"
+#include "disposable_observable.hpp"
+
 
 #include <stdexcept>
 #include <string>
@@ -285,4 +287,14 @@ TEST_CASE("merge doesn't produce extra copies")
         REQUIRE(verifier.get_copy_count() == 0);
         REQUIRE(verifier.get_move_count() == 1); // 1 move to final subscriber
     }
+}
+
+TEST_CASE("merge disposes original disposable on disposing")
+{
+    auto observable_disposable = std::make_shared<rpp::composite_disposable>();
+    auto observable = observable_with_disposable<int>(observable_disposable);
+
+    test_operator_with_disposable<int>(rpp::ops::merge_with(observable));
+
+    CHECK(observable_disposable->is_disposed());
 }
diff --git a/src/tests/rpp/test_repeat.cpp b/src/tests/rpp/test_repeat.cpp
@@ -16,6 +16,7 @@
 
 #include "mock_observer.hpp"
 #include "copy_count_tracker.hpp"
+#include "disposable_observable.hpp"
 
 
 TEST_CASE("repeat resubscribes")
@@ -130,4 +131,9 @@ TEST_CASE("repeat doesn't produce extra copies")
                                                             .move_count = 2} // 2 times 1 move to final subscriber
                                         });
     }
+}
+
+TEST_CASE("repeat disposes original disposable on disposing")
+{
+    test_operator_with_disposable<int>(rpp::ops::repeat());
 }
diff --git a/src/tests/rpp/test_scan.cpp b/src/tests/rpp/test_scan.cpp
@@ -15,6 +15,8 @@
 
 #include "mock_observer.hpp"
 #include "copy_count_tracker.hpp"
+#include "disposable_observable.hpp"
+
 #include "rpp/operators/subscribe.hpp"
 
 
@@ -166,4 +168,9 @@ TEST_CASE("scan doesn't produce extra copies")
             CHECK(tracker.get_move_count() == 3); // 2 times 1 move to internal state
         }
     }
+}
+
+TEST_CASE("scan disposes original disposable on disposing")
+{
+    test_operator_with_disposable<int>(rpp::ops::scan([](auto&& s, auto&&){return s; }));
 }
diff --git a/src/tests/rpp/test_skip.cpp b/src/tests/rpp/test_skip.cpp
@@ -17,6 +17,8 @@
 
 #include "mock_observer.hpp"
 #include "copy_count_tracker.hpp"
+#include "disposable_observable.hpp"
+
 
 TEMPLATE_TEST_CASE("skip ignores first `count` of items",
                    "",
@@ -79,4 +81,9 @@ TEST_CASE("skip doesn't produce extra copies")
                                                             .move_count = 1} // 1 move to final subscriber
                                         }, 2);
     }
+}
+
+TEST_CASE("skip disposes original disposable on disposing")
+{
+    test_operator_with_disposable<int>(rpp::ops::skip(1));
 }
diff --git a/src/tests/rpp/test_subscribe_on.cpp b/src/tests/rpp/test_subscribe_on.cpp
@@ -22,6 +22,8 @@
 #include <thread>
 
 #include "mock_observer.hpp"
+#include "disposable_observable.hpp"
+
 #include "rpp/disposables/composite_disposable.hpp"
 #include "rpp/disposables/fwd.hpp"
 #include "rpp/operators/fwd.hpp"
@@ -110,3 +112,9 @@ TEST_CASE("subscribe_on schedules job in another scheduler")
         CHECK(second->is_disposed());
     }
 }
+
+
+TEST_CASE("group_by disposes original disposable on disposing")
+{
+    test_operator_with_disposable<int>(rpp::ops::subscribe_on(rpp::schedulers::current_thread{}));
+}
diff --git a/src/tests/rpp/test_take.cpp b/src/tests/rpp/test_take.cpp
@@ -15,6 +15,8 @@
 
 #include "mock_observer.hpp"
 #include "copy_count_tracker.hpp"
+#include "disposable_observable.hpp"
+
 
 TEST_CASE("take operator limits emissions")
 {
@@ -106,4 +108,9 @@ TEST_CASE("take doesn't produce extra copies")
                                                             .move_count = 1} // 1 move to final subscriber
                                         }, 2);
     }
+}
+
+TEST_CASE("take disposes original disposable on disposing")
+{
+    test_operator_with_disposable<int>(rpp::ops::take(1));
 }
diff --git a/src/tests/rpp/test_take_while.cpp b/src/tests/rpp/test_take_while.cpp
@@ -10,6 +10,8 @@
 
 #include "mock_observer.hpp"
 #include "copy_count_tracker.hpp"
+#include "disposable_observable.hpp"
+
 
 #include <rpp/operators/take_while.hpp>
 #include <rpp/sources/create.hpp>
@@ -57,4 +59,9 @@ TEST_CASE("take_while doesn't produce extra copies")
                                                             .move_count = 1} // 1 move to final subscriber
                                         });
     }
+}
+
+TEST_CASE("take_while disposes original disposable on disposing")
+{
+    test_operator_with_disposable<int>(rpp::ops::take_while([](auto){return true; }));
 }
diff --git a/src/tests/rpp/test_with_lastest_from.cpp b/src/tests/rpp/test_with_lastest_from.cpp
@@ -16,6 +16,7 @@
 #include <rpp/operators/with_latest_from.hpp>
 
 #include "mock_observer.hpp"
+#include "disposable_observable.hpp"
 
 
 TEST_CASE("with_latest_from combines observables")
@@ -179,4 +180,15 @@ TEST_CASE("with_latest_from handles race condition", "[with_latest_from]")
             }
         }
     }
+}
+
+TEST_CASE("with_latest_from disposes original disposable on disposing")
+{
+    auto observable_disposable = std::make_shared<rpp::composite_disposable>();
+    auto observable = observable_with_disposable<int>(observable_disposable);
+
+    test_operator_with_disposable<int>(rpp::ops::with_latest_from(observable));
+
+
+    CHECK(observable_disposable->is_disposed());
 }
name	rxcpp	rpp	prev rpp	ratio
Subscribe empty callbacks to empty observable	731.41 ns	1.60 ns	1.67 ns	0.96
Subscribe empty callbacks to empty observable via pipe operator	733.80 ns	1.20 ns	2.01 ns	0.60
name	rxcpp	rpp	prev rpp	ratio
from array of 1 - create + subscribe + immediate	1900.69 ns	0.40 ns	0.33 ns	1.20
from array of 1 - create + subscribe + current_thread	2768.28 ns	39.21 ns	43.90 ns	0.89
concat_as_source of just(1 immediate) create + subscribe	5701.04 ns	0.40 ns	0.33 ns	1.20
name	rxcpp	rpp	prev rpp	ratio
create+take(1)+subscribe	1461.42 ns	0.40 ns	0.33 ns	1.20
create+filter(true)+subscribe	922.93 ns	0.40 ns	0.33 ns	1.20
create(1,2)+skip(1)+subscribe	1257.32 ns	0.40 ns	0.33 ns	1.20
create(1,1,2)+distinct_until_changed()+subscribe	875.93 ns	0.40 ns	0.33 ns	1.20
create(1,2)+first()+subscribe	1805.04 ns	0.40 ns	0.33 ns	1.20
create(1,2)+last()+subscribe	1132.56 ns	0.40 ns	0.33 ns	1.20
name	rxcpp	rpp	prev rpp	ratio
immediate scheduler create worker + schedule	763.76 ns	0.60 ns	1.00 ns	0.60
current_thread scheduler create worker + schedule	927.87 ns	6.82 ns	8.68 ns	0.79
current_thread scheduler create worker + schedule + recursive schedule	2153.87 ns	88.66 ns	89.75 ns	0.99
name	rxcpp	rpp	prev rpp	ratio
create+map(v*2)+subscribe	880.55 ns	0.40 ns	0.33 ns	1.20
create+scan(10, std::plus)+subscribe	1049.49 ns	0.40 ns	0.33 ns	1.20
create+flat_map(just(v*2))+subscribe	3209.21 ns	151.82 ns	150.84 ns	1.01
name	rxcpp	rpp	prev rpp	ratio
create+take_while(false)+subscribe	903.86 ns	-	-	0.00
create+take_while(true)+subscribe	865.19 ns	0.40 ns	0.33 ns	1.20
name	rxcpp	rpp	prev rpp	ratio
create(create(1), create(1)) + merge() + subscribe	4329.10 ns	181.23 ns	177.23 ns	1.02
create(1) + merge_with(create(2)) + subscribe	6850.65 ns	180.07 ns	176.67 ns	1.02
create(1) + with_latest_from(create(2)) + subscribe	-	141.19 ns	145.63 ns	0.97
name	rxcpp	rpp	prev rpp	ratio
Subscribe empty callbacks to empty observable	1076.68 ns	1.13 ns	1.12 ns	1.01
Subscribe empty callbacks to empty observable via pipe operator	1142.52 ns	1.12 ns	1.12 ns	1.00
name	rxcpp	rpp	prev rpp	ratio
from array of 1 - create + subscribe + immediate	2335.02 ns	0.44 ns	0.44 ns	1.01
from array of 1 - create + subscribe + current_thread	2933.83 ns	114.46 ns	102.52 ns	1.12
concat_as_source of just(1 immediate) create + subscribe	6806.94 ns	0.29 ns	0.28 ns	1.05
name	rxcpp	rpp	prev rpp	ratio
create+take(1)+subscribe	2019.37 ns	3.91 ns	4.18 ns	0.94
create+filter(true)+subscribe	1172.14 ns	2.78 ns	2.52 ns	1.10
create(1,2)+skip(1)+subscribe	1846.19 ns	5.46 ns	5.60 ns	0.98
create(1,1,2)+distinct_until_changed()+subscribe	1190.35 ns	8.39 ns	8.41 ns	1.00
create(1,2)+first()+subscribe	2348.94 ns	5.67 ns	5.91 ns	0.96
create(1,2)+last()+subscribe	1494.95 ns	7.69 ns	8.14 ns	0.94