From 9e5a83230f328cb61876c01b584bcdb0f9845fbf Mon Sep 17 00:00:00 2001
From: CorentinBT <corentinbt0@gmail.com>
Date: Sun, 18 Feb 2024 20:51:14 +0100
Subject: [PATCH 1/2] Add missing operators and sources fwd

---
 src/rpp/rpp/operators/fwd.hpp | 42 +++++++++++++++++++++++++++++++++++
 src/rpp/rpp/sources/fwd.hpp   | 23 +++++++++++++++++--
 2 files changed, 63 insertions(+), 2 deletions(-)

diff --git a/src/rpp/rpp/operators/fwd.hpp b/src/rpp/rpp/operators/fwd.hpp
index b1a6ba879..663bf948d 100644
--- a/src/rpp/rpp/operators/fwd.hpp
+++ b/src/rpp/rpp/operators/fwd.hpp
@@ -39,6 +39,8 @@ namespace rpp::operators
     template<rpp::schedulers::constraint::scheduler Scheduler>
     auto delay(rpp::schedulers::duration delay_duration, Scheduler&& scheduler);
 
+    auto distinct();
+
     template<typename EqualityFn = rpp::utils::equal_to>
         requires (!utils::is_not_template_callable<EqualityFn> || std::same_as<bool, std::invoke_result_t<EqualityFn, rpp::utils::convertible_to_any, rpp::utils::convertible_to_any>>)
     auto distinct_until_changed(EqualityFn&& equality_fn = {});
@@ -49,6 +51,10 @@ namespace rpp::operators
         requires (!utils::is_not_template_callable<Fn> || std::same_as<bool, std::invoke_result_t<Fn, rpp::utils::convertible_to_any>>)
     auto filter(Fn&& predicate);
 
+    template<typename Fn>
+        requires (!utils::is_not_template_callable<Fn> || rpp::constraint::observable<std::invoke_result_t<Fn, rpp::utils::convertible_to_any>>)
+    auto flat_map(Fn&& callable);
+
     template<typename KeySelector,
              typename ValueSelector = std::identity,
              typename KeyComparator = rpp::utils::less>
@@ -82,6 +88,13 @@ namespace rpp::operators
 
     auto publish();
 
+    template<typename Seed, typename Accumulator>
+        requires (!utils::is_not_template_callable<Accumulator> || std::same_as<std::decay_t<Seed>, std::invoke_result_t<Accumulator, std::decay_t<Seed> &&, rpp::utils::convertible_to_any>>)
+    auto reduce(Seed&& seed, Accumulator&& accumulator);
+
+    template<typename Accumulator>
+    auto reduce(Accumulator&& accumulator);
+
     auto ref_count();
 
     auto repeat(size_t count);
@@ -133,9 +146,31 @@ namespace rpp::operators
     template<rpp::constraint::observable TObservable>
     auto take_until(TObservable&& until_observable);
 
+    template<std::invocable<const std::exception_ptr&> OnError = rpp::utils::empty_function_t<std::exception_ptr>>
+    auto tap(OnError&& on_error);
+
+    template<std::invocable<> OnCompleted = rpp::utils::empty_function_t<>>
+    auto tap(OnCompleted&& on_completed);
+
+    template<typename OnNext,
+             std::invocable<> OnCompleted = rpp::utils::empty_function_t<>>
+    auto tap(OnNext&&      on_next,
+             OnCompleted&& on_completed);
+
+    template<typename OnNext                                       = rpp::utils::empty_function_any_t,
+             std::invocable<const std::exception_ptr&> OnError     = rpp::utils::empty_function_t<std::exception_ptr>,
+             std::invocable<>                          OnCompleted = rpp::utils::empty_function_t<>>
+    auto tap(OnNext&&      on_next      = {},
+             OnError&&     on_error     = {},
+             OnCompleted&& on_completed = {});
+
     template<rpp::schedulers::constraint::scheduler Scheduler = rpp::schedulers::immediate>
     auto throttle(rpp::schedulers::duration period);
 
+    template<typename Selector>
+        requires rpp::constraint::observable<std::invoke_result_t<Selector, std::exception_ptr>>
+    auto on_error_resume_next(Selector&& selector);
+
     template<typename TSelector, rpp::constraint::observable TObservable, rpp::constraint::observable... TObservables>
         requires (!rpp::constraint::observable<TSelector> && (!utils::is_not_template_callable<TSelector> || std::invocable<TSelector, rpp::utils::convertible_to_any, utils::extract_observable_type_t<TObservable>, utils::extract_observable_type_t<TObservables>...>))
     auto with_latest_from(TSelector&& selector, TObservable&& observable, TObservables&&... observables);
@@ -148,6 +183,13 @@ namespace rpp::operators
     template<rpp::constraint::observable TOpeningsObservable, typename TClosingsSelectorFn>
         requires rpp::constraint::observable<std::invoke_result_t<TClosingsSelectorFn, rpp::utils::extract_observable_type_t<TOpeningsObservable>>>
     auto window_toggle(TOpeningsObservable&& openings, TClosingsSelectorFn&& closings_selector);
+
+    template<typename TSelector, rpp::constraint::observable TObservable, rpp::constraint::observable... TObservables>
+        requires (!rpp::constraint::observable<TSelector> && (!utils::is_not_template_callable<TSelector> || std::invocable<TSelector, rpp::utils::convertible_to_any, utils::extract_observable_type_t<TObservable>, utils::extract_observable_type_t<TObservables>...>))
+    auto zip(TSelector&& selector, TObservable&& observable, TObservables&&... observables);
+
+    template<rpp::constraint::observable TObservable, rpp::constraint::observable... TObservables>
+    auto zip(TObservable&& observable, TObservables&&... observables);
 } // namespace rpp::operators
 
 namespace rpp
