Implement flat_map operator

src/benchmarks/benchmarks.cpp

@@ -297,6 +297,22 @@ int main(int argc, char* argv[]) // NOLINT
                     | rxcpp::operators::subscribe<int>([](int v){ ankerl::nanobench::doNotOptimizeAway(v); });
             });
         }
+        SECTION("create+flat_map(just(v*2))+subscribe")
+        {
+            TEST_RPP([&]()
+            {
+                rpp::source::create<int>([](const auto& obs){ obs.on_next(1); })
+                    | 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<>::create<int>([v](const auto& obs){ obs.on_next(v * 2); }); })
+                    | rxcpp::operators::subscribe<int>([](int v){ ankerl::nanobench::doNotOptimizeAway(v); });
+            });
+        }
     };
 
     BENCHMARK("Filtering Operators")

src/rpp/rpp/operators.hpp

@@ -30,6 +30,7 @@
  * @ingroup operators
  */
 
+#include <rpp/operators/flat_map.hpp>
 #include <rpp/operators/map.hpp>
 #include <rpp/operators/scan.hpp>
 #include <rpp/operators/subscribe.hpp>

src/rpp/rpp/operators/flat_map.hpp

@@ -0,0 +1,68 @@
+//                  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/map.hpp>
+#include <rpp/operators/merge.hpp>
+#include <rpp/defs.hpp>
+
+namespace rpp::operators::details
+{
+
+template<rpp::constraint::decayed_type Fn>
+struct flat_map_t 
+{
+    RPP_NO_UNIQUE_ADDRESS Fn m_fn;
+
+    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)
+                | rpp::ops::map(m_fn)
+                | rpp::ops::merge();
+
+    }
+
+    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::move(m_fn))
+                | rpp::ops::merge();
+    }
+};
+
+} // namespace rpp::operators::details
+
+namespace rpp::operators
+{
+
+/**
+ * @brief Transform the items emitted by an Observable into Observables, then flatten the emissions from those into a single Observable
+ *
+ * @details Note that flat_map merges the emissions of these Observables, so that they may interleave. 
+ *
+ * @param a function that returns an observable for each item emitted by the source observable. 
+ * @warning #include <rpp/operators/flat_map.hpp>
+ *
+ * @ingroup transforming_operators
+ * @see https://reactivex.io/documentation/operators/flatmap.html
+ */
+template<typename Fn>
+    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)};
+}
+
+} // namespace rpp::operators

src/rpp/rpp/operators/fwd.hpp

@@ -33,6 +33,10 @@ template<typename Fn>
     requires (!utils::is_not_template_callable<Fn> || !std::same_as<void, std::invoke_result_t<Fn, utils::convertible_to_any>>)
 auto map(Fn&& callable);
 
+template<typename Fn>
+    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>
     requires constraint::observables_of_same_type<std::decay_t<TObservable>, std::decay_t<TObservables>...>
 auto merge_with(TObservable&& observable, TObservables&& ...observables);

src/tests/rpp/test_flat_map.cpp

@@ -0,0 +1,148 @@
+//                  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 <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>
+#include <rpp/sources/never.hpp>
+#include <rpp/schedulers/immediate.hpp>
+
+#include "copy_count_tracker.hpp"
+#include "mock_observer.hpp"
+
+#include <stdexcept>
+#include <string>
+
+TEMPLATE_TEST_CASE("flat_map", "", rpp::memory_model::use_stack, rpp::memory_model::use_shared)
+{
+	auto mock = mock_observer_strategy<int>();
+	SECTION("observable of items")
+    {
+        auto obs = rpp::source::just<TestType>(rpp::schedulers::immediate{}, 1, 2, 3);
+
+        SECTION("subscribe using flat_map")
+        {
+            obs | rpp::operators::flat_map([](int v) { return rpp::source::just(v * 2); })
+                | rpp::ops::subscribe(mock.get_observer());
+            SECTION("observer obtains values from underlying observables")
+            {
+                CHECK(mock.get_received_values() == std::vector{ 2, 4, 6 });
+                CHECK(mock.get_on_completed_count() == 1);
+                CHECK(mock.get_on_error_count() == 0);
+            }
+        }
+        SECTION("subscribe using flat_map with error")
+        {
+            obs | rpp::operators::flat_map([](int) { return rpp::source::error<int>(std::make_exception_ptr(std::runtime_error{""})); })
+                | rpp::ops::subscribe(mock.get_observer());
+            SECTION("observer obtains values from underlying observables")
+            {
+                CHECK(mock.get_total_on_next_count() == 0);
+                CHECK(mock.get_on_completed_count() == 0);
+                CHECK(mock.get_on_error_count() == 1);
+            }
+        }
+        SECTION("subscribe using flat_map with empty")
+        {
+            obs | rpp::operators::flat_map([](int) { return rpp::source::empty<int>(); })
+                | rpp::ops::subscribe(mock.get_observer());
+            SECTION("observer obtains values from underlying observables")
+            {
+                CHECK(mock.get_total_on_next_count() == 0);
+                CHECK(mock.get_on_completed_count() == 1);
+                CHECK(mock.get_on_error_count() == 0);
+            }
+        }
+        SECTION("subscribe using flat_map with never")
+        {
+            obs | rpp::operators::flat_map([](int) { return rpp::source::never<int>(); })
+                | rpp::ops::subscribe(mock.get_observer());
+            SECTION("observer obtains values from underlying observables")
+            {
+                CHECK(mock.get_total_on_next_count() == 0);
+                CHECK(mock.get_on_completed_count() == 0);
+                CHECK(mock.get_on_error_count() == 0);
+            }
+        }
+        SECTION("subscribe using flat_map with empty in middle")
+        {
+            obs | rpp::operators::flat_map([](int v) {
+                    if (v == 2) 
+                        return rpp::source::empty<int>().as_dynamic();
+                    return rpp::source::just(v).as_dynamic(); 
+                  })
+                | rpp::ops::subscribe(mock.get_observer());
+            SECTION("observer obtains values from underlying observables")
+            {
+                CHECK(mock.get_received_values() == std::vector{ 1, 3 });
+                CHECK(mock.get_on_completed_count() == 1);
+                CHECK(mock.get_on_error_count() == 0);
+            }
+        }
+        SECTION("subscribe using flat_map with error in middle")
+        {
+            obs | rpp::operators::flat_map([](int v) {
+                    if (v == 2) 
+                        return rpp::source::error<int>(std::make_exception_ptr(std::runtime_error{""})).as_dynamic();
+                    return rpp::source::just(v).as_dynamic(); 
+                  })
+                | rpp::ops::subscribe(mock.get_observer());
+            SECTION("observer obtains values from underlying observables")
+            {
+                CHECK(mock.get_total_on_next_count() == 0);
+                CHECK(mock.get_on_completed_count() == 0);
+                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::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([](auto){}); // subscribe by value so one additional copy
+            SECTION("no extra copies")
+            {
+                REQUIRE(verifier.get_copy_count() == 1 + 1);
+            }
+        }
+    }
+}