decouple group_by lifter from observable

6 files changed, 129 insertions, 71 deletions

diff --git a/Rx/v2/src/rxcpp/operators/rx-group_by.hpp b/Rx/v2/src/rxcpp/operators/rx-group_by.hpp
index 6414f22..4eadbab 100644
--- a/Rx/v2/src/rxcpp/operators/rx-group_by.hpp
+++ b/Rx/v2/src/rxcpp/operators/rx-group_by.hpp
@@ -2,6 +2,30 @@
 
 #pragma once
 
+/*! \file rx-group_by.hpp
+
+    \brief Return an observable that emits grouped_observables, each of which corresponds to a unique key value and each of which emits those items from the source observable that share that key value.
+
+    \tparam KeySelector      the type of the key extracting function
+    \tparam MarbleSelector   the type of the element extracting function
+    \tparam BinaryPredicate  the type of the key comparing function
+
+    \param  ks  a function that extracts the key for each item (optional)
+    \param  ms  a function that extracts the return element for each item (optional)
+    \param  p   a function that implements comparison of two keys (optional)
+
+    \return  Observable that emits values of grouped_observable type, each of which corresponds to a unique key value and each of which emits those items from the source observable that share that key value.
+
+    \sample
+    \snippet group_by.cpp group_by full intro
+    \snippet group_by.cpp group_by full sample
+    \snippet output.txt group_by full sample
+
+    \sample
+    \snippet group_by.cpp group_by sample
+    \snippet output.txt group_by sample
+*/
+
 #if !defined(RXCPP_OPERATORS_RX_GROUP_BY_HPP)
 #define RXCPP_OPERATORS_RX_GROUP_BY_HPP
 