diff --git a/src/rpp/rpp/sources/fwd.hpp b/src/rpp/rpp/sources/fwd.hpp
index e10575fc5..08691c68c 100644
--- a/src/rpp/rpp/sources/fwd.hpp
+++ b/src/rpp/rpp/sources/fwd.hpp
@@ -61,12 +61,31 @@ namespace rpp::source
         requires constraint::observable<utils::iterable_value_t<Iterable>>
     auto concat(Iterable&& iterable);
 
-    template<constraint::decayed_type Type>
-    auto never();
+    template<std::invocable Factory>
+        requires rpp::constraint::observable<std::invoke_result_t<Factory>>
+    auto defer(Factory&& observable_factory);
 
     template<constraint::decayed_type Type>
     auto error(std::exception_ptr err);
 
     template<constraint::decayed_type Type>
     auto empty();
+
+    template<schedulers::constraint::scheduler TScheduler>
+    auto interval(rpp::schedulers::duration initial, rpp::schedulers::duration period, TScheduler&& scheduler);
+
+    template<schedulers::constraint::scheduler TScheduler>
+    auto interval(rpp::schedulers::time_point initial, rpp::schedulers::duration period, TScheduler&& scheduler);
+
+    template<schedulers::constraint::scheduler TScheduler>
+    auto interval(rpp::schedulers::duration period, TScheduler&& scheduler);
+
+    template<constraint::decayed_type Type>
+    auto never();
+
+    template<schedulers::constraint::scheduler TScheduler>
+    auto timer(rpp::schedulers::duration when, TScheduler&& scheduler);
+
+    template<schedulers::constraint::scheduler TScheduler>
+    auto timer(rpp::schedulers::time_point when, TScheduler&& scheduler);
 } // namespace rpp::source
\ No newline at end of file

From 9f8313543a0ad7c1a0b51652683a6d030e36d2a4 Mon Sep 17 00:00:00 2001
From: CorentinBT <corentinbt0@gmail.com>
Date: Sun, 18 Feb 2024 21:00:56 +0100
Subject: [PATCH 2/2] Fix build

---
 src/rpp/rpp/operators/tap.hpp | 18 +++++++++---------
 1 file changed, 9 insertions(+), 9 deletions(-)

diff --git a/src/rpp/rpp/operators/tap.hpp b/src/rpp/rpp/operators/tap.hpp
index d0d0fcb48..74d20b052 100644
--- a/src/rpp/rpp/operators/tap.hpp
+++ b/src/rpp/rpp/operators/tap.hpp
@@ -89,7 +89,7 @@ namespace rpp::operators
      * @ingroup utility_operators
      * @see https://reactivex.io/documentation/operators/do.html
      */
-    template<std::invocable<const std::exception_ptr&> OnError = rpp::utils::empty_function_t<std::exception_ptr>>
+    template<std::invocable<const std::exception_ptr&> OnError /* = rpp::utils::empty_function_t<std::exception_ptr> */>
     auto tap(OnError&& on_error)
     {
         using OnNext      = rpp::utils::empty_function_any_t;
@@ -109,7 +109,7 @@ namespace rpp::operators
      * @ingroup utility_operators
      * @see https://reactivex.io/documentation/operators/do.html
      */
-    template<std::invocable<> OnCompleted = rpp::utils::empty_function_t<>>
+    template<std::invocable<> OnCompleted /* = rpp::utils::empty_function_t<> */>
     auto tap(OnCompleted&& on_completed)
     {
         using OnNext  = rpp::utils::empty_function_any_t;
@@ -131,7 +131,7 @@ namespace rpp::operators
      * @see https://reactivex.io/documentation/operators/do.html
      */
     template<typename OnNext,
-             std::invocable<> OnCompleted = rpp::utils::empty_function_t<>>
+             std::invocable<> OnCompleted /* = rpp::utils::empty_function_t<> */>
     auto tap(OnNext&&      on_next,
              OnCompleted&& on_completed)
     {
@@ -153,12 +153,12 @@ namespace rpp::operators
      * @ingroup utility_operators
      * @see https://reactivex.io/documentation/operators/do.html
      */
-    template<typename OnNext                                       = rpp::utils::empty_function_any_t,
-             std::invocable<const std::exception_ptr&> OnError     = rpp::utils::empty_function_t<std::exception_ptr>,
-             std::invocable<>                          OnCompleted = rpp::utils::empty_function_t<>>
-    auto tap(OnNext&&      on_next      = {},
-             OnError&&     on_error     = {},
-             OnCompleted&& on_completed = {})
+    template<typename OnNext /* = rpp::utils::empty_function_any_t */,
+             std::invocable<const std::exception_ptr&> OnError /* = rpp::utils::empty_function_t<std::exception_ptr> */,
+             std::invocable<>                          OnCompleted /* = rpp::utils::empty_function_t<> */>
+    auto tap(OnNext&&      on_next /* = {} */,
+             OnError&&     on_error /* = {} */,
+             OnCompleted&& on_completed /* = {} */)
     {
         return details::tap_t<std::decay_t<OnNext>, std::decay_t<OnError>, std::decay_t<OnCompleted>>{
             std::forward<OnNext>(on_next),