20 #if !defined(RXCPP_OPERATORS_RX_DISTINCT_UNTIL_CHANGED_HPP) 21 #define RXCPP_OPERATORS_RX_DISTINCT_UNTIL_CHANGED_HPP 23 #include "../rx-includes.hpp" 32 struct distinct_until_changed_invalid_arguments {};
35 struct distinct_until_changed_invalid :
public rxo::operator_base<distinct_until_changed_invalid_arguments<AN...>> {
36 using type = observable<distinct_until_changed_invalid_arguments<
AN...>, distinct_until_changed_invalid<
AN...>>;
39 using distinct_until_changed_invalid_t =
typename distinct_until_changed_invalid<
AN...>::type;
41 template<
class T,
class BinaryPredicate>
44 typedef rxu::decay_t<T> source_value_type;
45 typedef rxu::decay_t<BinaryPredicate> predicate_type;
54 template<
class Subscriber>
55 struct distinct_until_changed_observer
57 typedef distinct_until_changed_observer<Subscriber> this_type;
58 typedef source_value_type value_type;
59 typedef rxu::decay_t<Subscriber> dest_type;
60 typedef observer<value_type, this_type> observer_type;
64 mutable rxu::detail::maybe<source_value_type> remembered;
66 distinct_until_changed_observer(dest_type d, predicate_type pred)
68 , pred(std::move(pred))
71 void on_next(source_value_type v)
const {
72 if (remembered.empty() || !pred(v, remembered.get())) {
77 void on_error(std::exception_ptr e)
const {
80 void on_completed()
const {
84 static subscriber<value_type, observer_type> make(dest_type d, predicate_type p) {
85 return make_subscriber<value_type>(d, this_type(d, std::move(p)));
89 template<
class Subscriber>
90 auto operator()(Subscriber dest)
const 91 -> decltype(distinct_until_changed_observer<Subscriber>::make(std::move(dest), pred)) {
92 return distinct_until_changed_observer<Subscriber>::make(std::move(dest), pred);
100 template<
class...
AN>
103 return operator_factory<distinct_until_changed_tag,
AN...>(std::make_tuple(std::forward<AN>(an)...));
111 template<
class Observable,
115 class DistinctUntilChanged = rxo::detail::distinct_until_changed<SourceValue, rxu::equal_to<>>>
117 -> decltype(o.template lift<SourceValue>(DistinctUntilChanged(
rxu::equal_to<>()))) {
118 return o.template lift<SourceValue>(DistinctUntilChanged(
rxu::equal_to<>()));
121 template<
class Observable,
122 class BinaryPredicate,
125 is_observable<Observable>>,
126 class DistinctUntilChanged = rxo::detail::distinct_until_changed<SourceValue, BinaryPredicate>>
127 static auto member(Observable&& o, BinaryPredicate&& pred)
128 -> decltype(o.template lift<SourceValue>(DistinctUntilChanged(std::forward<BinaryPredicate>(pred)))) {
129 return o.template lift<SourceValue>(DistinctUntilChanged(std::forward<BinaryPredicate>(pred)));
132 template<
class...
AN>
133 static operators::detail::distinct_until_changed_invalid_t<
AN...>
member(
AN...) {
136 static_assert(
sizeof...(
AN) == 10000,
"distinct_until_changed takes (optional BinaryPredicate)");
Definition: rx-all.hpp:26
typename std::decay< T >::type::value_type value_type_t
Definition: rx-util.hpp:35
Definition: rx-operators.hpp:69
auto distinct_until_changed(AN &&...an) -> operator_factory< distinct_until_changed_tag, AN... >
For each item from this observable, filter out consequentially repeated values and emit only changes ...
Definition: rx-distinct_until_changed.hpp:101
auto AN
Definition: rx-finally.hpp:105
Definition: rx-operators.hpp:199
Definition: rx-operators.hpp:47
typename std::enable_if< all_true_type< BN... >::value >::type enable_if_all_true_type_t
Definition: rx-util.hpp:114
static operators::detail::distinct_until_changed_invalid_t< AN... > member(AN...)
Definition: rx-distinct_until_changed.hpp:133
static auto member(Observable &&o) -> decltype(o.template lift< SourceValue >(DistinctUntilChanged(rxu::equal_to<>())))
Definition: rx-distinct_until_changed.hpp:116
Definition: rx-util.hpp:420
static auto member(Observable &&o, BinaryPredicate &&pred) -> decltype(o.template lift< SourceValue >(DistinctUntilChanged(std::forward< BinaryPredicate >(pred))))
Definition: rx-distinct_until_changed.hpp:127
Definition: rx-predef.hpp:177