@@ -13,6 +37,16 @@ namespace operators {
 
 namespace detail {
 
+template<class... AN>
+struct group_by_invalid_arguments {};
+
+template<class... AN>
+struct group_by_invalid : public rxo::operator_base<group_by_invalid_arguments<AN...>> {
+    using type = observable<group_by_invalid_arguments<AN...>, group_by_invalid<AN...>>;
+};
+template<class... AN>
+using group_by_invalid_t = typename group_by_invalid<AN...>::type;
+
 template<class T, class Selector>
 struct is_group_by_selector_for {
 
@@ -241,25 +275,76 @@ public:
 
 }
 
-template<class KeySelector, class MarbleSelector, class BinaryPredicate>
-inline auto group_by(KeySelector ks, MarbleSelector ms, BinaryPredicate p)
-    ->      detail::group_by_factory<KeySelector, MarbleSelector, BinaryPredicate> {
-    return  detail::group_by_factory<KeySelector, MarbleSelector, BinaryPredicate>(std::move(ks), std::move(ms), std::move(p));
+/*! @copydoc rx-group_by.hpp
+*/
+template<class... AN>
+auto group_by(AN&&... an) 
+    ->     operator_factory<group_by_tag, AN...> {
+    return operator_factory<group_by_tag, AN...>(std::make_tuple(std::forward<AN>(an)...));
 }
 
-template<class KeySelector, class MarbleSelector>
-inline auto group_by(KeySelector ks, MarbleSelector ms)
-    ->      detail::group_by_factory<KeySelector, MarbleSelector, rxu::less> {
-    return  detail::group_by_factory<KeySelector, MarbleSelector, rxu::less>(std::move(ks), std::move(ms), rxu::less(), identity_current_thread());
 }
 
-template<class KeySelector>
-inline auto group_by(KeySelector ks)
-    ->      detail::group_by_factory<KeySelector, rxu::detail::take_at<0>, rxu::less> {
-    return  detail::group_by_factory<KeySelector, rxu::detail::take_at<0>, rxu::less>(std::move(ks), rxu::take_at<0>(), rxu::less());
-}
+template<> 
+struct member_overload<group_by_tag>
+{
+    template<class Observable, class KeySelector, class MarbleSelector, class BinaryPredicate, 
+        class SourceValue = rxu::value_type_t<Observable>,
+        class Traits = rxo::detail::group_by_traits<SourceValue, rxu::decay_t<Observable>, KeySelector, MarbleSelector, BinaryPredicate>,
+        class GroupBy = rxo::detail::group_by<SourceValue, rxu::decay_t<Observable>, rxu::decay_t<KeySelector>, rxu::decay_t<MarbleSelector>, rxu::decay_t<BinaryPredicate>>,
+        class Value = typename Traits::grouped_observable_type>
+    static auto member(Observable&& o, KeySelector&& ks, MarbleSelector&& ms, BinaryPredicate&& p)
+        -> decltype(o.template lift<Value>(GroupBy(std::forward<KeySelector>(ks), std::forward<MarbleSelector>(ms), std::forward<BinaryPredicate>(p)))) {
+        return      o.template lift<Value>(GroupBy(std::forward<KeySelector>(ks), std::forward<MarbleSelector>(ms), std::forward<BinaryPredicate>(p)));
+    }
 
-}
+    template<class Observable, class KeySelector, class MarbleSelector, 
+        class BinaryPredicate=rxu::less, 
+        class SourceValue = rxu::value_type_t<Observable>,
+        class Traits = rxo::detail::group_by_traits<SourceValue, rxu::decay_t<Observable>, KeySelector, MarbleSelector, BinaryPredicate>,
+        class GroupBy = rxo::detail::group_by<SourceValue, rxu::decay_t<Observable>, rxu::decay_t<KeySelector>, rxu::decay_t<MarbleSelector>, rxu::decay_t<BinaryPredicate>>,
+        class Value = typename Traits::grouped_observable_type>
+    static auto member(Observable&& o, KeySelector&& ks, MarbleSelector&& ms)
+        -> decltype(o.template lift<Value>(GroupBy(std::forward<KeySelector>(ks), std::forward<MarbleSelector>(ms), rxu::less()))) {
+        return      o.template lift<Value>(GroupBy(std::forward<KeySelector>(ks), std::forward<MarbleSelector>(ms), rxu::less()));
+    }
+
+
+    template<class Observable, class KeySelector, 
+        class MarbleSelector=rxu::detail::take_at<0>, 
+        class BinaryPredicate=rxu::less, 
+        class SourceValue = rxu::value_type_t<Observable>,
+        class Traits = rxo::detail::group_by_traits<SourceValue, rxu::decay_t<Observable>, KeySelector, MarbleSelector, BinaryPredicate>,
+        class GroupBy = rxo::detail::group_by<SourceValue, rxu::decay_t<Observable>, rxu::decay_t<KeySelector>, rxu::decay_t<MarbleSelector>, rxu::decay_t<BinaryPredicate>>,
+        class Value = typename Traits::grouped_observable_type>
+    static auto member(Observable&& o, KeySelector&& ks)
+        -> decltype(o.template lift<Value>(GroupBy(std::forward<KeySelector>(ks), rxu::detail::take_at<0>(), rxu::less()))) {
+        return      o.template lift<Value>(GroupBy(std::forward<KeySelector>(ks), rxu::detail::take_at<0>(), rxu::less()));
+    }
+
+    template<class Observable, 
+        class KeySelector=rxu::detail::take_at<0>, 
+        class MarbleSelector=rxu::detail::take_at<0>, 
+        class BinaryPredicate=rxu::less, 
+        class Enabled = rxu::enable_if_all_true_type_t<
+            all_observables<Observable>>,
+        class SourceValue = rxu::value_type_t<Observable>,
+        class Traits = rxo::detail::group_by_traits<SourceValue, rxu::decay_t<Observable>, KeySelector, MarbleSelector, BinaryPredicate>,
+        class GroupBy = rxo::detail::group_by<SourceValue, rxu::decay_t<Observable>, rxu::decay_t<KeySelector>, rxu::decay_t<MarbleSelector>, rxu::decay_t<BinaryPredicate>>,
+        class Value = typename Traits::grouped_observable_type>
+    static auto member(Observable&& o)
+        -> decltype(o.template lift<Value>(GroupBy(rxu::detail::take_at<0>(), rxu::detail::take_at<0>(), rxu::less()))) {
+        return      o.template lift<Value>(GroupBy(rxu::detail::take_at<0>(), rxu::detail::take_at<0>(), rxu::less()));
+    }
+
+    template<class... AN>
+    static operators::detail::group_by_invalid_t<AN...> member(const AN&...) {
+        std::terminate();
+        return {};
+        static_assert(sizeof...(AN) == 10000, "group_by takes (optional KeySelector, optional MarbleSelector, optional BinaryKeyPredicate), KeySelector takes (Observable::value_type) -> KeyValue, MarbleSelector takes (Observable::value_type) -> MarbleValue, BinaryKeyPredicate takes (KeyValue, KeyValue) -> bool");
+    } 
+
+};
 
 }
 
diff --git a/Rx/v2/src/rxcpp/operators/rx-lift.hpp b/Rx/v2/src/rxcpp/operators/rx-lift.hpp
index 4e95cb2..4d85f77 100644
--- a/Rx/v2/src/rxcpp/operators/rx-lift.hpp
+++ b/Rx/v2/src/rxcpp/operators/rx-lift.hpp
@@ -20,10 +20,14 @@ struct is_lift_function_for {
     template<class CS, class CF>
     static tag_not_valid check(...);
 
-    typedef rxu::decay_t<S> for_type;
-    typedef rxu::decay_t<F> func_type;
-    typedef decltype(check<for_type, func_type>(0)) detail_result;
-    static const bool value = is_subscriber<detail_result>::value && is_subscriber<for_type>::value && std::is_convertible<V, typename rxu::value_type_from<detail_result>::type>::value;
+    using for_type = rxu::decay_t<S>;
+    using func_type = rxu::decay_t<F>;
+    using detail_result = decltype(check<for_type, func_type>(0));
+
+    static const bool value = rxu::all_true_type<
+        is_subscriber<detail_result>,
+        is_subscriber<for_type>,
+        std::is_convertible<V, typename rxu::value_type_from<detail_result>::type>>::value;
 };
 
 }
