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 
5 /*! \file rx-skip_last.hpp
6 
7     \brief Make new observable with skipped last count items from this observable.
8 
9     \tparam Count the type of the items counter.
10 
11     \param  t the number of last items to skip.
12 
13     \return  An observable that is identical to the source observable except that it does not emit the last t items that the source observable emits.
14 
15     \sample
16     \snippet skip_last.cpp skip_last sample
17     \snippet output.txt skip_last sample
18 */
19 
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     {
valuesrxcpp::operators::detail::skip_last::values52         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 
skip_lastrxcpp::operators::detail::skip_last62     skip_last(source_type s, count_type t)
63         : initial(std::move(s), std::move(t))
64     {
65     }
66 
67     template<class Subscriber>
on_subscriberxcpp::operators::detail::skip_last68     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](rxu::error_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 
119 /*! @copydoc rx-skip_last.hpp
120 */
121 template<class... AN>
skip_last(AN &&...an)122 auto skip_last(AN&&... an)
123     ->     operator_factory<skip_last_tag, AN...> {
124     return operator_factory<skip_last_tag, AN...>(std::make_tuple(std::forward<AN>(an)...));
125 }
126 
127 }
128 
129 template<>
130 struct member_overload<skip_last_tag>
131 {
132     template<class Observable, class Count,
133         class Enabled = rxu::enable_if_all_true_type_t<
134             is_observable<Observable>>,
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>>
memberrxcpp::member_overload139     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>
memberrxcpp::member_overload144     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
154