decouple pairwise from observable

5 files changed, 105 insertions, 33 deletions

diff --git a/Rx/v2/src/rxcpp/operators/rx-pairwise.hpp b/Rx/v2/src/rxcpp/operators/rx-pairwise.hpp
index 8531e8c..584a6bf 100644
--- a/Rx/v2/src/rxcpp/operators/rx-pairwise.hpp
+++ b/Rx/v2/src/rxcpp/operators/rx-pairwise.hpp
@@ -2,6 +2,21 @@
 
 #pragma once
 
+/*! \file rx-pairwise.hpp
+
+    \brief Take values pairwise from this observable.
+
+    \return Observable that emits tuples of two the most recent items emitted by the source observable.
+
+    \sample
+    \snippet pairwise.cpp pairwise sample
+    \snippet output.txt pairwise sample
+
+    If the source observable emits less than two items, no pairs are emitted  by the source observable:
+    \snippet pairwise.cpp pairwise short sample
+    \snippet output.txt pairwise short sample
+*/
+
 #if !defined(RXCPP_OPERATORS_RX_PAIRWISE_HPP)
 #define RXCPP_OPERATORS_RX_PAIRWISE_HPP
 
@@ -13,6 +28,16 @@ namespace operators {
 
 namespace detail {
 
+template<class... AN>
+struct pairwise_invalid_arguments {};
+
+template<class... AN>
+struct pairwise_invalid : public rxo::operator_base<pairwise_invalid_arguments<AN...>> {
+    using type = observable<pairwise_invalid_arguments<AN...>, pairwise_invalid<AN...>>;
+};
+template<class... AN>
+using pairwise_invalid_t = typename pairwise_invalid<AN...>::type;
+
 template<class T>
 struct pairwise
 {
@@ -62,25 +87,40 @@ struct pairwise
     }
 };
 
-class pairwise_factory
-{
-public:
-    template<class Observable>
-    auto operator()(Observable&& source)
-        -> decltype(source.template lift<rxu::value_type_t<pairwise<rxu::value_type_t<rxu::decay_t<Observable>>>>>(pairwise<rxu::value_type_t<rxu::decay_t<Observable>>>())) {
-        return      source.template lift<rxu::value_type_t<pairwise<rxu::value_type_t<rxu::decay_t<Observable>>>>>(pairwise<rxu::value_type_t<rxu::decay_t<Observable>>>());
-    }
-};
-
 }
 
-inline auto pairwise()
-    ->      detail::pairwise_factory {
-    return  detail::pairwise_factory();
+/*! @copydoc rx-pairwise.hpp
+*/
+template<class... AN>
+auto pairwise(AN&&... an)
+    ->     operator_factory<pairwise_tag, AN...> {
+    return operator_factory<pairwise_tag, AN...>(std::make_tuple(std::forward<AN>(an)...));
 }
 
 }
 
+template<>
+struct member_overload<pairwise_tag>
+{
+    template<class Observable,
+        class Enabled = rxu::enable_if_all_true_type_t<
+            is_observable<Observable>>,
+        class SourceValue = rxu::value_type_t<Observable>,
+        class Pairwise = rxo::detail::pairwise<SourceValue>,
+        class Value = rxu::value_type_t<Pairwise>>
+    static auto member(Observable&& o)
+    -> decltype(o.template lift<Value>(Pairwise())) {
+        return  o.template lift<Value>(Pairwise());
+    }
+
+    template<class... AN>
+    static operators::detail::pairwise_invalid_t<AN...> member(AN...) {
+        std::terminate();
+        return {};
+        static_assert(sizeof...(AN) == 10000, "pairwise takes no arguments");
+    }
+};
+
 }
 
 #endif
diff --git a/Rx/v2/src/rxcpp/rx-includes.hpp b/Rx/v2/src/rxcpp/rx-includes.hpp
index 283793e..214b46c 100644
--- a/Rx/v2/src/rxcpp/rx-includes.hpp
+++ b/Rx/v2/src/rxcpp/rx-includes.hpp
@@ -196,6 +196,7 @@
 #include "operators/rx-ignore_elements.hpp"
 #include "operators/rx-map.hpp"
 #include "operators/rx-on_error_resume_next.hpp"
+#include "operators/rx-pairwise.hpp"
 #include "operators/rx-reduce.hpp"
 #include "operators/rx-sequence_equal.hpp"
 #include "operators/rx-take_while.hpp"
diff --git a/Rx/v2/src/rxcpp/rx-observable.hpp b/Rx/v2/src/rxcpp/rx-observable.hpp
index ff68936..ae67d2b 100644
--- a/Rx/v2/src/rxcpp/rx-observable.hpp
+++ b/Rx/v2/src/rxcpp/rx-observable.hpp
@@ -2958,26 +2958,15 @@ public:
         return      rxo::start_with(std::move(v0), std::move(vn)...)(*this);
     }
 
