Address comments

victimsnino · Jun 10, 2023 · a748a6f · a748a6f
1 parent 4f58a84
commit a748a6f
Show file tree

Hide file tree

Showing 4 changed files with 39 additions and 17 deletions.
diff --git a/src/benchmarks/benchmarks.cpp b/src/benchmarks/benchmarks.cpp
@@ -302,14 +302,14 @@ int main(int argc, char* argv[]) // NOLINT
             TEST_RPP([&]()
             {
                 rpp::source::create<int>([](const auto& obs){ obs.on_next(1); })
-                    | rpp::operators::flat_map([](int v) { return rpp::source::just(v * 2); })
+                    | rpp::operators::flat_map([](int v) {  return rpp::source::create<int>([v](const auto& obs){ obs.on_next(v * 2); }); })
                     | rpp::operators::subscribe([](int v){ ankerl::nanobench::doNotOptimizeAway(v); });
             });
 
             TEST_RXCPP([&]()
             {
                 rxcpp::observable<>::create<int>([](const auto& obs){obs.on_next(1);})
-                    | rxcpp::operators::flat_map([](int v) { return rxcpp::observable<>::just(v * 2); })
+                    | rxcpp::operators::flat_map([](int v) { return rxcpp::observable<>::create<int>([v](const auto& obs){ obs.on_next(v * 2); }); })
                     | rxcpp::operators::subscribe<int>([](int v){ ankerl::nanobench::doNotOptimizeAway(v); });
             });
         }

diff --git a/src/rpp/rpp/operators/flat_map.hpp b/src/rpp/rpp/operators/flat_map.hpp
@@ -22,8 +22,8 @@ struct flat_map_t
 {
     RPP_NO_UNIQUE_ADDRESS Fn m_fn;
 
-    template<rpp::constraint::observable TObservable>
-        requires std::invocable<Fn, rpp::utils::extract_observable_type_t<TObservable>>
+    template<rpp::constraint::observable TObservable, typename ValueType = rpp::utils::extract_observable_type_t<TObservable>>
+        requires (std::invocable<Fn, ValueType> && rpp::constraint::observable<std::invoke_result_t<Fn, ValueType>>)
     auto operator()(TObservable&& observable) const &
     {
         return std::forward<TObservable>(observable)
@@ -32,12 +32,12 @@ struct flat_map_t
 
     }
 
-    template<rpp::constraint::observable TObservable>
-        requires std::invocable<Fn, rpp::utils::extract_observable_type_t<TObservable>>
+    template<rpp::constraint::observable TObservable, typename ValueType = rpp::utils::extract_observable_type_t<TObservable>>
+        requires (std::invocable<Fn, ValueType> && rpp::constraint::observable<std::invoke_result_t<Fn, ValueType>>)
     auto operator()(TObservable&& observable) &&
     {
         return std::forward<TObservable>(observable)
-                | rpp::ops::map(std::forward<Fn>(m_fn))
+                | rpp::ops::map(std::move(m_fn))
                 | rpp::ops::merge();
     }
 };
@@ -59,7 +59,7 @@ namespace rpp::operators
  * @see https://reactivex.io/documentation/operators/flatmap.html
  */
 template<typename Fn>
-    requires rpp::constraint::observable<std::invoke_result_t<Fn, utils::convertible_to_any>>
+    requires (!utils::is_not_template_callable<Fn> || rpp::constraint::observable<std::invoke_result_t<Fn, utils::convertible_to_any>>)
 auto flat_map(Fn&& callable)
 {
     return details::flat_map_t<std::decay_t<Fn>>{std::forward<Fn>(callable)};

diff --git a/src/rpp/rpp/operators/fwd.hpp b/src/rpp/rpp/operators/fwd.hpp
@@ -34,7 +34,7 @@ template<typename Fn>
 auto map(Fn&& callable);
 
 template<typename Fn>
-    requires rpp::constraint::observable<std::invoke_result_t<Fn, utils::convertible_to_any>>
+    requires (!utils::is_not_template_callable<Fn> || rpp::constraint::observable<std::invoke_result_t<Fn, utils::convertible_to_any>>)
 auto flat_map(Fn&& callable);
 
 template<rpp::constraint::observable TObservable, rpp::constraint::observable... TObservables>

diff --git a/src/tests/rpp/test_flat_map.cpp b/src/tests/rpp/test_flat_map.cpp
@@ -12,6 +12,7 @@
 #include <snitch/snitch_macros_check.hpp>
 
 #include <rpp/operators/flat_map.hpp>
+#include <rpp/sources/create.hpp>
 #include <rpp/sources/just.hpp>
 #include <rpp/sources/empty.hpp>
 #include <rpp/sources/error.hpp>
@@ -105,22 +106,43 @@ TEMPLATE_TEST_CASE("flat_map", "", rpp::memory_model::use_stack, rpp::memory_mod
                 CHECK(mock.get_on_error_count() == 1);
             }
         }
+        SECTION("subscribe using flat_map with templated lambda")
+        {
+            obs | rpp::operators::flat_map([](auto v) { return rpp::source::just(v * 2); })
+                | rpp::ops::subscribe(mock.get_observer());
+            SECTION("observer obtains values from underlying observables")
+            {
+                CHECK(mock.get_on_completed_count() == 1);
+            }
+        }
     }
-    /*
+}
+
+TEST_CASE("flat_map copies/moves")
+{
     SECTION("flat_map doesn't produce extra copies")
     {
         copy_count_tracker verifier{};
-        auto obs = rpp::source::just<TestType>(std::move(verifier))
-                 | rpp::ops::flat_map([](copy_count_tracker&& verifier) { return rpp::source::just<TestType>(std::move(verifier)); });
-        SECTION("subscribe")
+        auto obs = rpp::source::create<copy_count_tracker>([verifier = std::move(verifier)](const auto& obs) { obs.on_next(verifier); })
+                 | rpp::ops::map([](copy_count_tracker verifier) { return std::move(verifier); }) // copy from source to map
+                 | rpp::ops::flat_map([](copy_count_tracker&& verifier) { // no copy
+                    return rpp::source::create<copy_count_tracker>([verifier = std::move(verifier)](const auto& obs) { obs.on_next(std::move(verifier)); }); 
+                   });
+        SECTION("first subscribe")
+        {
+            obs.subscribe([](const auto&){}); // subscribe by const lvalue reference so no copy
+            SECTION("no extra copies")
+            {
+                REQUIRE(verifier.get_copy_count() == 1); // only one copy from source to first operator
+            }
+        }
+        SECTION("second subscribe")
         {
-            obs.subscribe([](const auto&){});
+            obs.subscribe([](auto){}); // subscribe by value so one additional copy
             SECTION("no extra copies")
             {
-                REQUIRE(verifier.get_copy_count() == ??);
-                REQUIRE(verifier.get_move_count() == ??);
+                REQUIRE(verifier.get_copy_count() == 1 + 1);
             }
         }
     }
-    */
 }