18 #if !defined(RXCPP_OPERATORS_RX_ELEMENT_AT_HPP)
19 #define RXCPP_OPERATORS_RX_ELEMENT_AT_HPP
21 #include "../rx-includes.hpp"
30 struct element_at_invalid_arguments {};
33 struct element_at_invalid :
public rxo::operator_base<element_at_invalid_arguments<AN...>> {
34 using type = observable<element_at_invalid_arguments<
AN...>, element_at_invalid<
AN...>>;
37 using element_at_invalid_t =
typename element_at_invalid<
AN...>::type;
41 typedef rxu::decay_t<T> source_value_type;
43 struct element_at_values {
44 element_at_values(
int i)
51 element_at_values initial;
58 template<
class Subscriber>
59 struct element_at_observer :
public element_at_values
61 typedef element_at_observer<Subscriber> this_type;
62 typedef source_value_type value_type;
63 typedef rxu::decay_t<Subscriber> dest_type;
64 typedef observer<value_type, this_type> observer_type;
68 element_at_observer(dest_type d, element_at_values v)
69 : element_at_values(v),
74 void on_next(source_value_type v)
const {
75 if (current++ == this->index) {
83 void on_completed()
const {
84 if(current <= this->index) {
89 static subscriber<value_type, observer_type> make(dest_type d, element_at_values v) {
90 return make_subscriber<value_type>(d, this_type(d, v));
94 template<
class Subscriber>
95 auto operator()(Subscriber dest)
const
96 -> decltype(element_at_observer<Subscriber>::make(std::move(dest), initial)) {
97 return element_at_observer<Subscriber>::make(std::move(dest), initial);
105 template<
class...
AN>
116 template<
class Observable,
121 class element_at = rxo::detail::element_at<SourceValue>>
122 static auto member(Observable&& o,
int index)
123 -> decltype(o.template lift<SourceValue>(
element_at(index))) {
124 return o.template lift<SourceValue>(
element_at(index));
127 template<
class...
AN>
128 static operators::detail::element_at_invalid_t<
AN...>
member(
const AN...) {
131 static_assert(
sizeof...(
AN) == 10000,
"element_at takes (required int)");