From cbfc8a05d81686f4e236da87fc74c172d6c9104d Mon Sep 17 00:00:00 2001
From: Aleksey Loginov <victimsnino@gmail.com>
Date: Mon, 12 Dec 2022 22:17:35 +0300
Subject: [PATCH 1/3] Use shared_ptr for buffer

---
 src/rpp/rpp/operators/buffer.hpp | 48 +++++++++++++++-----------------
 src/rpp/rpp/utils/functors.hpp   | 32 ++++++++++-----------
 2 files changed, 38 insertions(+), 42 deletions(-)

diff --git a/src/rpp/rpp/operators/buffer.hpp b/src/rpp/rpp/operators/buffer.hpp
index 77d13297e..7261779e2 100644
--- a/src/rpp/rpp/operators/buffer.hpp
+++ b/src/rpp/rpp/operators/buffer.hpp
@@ -26,21 +26,19 @@ namespace rpp::details
 {
 /// A non-copyable class that provides a copyable on_next for the subscriber and
 /// allows copies of on_next(s) to share the same states.
-template<constraint::decayed_type UpstreamType>
+template<constraint::decayed_type UpstreamType, constraint::subscriber_of_type<UpstreamType> Subscriber>
 struct buffer_state
 {
-    /// \param count Number of items being bundled. Note when count == 0, we'll
-    /// treat the behavior like when count == 1.
-    explicit buffer_state(size_t count)
+    template<constraint::decayed_same_as<Subscriber> TSub>
+    explicit buffer_state(size_t count, TSub&& sub)
         : max(std::max(size_t{1}, count))
+        , subscriber{std::forward<TSub>(sub)}
     {
         clear_and_reserve_buckets();
     }
 
     buffer_state(const buffer_state& other)          = delete;
-    buffer_state(buffer_state&&) noexcept            = default;
-    buffer_state& operator=(const buffer_state&)     = delete;
-    buffer_state& operator=(buffer_state&&) noexcept = default;
+    buffer_state(buffer_state&&) noexcept            = delete;
 
     void clear_and_reserve_buckets() const
     {
@@ -48,32 +46,33 @@ struct buffer_state
         buckets.reserve(max);
     }
 
-    const size_t                             max;
-    mutable buffer_bundle_type<UpstreamType> buckets;
+    const size_t                     max;
+    buffer_bundle_type<UpstreamType> buckets;
+    Subscriber                       subscriber;
 };
 
 struct buffer_on_next
 {
-    template<constraint::decayed_type UpstreamType>
-    void operator()(auto&& value, const auto& subscriber, const buffer_state<UpstreamType>& state) const
+    template<typename T, constraint::decayed_type UpstreamType, constraint::subscriber_of_type<UpstreamType> Subscriber>
+    void operator()(T&& value, const std::shared_ptr<buffer_state<UpstreamType, Subscriber>>& state) const
     {
-        state.buckets.push_back(std::forward<decltype(value)>(value));
+        state->buckets.push_back(std::forward<T>(value));
         if (state.buckets.size() == state.max)
         {
-            subscriber.on_next(std::move(state.buckets));
-            state.clear_and_reserve_buckets();
+            state->subscriber.on_next(std::move(state.buckets));
+            state->clear_and_reserve_buckets();
         }
     }
 };
 
 struct buffer_on_completed
 {
-    template<constraint::decayed_type UpstreamType>
-    void operator()(const auto& subscriber, const buffer_state<UpstreamType>& state) const
+    template<constraint::decayed_type UpstreamType, constraint::subscriber_of_type<UpstreamType> Subscriber>
+    void operator()(const buffer_state<UpstreamType, Subscriber>& state) const
     {
         if (!state.buckets.empty())
-            subscriber.on_next(std::move(state.buckets));
-        subscriber.on_completed();
+            state->subscriber.on_next(std::move(state.buckets));
+        state->subscriber.on_completed();
     }
 };
 
@@ -86,14 +85,13 @@ struct buffer_impl
     auto operator()(TSub&& subscriber) const
     {
         auto subscription = subscriber.get_subscription();
+        auto state        = std::make_shared<buffer_state<Type, std::decay_t<TSub>>>(count, std::forward<TSub>(subscriber));
 
-        // dynamic_state there to make shared_ptr for observer instead of making shared_ptr for state
-        return create_subscriber_with_dynamic_state<Type>(std::move(subscription),
-                                                          buffer_on_next{},
-                                                          utils::forwarding_on_error{},
-                                                          buffer_on_completed{},
-                                                          std::forward<TSub>(subscriber),
-                                                          buffer_state<Type>{count});
+        return create_subscriber_with_state<Type>(std::move(subscription),
+                                                  buffer_on_next{},
+                                                  utils::forwarding_on_error{},
+                                                  buffer_on_completed{},
+                                                  std::move(state));
     }
 };
 