@@ -40,8 +44,6 @@ struct lift_traits
     typedef rxu::decay_t<Operator> operator_type;
 
     typedef typename source_operator_type::value_type source_value_type;
-
-    static const bool value = rxcpp::detail::is_lift_function_for<source_value_type, subscriber<result_value_type>, operator_type>::value;
 };
 
 template<class ResultType, class SourceOperator, class Operator>
diff --git a/Rx/v2/src/rxcpp/rx-includes.hpp b/Rx/v2/src/rxcpp/rx-includes.hpp
index c9d2794..7ae0c5f 100644
--- a/Rx/v2/src/rxcpp/rx-includes.hpp
+++ b/Rx/v2/src/rxcpp/rx-includes.hpp
@@ -182,6 +182,7 @@
 #include "rx-grouped_observable.hpp"
 
 #if !defined(RXCPP_LITE)
+#include "operators/rx-group_by.hpp"
 #include "operators/rx-zip.hpp"
 #endif
 
diff --git a/Rx/v2/src/rxcpp/rx-observable.hpp b/Rx/v2/src/rxcpp/rx-observable.hpp
index 2f34df3..a4b62bf 100644
--- a/Rx/v2/src/rxcpp/rx-observable.hpp
+++ b/Rx/v2/src/rxcpp/rx-observable.hpp
@@ -2472,56 +2472,18 @@ public:
         return      observable_member(zip_tag{}, *this,                std::forward<AN>(an)...);
     }
 
-    /*! Return an observable that emits grouped_observables, each of which corresponds to a unique key value and each of which emits those items from the source observable that share that key value.
-
-        \tparam KeySelector      the type of the key extracting function
-        \tparam MarbleSelector   the type of the element extracting function
-        \tparam BinaryPredicate  the type of the key comparing function
-
-        \param  ks  a function that extracts the key for each item
-        \param  ms  a function that extracts the return element for each item
-        \param  p   a function that implements comparison of two keys
-
-        \return  Observable that emits values of grouped_observable type, each of which corresponds to a unique key value and each of which emits those items from the source observable that share that key value.
-
-        \sample
-        \snippet group_by.cpp group_by full intro
-        \snippet group_by.cpp group_by full sample
-        \snippet output.txt group_by full sample
-    */
-    template<class KeySelector, class MarbleSelector, class BinaryPredicate>
-    inline auto group_by(KeySelector ks, MarbleSelector ms, BinaryPredicate p) const
-        /// \cond SHOW_SERVICE_MEMBERS
-        -> decltype(EXPLICIT_THIS lift<typename rxo::detail::group_by_traits<T, this_type, KeySelector, MarbleSelector, BinaryPredicate>::grouped_observable_type>(rxo::detail::group_by<T, this_type, KeySelector, MarbleSelector, BinaryPredicate>(std::move(ks), std::move(ms), std::move(p))))
-        /// \endcond
-    {
-        return                    lift<typename rxo::detail::group_by_traits<T, this_type, KeySelector, MarbleSelector, BinaryPredicate>::grouped_observable_type>(rxo::detail::group_by<T, this_type, KeySelector, MarbleSelector, BinaryPredicate>(std::move(ks), std::move(ms), std::move(p)));
-    }
-
-    /*! Return an observable that emits grouped_observables, each of which corresponds to a unique key value and each of which emits those items from the source observable that share that key value.
-
-        \tparam KeySelector     the type of the key extracting function
-        \tparam MarbleSelector  the type of the element extracting function
-
-        \param  ks  a function that extracts the key for each item
-        \param  ms  a function that extracts the return element for each item
-
-        \return  Observable that emits values of grouped_observable type, each of which corresponds to a unique key value and each of which emits those items from the source observable that share that key value.
-
-        \sample
-        \snippet group_by.cpp group_by sample
-        \snippet output.txt group_by sample
-    */
-    template<class KeySelector, class MarbleSelector>
-    inline auto group_by(KeySelector ks, MarbleSelector ms) const
+    /*! @copydoc rx-group_by.hpp
+     */
+    template<class... AN>
+    inline auto group_by(AN&&... an) const
         /// \cond SHOW_SERVICE_MEMBERS
