Add on_error_resume_next operator (#528)

victimsnino · Feb 18, 2024 · 151959c · 151959c · github-actions · Feb 18, 2024
1 parent 09b74b8
commit 151959c
Show file tree

Hide file tree

Showing 4 changed files with 298 additions and 0 deletions.
diff --git a/src/benchmarks/benchmarks.cpp b/src/benchmarks/benchmarks.cpp
@@ -581,6 +581,30 @@ int main(int argc, char* argv[]) // NOLINT(bugprone-exception-escape)
         }
     } // BENCHMARK("Aggregating Operators")
 
+    BENCHMARK("Error Handling Operators")
+    {
+        SECTION("create(on_next(1), on_error())+on_error_resume_next(immediate_just(2)))+subscribe")
+        {
+            TEST_RPP([&]() {
+                rpp::source::create<int>([&](auto&& observer) {
+                    observer.on_next(1);
+                    observer.on_error(std::make_exception_ptr(std::runtime_error{""}));
+                })
+                    | rpp::operators::on_error_resume_next([](const std::exception_ptr&) { return rpp::immediate_just(2); })
+                    | rpp::operators::subscribe([](int v) { ankerl::nanobench::doNotOptimizeAway(v); });
+            });
+
+            TEST_RXCPP([&]() {
+                rxcpp::observable<>::create<int>([&](auto&& observer) {
+                    observer.on_next(1);
+                    observer.on_error(std::make_exception_ptr(std::runtime_error{""}));
+                })
+                    | rxcpp::operators::on_error_resume_next([](const std::exception_ptr&) { return rxcpp::immediate_just(2); })
+                    | rxcpp::operators::subscribe<int>([](int v) { ankerl::nanobench::doNotOptimizeAway(v); });
+            });
+        }
+    } // BENCHMARK("Error Handling Operators")
+
     BENCHMARK("Subjects")
     {
         SECTION("publish_subject with 1 observer - on_next")

diff --git a/src/rpp/rpp/operators.hpp b/src/rpp/rpp/operators.hpp
@@ -117,3 +117,12 @@
 
 #include <rpp/operators/concat.hpp>
 #include <rpp/operators/reduce.hpp>
+
+/**
+ * @defgroup error_handling_operators Error Handling Operators
+ * @brief Operators that help to recover from error notifications from an Observable
+ * @see https://reactivex.io/documentation/operators.html#error
+ * @ingroup operators
+ */
+
+#include <rpp/operators/on_error_resume_next.hpp>
diff --git a/src/rpp/rpp/operators/on_error_resume_next.hpp b/src/rpp/rpp/operators/on_error_resume_next.hpp
@@ -0,0 +1,109 @@
+//                  ReactivePlusPlus library
+//
+//          Copyright Aleksey Loginov 2023 - present.
+// Distributed under the Boost Software License, Version 1.0.
+//    (See accompanying file LICENSE_1_0.txt or copy at
+//          https://www.boost.org/LICENSE_1_0.txt)
+//
+// Project home: https://github.com/victimsnino/ReactivePlusPlus
+//
+
+#pragma once
+
+#include <rpp/operators/fwd.hpp>
+
+#include <rpp/defs.hpp>
+#include <rpp/operators/details/strategy.hpp>
+
+namespace rpp::operators::details
+{
+    template<rpp::constraint::observer TObserver, rpp::constraint::decayed_type Selector>
+    struct on_error_resume_next_observer_strategy
+    {
+        using preferred_disposable_strategy = rpp::details::observers::none_disposable_strategy;
+
+        RPP_NO_UNIQUE_ADDRESS mutable TObserver observer;
+        RPP_NO_UNIQUE_ADDRESS Selector          selector;
+        // Manually control disposable to ensure observer is not used after move in on_error emission
+        mutable rpp::composite_disposable_wrapper disposable = composite_disposable_wrapper::make();
+
+        RPP_CALL_DURING_CONSTRUCTION(
+            {
+                observer.set_upstream(disposable);
+            });
+
+        template<typename T>
+        void on_next(T&& v) const
+        {
+            observer.on_next(std::forward<T>(v));
+        }
+
+        void on_error(const std::exception_ptr& err) const
+        {
+            disposable.dispose();
+            selector(err).subscribe(std::move(observer));
+        }
+
+        void on_completed() const
+        {
+            disposable.dispose();
+            observer.on_completed();
+        }
+
+        void set_upstream(const disposable_wrapper& d)
+        {
+            disposable.add(d);
+        }
+
+        bool is_disposed() const { return disposable.is_disposed(); }
+    };
+
+    template<rpp::constraint::decayed_type Selector>
+    struct on_error_resume_next_t : lift_operator<on_error_resume_next_t<Selector>, Selector>
+    {
+        template<rpp::constraint::decayed_type T>
+        struct operator_traits
+        {
+            using selector_observable_result_type =
+                rpp::utils::extract_observable_type_t<std::invoke_result_t<Selector, std::exception_ptr>>;
+
+            static_assert(
+                rpp::constraint::decayed_same_as<selector_observable_result_type, T>,
+                "Selector observable result type is not the same as T");
+
+            using result_type = T;
+
+            template<rpp::constraint::observer_of_type<result_type> TObserver>
+            using observer_strategy = on_error_resume_next_observer_strategy<TObserver, Selector>;
+        };
+
+        template<rpp::details::observables::constraint::disposable_strategy Prev>
+        using updated_disposable_strategy = rpp::details::observables::atomic_dynamic_disposable_strategy_selector<1>;
+    };
+} // namespace rpp::operators::details
+
+namespace rpp::operators
+{
+    /**
+     * @brief If an error occurs, take the result from the Selector and subscribe to that instead.
+     *
+     * @marble on_error_resume_next
+       {
+           source observable                                  : +-1-x
+           operator "on_error_resume_next: () => obs(-2-3-|)" : +-1-2-3-|
+       }
+     *
+     * @param selector callable taking a std::exception_ptr and returning observable to continue on
+     *
+     * @warning #include <rpp/operators/on_error_resume_next.hpp>
+     *
+     * @ingroup error_handling_operators
+     * @see https://reactivex.io/documentation/operators/catch.html
+     */
+    template<typename Selector>
+        requires rpp::constraint::observable<std::invoke_result_t<Selector, std::exception_ptr>>
+    auto on_error_resume_next(Selector&& selector)
+    {
+        return details::on_error_resume_next_t<std::decay_t<Selector>>{std::forward<Selector>(selector)};
+    }
+} // namespace rpp::operators
diff --git a/src/tests/rpp/test_on_error_resume_next.cpp b/src/tests/rpp/test_on_error_resume_next.cpp
@@ -0,0 +1,156 @@
+//                  ReactivePlusPlus library
+//
+//          Copyright Aleksey Loginov 2023 - present.
+// Distributed under the Boost Software License, Version 1.0.
+//    (See accompanying file LICENSE_1_0.txt or copy at
+//          https://www.boost.org/LICENSE_1_0.txt)
+//
+// Project home: https://github.com/victimsnino/ReactivePlusPlus
+//
+
+#include <snitch/snitch.hpp>
+
+#include <rpp/operators/on_error_resume_next.hpp>
+#include <rpp/schedulers/immediate.hpp>
+#include <rpp/sources/create.hpp>
+#include <rpp/sources/empty.hpp>
+#include <rpp/sources/just.hpp>
+
+#include "disposable_observable.hpp"
+#include "mock_observer.hpp"
+
+TEMPLATE_TEST_CASE("on_error_resume_next switches observable on error", "", rpp::memory_model::use_stack, rpp::memory_model::use_shared)
+{
+    auto mock = mock_observer_strategy<int>();
+    SECTION("observable without error emission")
+    {
+        auto obs = rpp::source::just<TestType>(rpp::schedulers::immediate{}, 1, 2, 3);
+        SECTION("subscribe")
+        {
+            obs | rpp::operators::on_error_resume_next([](const std::exception_ptr&) { return rpp::source::empty<int>(); })
+                | rpp::ops::subscribe(mock);
+            SECTION("observer obtains values from observable")
+            {
+                CHECK(mock.get_received_values() == std::vector{1, 2, 3});
+                CHECK(mock.get_total_on_next_count() == 3);
+                CHECK(mock.get_on_error_count() == 0);
+                CHECK(mock.get_on_completed_count() == 1);
+            }
+        }
+    }
+
+    SECTION("observable with one error emission")
+    {
+        auto obs = rpp::source::create<int>([](const auto& sub) {
+            sub.on_next(1);
+            sub.on_next(2);
+            sub.on_error(std::make_exception_ptr(std::runtime_error{""}));
+        });
+        SECTION("subscribe")
+        {
+            obs | rpp::operators::on_error_resume_next([](const std::exception_ptr&) {
+                return rpp::source::just<TestType>(rpp::schedulers::immediate{}, 3);
+            })
+                | rpp::ops::subscribe(mock);
+            SECTION("observer obtains values from both outer and inner observable")
+            {
+                CHECK(mock.get_received_values() == std::vector{1, 2, 3});
+                CHECK(mock.get_total_on_next_count() == 3);
+                CHECK(mock.get_on_error_count() == 0);
+                CHECK(mock.get_on_completed_count() == 1);
+            }
+        }
+    }
+
+    SECTION("observable with two error emissions")
+    {
+        auto obs = rpp::source::create<int>([](const auto& sub) {
+            sub.on_next(1);
+            sub.on_next(2);
+            sub.on_error(std::make_exception_ptr(std::runtime_error{""}));
+            sub.on_error(std::make_exception_ptr(std::runtime_error{""}));
+        });
+        SECTION("subscribe")
+        {
+            obs | rpp::operators::on_error_resume_next([](const std::exception_ptr&) {
+                return rpp::source::just<TestType>(rpp::schedulers::immediate{}, 3);
+            })
+                | rpp::ops::subscribe(mock);
+            SECTION("observer only receives values from first inner observable")
+            {
+                CHECK(mock.get_received_values() == std::vector{1, 2, 3});
+                CHECK(mock.get_total_on_next_count() == 3);
+                CHECK(mock.get_on_error_count() == 0);
+                CHECK(mock.get_on_completed_count() == 1);
+            }
+        }
+    }
+
+    SECTION("inner observable different emission type")
+    {
+        auto obs = rpp::source::create<int>([](const auto& sub) {
+            sub.on_next(1);
+            sub.on_next(2);
+            sub.on_error(std::make_exception_ptr(std::runtime_error{""}));
+            sub.on_error(std::make_exception_ptr(std::runtime_error{""}));
+        });
+        SECTION("subscribe")
+        {
+            obs | rpp::operators::on_error_resume_next([](const std::exception_ptr&) {
+                return rpp::source::just<TestType>(rpp::schedulers::immediate{}, 3);
+            })
+                | rpp::ops::subscribe(mock);
+            SECTION("observer only receives values from first inner observable")
+            {
+                CHECK(mock.get_received_values() == std::vector{1, 2, 3});
+                CHECK(mock.get_total_on_next_count() == 3);
+                CHECK(mock.get_on_error_count() == 0);
+                CHECK(mock.get_on_completed_count() == 1);
+            }
+        }
+    }
+
+    SECTION("nested on_error_resume_next operators")
+    {
+        auto obs = rpp::source::create<int>([](const auto& sub) {
+            sub.on_next(1);
+            sub.on_error(std::make_exception_ptr(std::runtime_error{""}));
+        });
+        SECTION("subscribe")
+        {
+            obs | rpp::operators::on_error_resume_next([](const std::exception_ptr&) {
+                return rpp::source::create<int>([](const auto& sub) {
+                           sub.on_next(2);
+                           sub.on_error(std::make_exception_ptr(std::runtime_error{""}));
+                       })
+                     | rpp::operators::on_error_resume_next([](const std::exception_ptr&) {
+                           return rpp::source::create<int>([](const auto& sub) {
+                               sub.on_next(3);
+                               sub.on_completed();
+                           });
+                       });
+            })
+                | rpp::ops::subscribe(mock);
+            SECTION("observer receives values without any errors")
+            {
+                CHECK(mock.get_received_values() == std::vector{1, 2, 3});
+                CHECK(mock.get_total_on_next_count() == 3);
+                CHECK(mock.get_on_error_count() == 0);
+                CHECK(mock.get_on_completed_count() == 1);
+            }
+        }
+    }
+}
+
+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);
+
+        test_operator_with_disposable<int>(
+            rpp::ops::on_error_resume_next([](const std::exception_ptr&) { return rpp::source::empty<int>(); }));
+    }
+
+    CHECK(observable_disposable.is_disposed() || observable_disposable.lock().use_count() == 2);
+}
name	rxcpp	rpp	prev rpp	ratio
Subscribe empty callbacks to empty observable	302.48 ns	2.16 ns	2.16 ns	1.00
Subscribe empty callbacks to empty observable via pipe operator	303.05 ns	2.16 ns	2.16 ns	1.00
name	rxcpp	rpp	prev rpp	ratio
from array of 1 - create + subscribe + immediate	690.03 ns	0.31 ns	0.62 ns	0.50
from array of 1 - create + subscribe + current_thread	1029.86 ns	5.25 ns	5.25 ns	1.00
concat_as_source of just(1 immediate) create + subscribe	4020.41 ns	184.60 ns	121.85 ns	1.51
defer from array of 1 - defer + create + subscribe + immediate	1286.79 ns	0.63 ns	0.31 ns	2.03
interval - interval + take(3) + subscribe + immediate	3031.00 ns	61.93 ns	58.01 ns	1.07
interval - interval + take(3) + subscribe + current_thread	3342.33 ns	41.53 ns	32.11 ns	1.29
name	rxcpp	rpp	prev rpp	ratio
immediate_just+take(1)+subscribe	1067.97 ns	0.31 ns	0.31 ns	1.00
immediate_just+filter(true)+subscribe	836.34 ns	0.31 ns	0.32 ns	0.96
immediate_just(1,2)+skip(1)+subscribe	982.51 ns	0.34 ns	0.32 ns	1.08
immediate_just(1,1,2)+distinct_until_changed()+subscribe	880.07 ns	0.31 ns	0.32 ns	0.98
immediate_just(1,2)+first()+subscribe	1279.21 ns	0.62 ns	0.32 ns	1.94
immediate_just(1,2)+last()+subscribe	957.44 ns	0.31 ns	0.32 ns	0.98
immediate_just+take_last(1)+subscribe	1090.91 ns	17.59 ns	18.52 ns	0.95
name	rxcpp	rpp	prev rpp	ratio
immediate scheduler create worker + schedule	282.40 ns	2.92 ns	2.16 ns	1.35
current_thread scheduler create worker + schedule	370.25 ns	8.15 ns	7.41 ns	1.10
current_thread scheduler create worker + schedule + recursive schedule	814.45 ns	68.25 ns	65.58 ns	1.04
name	rxcpp	rpp	prev rpp	ratio
immediate_just+map(v*2)+subscribe	817.79 ns	0.31 ns	0.31 ns	1.00
immediate_just+scan(10, std::plus)+subscribe	885.00 ns	0.31 ns	0.31 ns	1.00
immediate_just+flat_map(immediate_just(v*2))+subscribe	2344.85 ns	153.71 ns	176.43 ns	0.87
immediate_just+buffer(2)+subscribe	1553.48 ns	13.89 ns	13.89 ns	1.00
immediate_just+window(2)+subscribe + subscsribe inner	2377.34 ns	1066.27 ns	1202.84 ns	0.89
name	rxcpp	rpp	prev rpp	ratio
immediate_just+take_while(false)+subscribe	824.61 ns	-	-	0.00
immediate_just+take_while(true)+subscribe	834.65 ns	0.31 ns	0.31 ns	1.00
name	rxcpp	rpp	prev rpp	ratio
immediate_just(immediate_just(1), immediate_just(1)) + merge() + subscribe	3466.86 ns	177.62 ns	199.58 ns	0.89
immediate_just(1) + merge_with(immediate_just(2)) + subscribe	3958.73 ns	189.10 ns	171.54 ns	1.10
immediate_just(1) + with_latest_from(immediate_just(2)) + subscribe	-	136.30 ns	134.21 ns	1.02
immediate_just(immediate_just(1),immediate_just(1)) + switch_on_next() + subscribe	3504.83 ns	1197.14 ns	966.65 ns	1.24
immediate_just(1) + zip(immediate_just(2)) + subscribe	2134.11 ns	213.22 ns	212.65 ns	1.00
name	rxcpp	rpp	prev rpp	ratio
basic sample	1375.83 ns	16.37 ns	16.66 ns	0.98
basic sample with immediate scheduler	1441.12 ns	5.56 ns	7.71 ns	0.72
name	rxcpp	rpp	prev rpp	ratio
Subscribe empty callbacks to empty observable	1008.45 ns	3.91 ns	3.98 ns	0.98
Subscribe empty callbacks to empty observable via pipe operator	984.84 ns	3.93 ns	3.86 ns	1.02
name	rxcpp	rpp	prev rpp	ratio
from array of 1 - create + subscribe + immediate	1935.91 ns	0.23 ns	0.23 ns	1.00
from array of 1 - create + subscribe + current_thread	2293.58 ns	25.26 ns	25.15 ns	1.00
concat_as_source of just(1 immediate) create + subscribe	5124.40 ns	305.69 ns	331.76 ns	0.92
defer from array of 1 - defer + create + subscribe + immediate	2490.09 ns	0.24 ns	0.23 ns	1.04
interval - interval + take(3) + subscribe + immediate	5603.02 ns	125.56 ns	114.71 ns	1.09
interval - interval + take(3) + subscribe + current_thread	6810.62 ns	121.03 ns	110.75 ns	1.09