RxCpp
The Reactive Extensions for Native (RxCpp) is a library for composing asynchronous and event-based programs using observable sequences and LINQ-style query operators in both C and C++.
rx-skip_last.hpp
Go to the documentation of this file.
1 // Copyright (c) Microsoft Open Technologies, Inc. All rights reserved. See License.txt in the project root for license information.
2 
3 #pragma once
4 
20 #if !defined(RXCPP_OPERATORS_RX_SKIP_LAST_HPP)
21 #define RXCPP_OPERATORS_RX_SKIP_LAST_HPP
22 
23 #include "../rx-includes.hpp"
24 
25 namespace rxcpp {
26 
27 namespace operators {
28 
29 namespace detail {
30 
31 template<class... AN>
32 struct skip_last_invalid_arguments {};
33 
34 template<class... AN>
35 struct skip_last_invalid : public rxo::operator_base<skip_last_invalid_arguments<AN...>> {
36  using type = observable<skip_last_invalid_arguments<AN...>, skip_last_invalid<AN...>>;
37 };
38 template<class... AN>
39 using skip_last_invalid_t = typename skip_last_invalid<AN...>::type;
40 
41 template<class T, class Observable, class Count>
42 struct skip_last : public operator_base<T>
43 {
44  typedef rxu::decay_t<Observable> source_type;
45  typedef rxu::decay_t<Count> count_type;
46 
47  typedef std::queue<T> queue_type;
48  typedef typename queue_type::size_type queue_size_type;
49 
50  struct values
51  {
52  values(source_type s, count_type t)
53  : source(std::move(s))
54  , count(static_cast<queue_size_type>(t))
55  {
56  }
57  source_type source;
58  queue_size_type count;
59  };
60  values initial;
61 
62  skip_last(source_type s, count_type t)
63  : initial(std::move(s), std::move(t))
64  {
65  }
66 
67  template<class Subscriber>
68  void on_subscribe(const Subscriber& s) const {
69 
70  typedef Subscriber output_type;
71  struct state_type
72  : public std::enable_shared_from_this<state_type>
73  , public values
74  {
75  state_type(const values& i, const output_type& oarg)
76  : values(i)
77  , out(oarg)
78  {
79  }
80  queue_type items;
81  output_type out;
82  };
83  // take a copy of the values for each subscription
84  auto state = std::make_shared<state_type>(initial, s);
85 
86  composite_subscription source_lifetime;
87 
88  s.add(source_lifetime);
89 
90  state->source.subscribe(
91  // split subscription lifetime
92  source_lifetime,
93  // on_next
94  [state](T t) {
95  if(state->count > 0) {
96  if (state->items.size() == state->count) {
97  state->out.on_next(std::move(state->items.front()));
98  state->items.pop();
99  }
100  state->items.push(t);
101  } else {
102  state->out.on_next(t);
103  }
104  },
105  // on_error
106  [state](std::exception_ptr e) {
107  state->out.on_error(e);
108  },
109  // on_completed
110  [state]() {
111  state->out.on_completed();
112  }
113  );
114  }
115 };
116 
117 }
118 
121 template<class... AN>
122 auto skip_last(AN&&... an)
124  return operator_factory<skip_last_tag, AN...>(std::make_tuple(std::forward<AN>(an)...));
125 }
126 
127 }
128 
129 template<>
131 {
132  template<class Observable, class Count,
133  class Enabled = rxu::enable_if_all_true_type_t<
135  class SourceValue = rxu::value_type_t<Observable>,
136  class SkipLast = rxo::detail::skip_last<SourceValue, rxu::decay_t<Observable>, rxu::decay_t<Count>>,
137  class Value = rxu::value_type_t<SkipLast>,
138  class Result = observable<Value, SkipLast>>
139  static Result member(Observable&& o, Count&& c) {
140  return Result(SkipLast(std::forward<Observable>(o), std::forward<Count>(c)));
141  }
142 
143  template<class... AN>
144  static operators::detail::skip_last_invalid_t<AN...> member(AN...) {
145  std::terminate();
146  return {};
147  static_assert(sizeof...(AN) == 10000, "skip_last takes (Count)");
148  }
149 };
150 
151 }
152 
153 #endif
Definition: rx-operators.hpp:374
auto count() -> operator_factory< reduce_tag, int, rxu::count, rxu::detail::take_at< 0 >>
For each item from this observable reduce it by incrementing a count.
Definition: rx-reduce.hpp:412
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 AN
Definition: rx-finally.hpp:105
typename std::decay< T >::type decay_t
Definition: rx-util.hpp:36
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
a source of values. subscribe or use one of the operator methods that return a new observable...
Definition: rx-observable.hpp:510
static Result member(Observable &&o, Count &&c)
Definition: rx-skip_last.hpp:139
static operators::detail::skip_last_invalid_t< AN... > member(AN...)
Definition: rx-skip_last.hpp:144
auto skip_last(AN &&...an) -> operator_factory< skip_last_tag, AN... >
Make new observable with skipped last count items from this observable.
Definition: rx-skip_last.hpp:122
Definition: rx-predef.hpp:177