diff --git a/src/rpp/rpp/utils/functors.hpp b/src/rpp/rpp/utils/functors.hpp
index adb55bcf0..f3605f218 100644
--- a/src/rpp/rpp/utils/functors.hpp
+++ b/src/rpp/rpp/utils/functors.hpp
@@ -13,6 +13,7 @@
 #include <rpp/subscribers/constraints.hpp>
 
 #include <exception>
+#include <memory>
 #include <utility>
 #include <tuple>
 
@@ -62,31 +63,28 @@ struct get
 
 struct forwarding_on_next
 {
-    void operator()(auto&& v, const auto& sub, const auto&...) const { sub.on_next(std::forward<decltype(v)>(v)); }
-};
+    template<typename T, constraint::subscriber_of_type<std::decay_t<T>> TSub>
+    void operator()(T&& v, const TSub& sub, const auto&...) const { sub.on_next(std::forward<T>(v)); }
 
-struct forwarding_on_error
-{
-    void operator()(const std::exception_ptr& err, const auto& sub, const auto&...) const { sub.on_error(err); }
+    template<typename T, typename State>
+    void operator()(T&& v, const std::shared_ptr<State>& state, const auto&...) const { state->subscriber.on_next(std::forward<T>(v)); }
 };
 
-struct forwarding_on_completed
+struct forwarding_on_error
 {
-    void operator()(const auto& sub, const auto&...) const { sub.on_completed(); }
-};
+    template<constraint::subscriber TSub>
+    void operator()(const std::exception_ptr& err, const TSub& sub, const auto&...) const { sub.on_error(err); }
 
-struct forwarding_on_next_for_pointer
-{
-    void operator()(auto&& v, const auto& sub) const { sub->on_next(std::forward<decltype(v)>(v)); }
+    template<typename State>
+    void operator()(const std::exception_ptr& err, const std::shared_ptr<State>& state, const auto&...) const { state->subscriber.on_error(err); }
 };
 
-struct forwarding_on_error_for_pointer
+struct forwarding_on_completed
 {
-    void operator()(const std::exception_ptr& err, const auto& sub) const { sub->on_error(err); }
-};
+    template<constraint::subscriber TSub>
+    void operator()(const TSub& sub, const auto&...) const { sub.on_completed(); }
 
-struct forwarding_on_completed_for_pointer
-{
-    void operator()(const auto& sub) const { sub->on_completed(); }
+    template<typename State>
+    void operator()(const std::shared_ptr<State>& state, const auto&...) const { state->subscriber.on_completed(); }
 };
 } // namespace rpp::utils

From 056e69cff0b33a6af9fa44dfa6be53adc26ebb9e Mon Sep 17 00:00:00 2001
From: Aleksey Loginov <victimsnino@gmail.com>
Date: Mon, 12 Dec 2022 22:21:57 +0300
Subject: [PATCH 2/3] Fix

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