-    /*! Take values pairwise from this observable.
-
-        \return  Observable that emits tuples of two the most recent items emitted by the source observable.
-
-        \sample
-        \snippet pairwise.cpp pairwise sample
-        \snippet output.txt pairwise sample
-
-        If the source observable emits less than two items, no pairs are emitted  by the source observable:
-        \snippet pairwise.cpp pairwise short sample
-        \snippet output.txt pairwise short sample
-    */
+    /*! @copydoc rx-pairwise.hpp
+     */
     template<class... AN>
-    auto pairwise(AN**...) const
+    auto pairwise(AN... an) const
         /// \cond SHOW_SERVICE_MEMBERS
-        -> decltype(EXPLICIT_THIS lift<rxu::value_type_t<rxo::detail::pairwise<T>>>(rxo::detail::pairwise<T>()))
+        -> decltype(observable_member(pairwise_tag{}, *(this_type*)nullptr, std::forward<AN>(an)...))
         /// \endcond
     {
-        return                    lift<rxu::value_type_t<rxo::detail::pairwise<T>>>(rxo::detail::pairwise<T>());
-        static_assert(sizeof...(AN) == 0, "pairwise() was passed too many arguments.");
+        return      observable_member(pairwise_tag{},                *this, std::forward<AN>(an)...);
     }
 };
 
diff --git a/Rx/v2/src/rxcpp/rx-operators.hpp b/Rx/v2/src/rxcpp/rx-operators.hpp
index 975c4e6..5c7621f 100644
--- a/Rx/v2/src/rxcpp/rx-operators.hpp
+++ b/Rx/v2/src/rxcpp/rx-operators.hpp
@@ -107,7 +107,6 @@ public:
 #include "operators/rx-merge.hpp"
 #include "operators/rx-multicast.hpp"
 #include "operators/rx-observe_on.hpp"
-#include "operators/rx-pairwise.hpp"
 #include "operators/rx-publish.hpp"
 #include "operators/rx-ref_count.hpp"
 #include "operators/rx-repeat.hpp"
@@ -266,6 +265,13 @@ struct take_while_tag {
     };
 };
 
+struct pairwise_tag {
+    template<class Included>
+    struct include_header{
+        static_assert(Included::value, "missing include: please #include <rxcpp/operators/rx-pairwise.hpp>");
+    };
+};
+
 struct retry_tag {
     template<class Included>
     struct include_header{
diff --git a/Rx/v2/test/operators/pairwise.cpp b/Rx/v2/test/operators/pairwise.cpp
index 20da0d8..7ba0c42 100644
--- a/Rx/v2/test/operators/pairwise.cpp
+++ b/Rx/v2/test/operators/pairwise.cpp
@@ -1,6 +1,7 @@
 #include "../test.h"
+#include "rxcpp/operators/rx-pairwise.hpp"
 
-SCENARIO("pairwise", "[pairwise][operators]") {
+SCENARIO("pairwise - enough items to create pairs", "[pairwise][operators]") {
     GIVEN("a cold observable of n ints") {
         auto sc = rxsc::make_test();
         auto w = sc.create_worker();
@@ -22,9 +23,9 @@ SCENARIO("pairwise", "[pairwise][operators]") {
             auto res = w.start(
                 [xs, &invoked]() {
                     return xs
-                        .pairwise()
+                        | rxo::pairwise()
                         // forget type to workaround lambda deduction bug on msvc 2013
-                        .as_dynamic();
+                        | rxo::as_dynamic();
                 }
             );
 
@@ -43,3 +44,38 @@ SCENARIO("pairwise", "[pairwise][operators]") {
         }
     }
 }
+
+SCENARIO("pairwise - not enough items to create a pair", "[pairwise][operators]") {
+    GIVEN("a cold observable of 1 ints") {
+        auto sc = rxsc::make_test();
+        auto w = sc.create_worker();
+        const rxsc::test::messages<int> on;
+        const rxsc::test::messages<std::tuple<int, int>> on_pairwise;
+
+        auto xs = sc.make_cold_observable({
+            on.next(180, 1),
+            on.completed(400),
+        });
+
+        WHEN("taken pairwise") {
+
+            auto res = w.start(
+                [xs]() {
+                    return xs
+                        .pairwise()
+                        // forget type to workaround lambda deduction bug on msvc 2013
+                        .as_dynamic();
+                }
+            );
+
+            THEN("the output contains no tuples of ints"){
+                auto delay = rxcpp::schedulers::test::subscribed_time;
+                auto required = rxu::to_vector({
+                    on_pairwise.completed(400 + delay)
+                });
+                auto actual = res.get_observer().messages();
+                REQUIRE(required == actual);
+            }
+        }
+    }
+}