-        -> decltype(EXPLICIT_THIS lift<typename rxo::detail::group_by_traits<T, this_type, KeySelector, MarbleSelector, rxu::less>::grouped_observable_type>(rxo::detail::group_by<T, this_type, KeySelector, MarbleSelector, rxu::less>(std::move(ks), std::move(ms), rxu::less())))
+        -> decltype(observable_member(group_by_tag{}, *(this_type*)nullptr, std::forward<AN>(an)...))
         /// \endcond
     {
-        return                    lift<typename rxo::detail::group_by_traits<T, this_type, KeySelector, MarbleSelector, rxu::less>::grouped_observable_type>(rxo::detail::group_by<T, this_type, KeySelector, MarbleSelector, rxu::less>(std::move(ks), std::move(ms), rxu::less()));
+        return      observable_member(group_by_tag{}, *this,                std::forward<AN>(an)...);
     }
 
-      /*! Do not emit any items from the source Observable, but allow termination notification (either onError or onCompleted) to pass through unchanged.
+    /*! Do not emit any items from the source Observable, but allow termination notification (either onError or onCompleted) to pass through unchanged.
 
         \return  Observable that emits termination notification from the source observable.
 
diff --git a/Rx/v2/src/rxcpp/rx-operators.hpp b/Rx/v2/src/rxcpp/rx-operators.hpp
index 7fc6256..a121983 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-filter.hpp"
 #include "operators/rx-finally.hpp"
 #include "operators/rx-flat_map.hpp"
-#include "operators/rx-group_by.hpp"
 #include "operators/rx-ignore_elements.hpp"
 #include "operators/rx-lift.hpp"
 #include "operators/rx-map.hpp"
@@ -147,6 +146,7 @@ public:
 #include "operators/rx-window_toggle.hpp"
 
 namespace rxcpp {
+
 struct zip_tag {
     template<class Included>
     struct include_header{
@@ -154,6 +154,13 @@ struct zip_tag {
     };
 };
 
+struct group_by_tag {
+    template<class Included>
+    struct include_header{
+        static_assert(Included::value, "missing include: please #include <rxcpp/operators/rx-group_by.hpp>");
+    };
+};
+
 }
 
 #endif
diff --git a/Rx/v2/test/operators/group_by.cpp b/Rx/v2/test/operators/group_by.cpp
index 6a35279..152f647 100644
--- a/Rx/v2/test/operators/group_by.cpp
+++ b/Rx/v2/test/operators/group_by.cpp
@@ -1,4 +1,5 @@
 #include "../test.h"
+#include <rxcpp/operators/rx-group_by.hpp>
 
 #include <locale>
 
@@ -285,7 +286,7 @@ SCENARIO("group_by take 1", "[group_by][take][operators]"){
             auto res = w.start(
                 [&]() {
                     return xs
-                        .group_by(
+                        | rxo::group_by(
                             [&](long v) {
                                 ++keyInvoked;
                                 return v % 2;
@@ -294,14 +295,14 @@ SCENARIO("group_by take 1", "[group_by][take][operators]"){
                                 ++marbleInvoked;
                                 return v;
                             })
-                        .take(1)
-                        .map([&](const rxcpp::grouped_observable<long, long>& g) -> rxcpp::observable<long> {
+                        | rxo::take(1)
+                        | rxo::map([&](const rxcpp::grouped_observable<long, long>& g) -> rxcpp::observable<long> {
                             ++groupEmitted;
                             return g;
                         })
-                        .merge()
+                        | rxo::merge()
                         // forget type to workaround lambda deduction bug on msvc 2013
-                        .as_dynamic();
+                        | rxo::as_dynamic();
                 }
             );