diff --git a/src/rpp/rpp/operators/buffer.hpp b/src/rpp/rpp/operators/buffer.hpp
index 7261779e2..58d1c37a3 100644
--- a/src/rpp/rpp/operators/buffer.hpp
+++ b/src/rpp/rpp/operators/buffer.hpp
@@ -26,7 +26,7 @@ namespace rpp::details
 {
 /// A non-copyable class that provides a copyable on_next for the subscriber and
 /// allows copies of on_next(s) to share the same states.
-template<constraint::decayed_type UpstreamType, constraint::subscriber_of_type<UpstreamType> Subscriber>
+template<constraint::decayed_type UpstreamType, constraint::subscriber Subscriber>
 struct buffer_state
 {
     template<constraint::decayed_same_as<Subscriber> TSub>
@@ -40,7 +40,7 @@ struct buffer_state
     buffer_state(const buffer_state& other)          = delete;
     buffer_state(buffer_state&&) noexcept            = delete;
 
-    void clear_and_reserve_buckets() const
+    void clear_and_reserve_buckets()
     {
         buckets.clear();
         buckets.reserve(max);
@@ -53,13 +53,13 @@ struct buffer_state
 
 struct buffer_on_next
 {
-    template<typename T, constraint::decayed_type UpstreamType, constraint::subscriber_of_type<UpstreamType> Subscriber>
+    template<typename T, constraint::decayed_type UpstreamType, constraint::subscriber Subscriber>
     void operator()(T&& value, const std::shared_ptr<buffer_state<UpstreamType, Subscriber>>& state) const
     {
         state->buckets.push_back(std::forward<T>(value));
-        if (state.buckets.size() == state.max)
+        if (state->buckets.size() == state->max)
         {
-            state->subscriber.on_next(std::move(state.buckets));
+            state->subscriber.on_next(std::move(state->buckets));
             state->clear_and_reserve_buckets();
         }
     }
@@ -67,11 +67,11 @@ struct buffer_on_next
 
 struct buffer_on_completed
 {
-    template<constraint::decayed_type UpstreamType, constraint::subscriber_of_type<UpstreamType> Subscriber>
-    void operator()(const buffer_state<UpstreamType, Subscriber>& state) const
+    template<constraint::decayed_type UpstreamType, constraint::subscriber Subscriber>
+    void operator()(const std::shared_ptr<buffer_state<UpstreamType, Subscriber>>& state) const
     {
-        if (!state.buckets.empty())
-            state->subscriber.on_next(std::move(state.buckets));
+        if (!state->buckets.empty())
+            state->subscriber.on_next(std::move(state->buckets));
         state->subscriber.on_completed();
     }
 };

From 8893ba913871e6f2ff2b54f0df44a0886da4b454 Mon Sep 17 00:00:00 2001
From: Aleksey Loginov <victimsnino@gmail.com>
Date: Mon, 12 Dec 2022 22:37:21 +0300
Subject: [PATCH 3/3] fix

---
 src/rpp/rpp/utils/functors.hpp | 9 +++++++++
 1 file changed, 9 insertions(+)

diff --git a/src/rpp/rpp/utils/functors.hpp b/src/rpp/rpp/utils/functors.hpp
index f3605f218..bc12e8283 100644
--- a/src/rpp/rpp/utils/functors.hpp
+++ b/src/rpp/rpp/utils/functors.hpp
@@ -66,6 +66,9 @@ struct forwarding_on_next
     template<typename T, constraint::subscriber_of_type<std::decay_t<T>> TSub>
     void operator()(T&& v, const TSub& sub, const auto&...) const { sub.on_next(std::forward<T>(v)); }
 
+    template<typename T, constraint::observer_of_type<std::decay_t<T>> TObs>
+    void operator()(T&& v, const TObs& obs, const auto&...) const { obs.on_next(std::forward<T>(v)); }
+
     template<typename T, typename State>
     void operator()(T&& v, const std::shared_ptr<State>& state, const auto&...) const { state->subscriber.on_next(std::forward<T>(v)); }
 };
@@ -75,6 +78,9 @@ struct forwarding_on_error
     template<constraint::subscriber TSub>
     void operator()(const std::exception_ptr& err, const TSub& sub, const auto&...) const { sub.on_error(err); }
 
+    template<constraint::observer TObs>
+    void operator()(const std::exception_ptr& err, const TObs& obs, const auto&...) const { obs.on_error(err); }
+
     template<typename State>
     void operator()(const std::exception_ptr& err, const std::shared_ptr<State>& state, const auto&...) const { state->subscriber.on_error(err); }
 };
@@ -84,6 +90,9 @@ struct forwarding_on_completed
     template<constraint::subscriber TSub>
     void operator()(const TSub& sub, const auto&...) const { sub.on_completed(); }
 
+    template<constraint::observer TObs>
+    void operator()(const TObs& obs, const auto&...) const { obs.on_completed(); }
+
     template<typename State>
     void operator()(const std::shared_ptr<State>& state, const auto&...) const { state->subscriber.on_completed(); }
 };