1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
|
// Copyright (c) Microsoft Open Technologies, Inc. All rights reserved. See License.txt in the project root for license information.
#pragma once
/*! \file rx-take.hpp
\brief For the first count items from this observable emit them from the new observable that is returned.
\tparam Count the type of the items counter.
\param t the number of items to take.
\return An observable that emits only the first t items emitted by the source Observable, or all of the items from the source observable if that observable emits fewer than t items.
\sample
\snippet take.cpp take sample
\snippet output.txt take sample
*/
#if !defined(RXCPP_OPERATORS_RX_TAKE_HPP)
#define RXCPP_OPERATORS_RX_TAKE_HPP
#include "../rx-includes.hpp"
namespace rxcpp {
namespace operators {
namespace detail {
template<class... AN>
struct take_invalid_arguments {};
template<class... AN>
struct take_invalid : public rxo::operator_base<take_invalid_arguments<AN...>> {
using type = observable<take_invalid_arguments<AN...>, take_invalid<AN...>>;
};
template<class... AN>
using take_invalid_t = typename take_invalid<AN...>::type;
template<class T, class Observable, class Count>
struct take : public operator_base<T>
{
typedef rxu::decay_t<Observable> source_type;
typedef rxu::decay_t<Count> count_type;
struct values
{
values(source_type s, count_type t)
: source(std::move(s))
, count(std::move(t))
{
}
source_type source;
count_type count;
};
values initial;
take(source_type s, count_type t)
: initial(std::move(s), std::move(t))
{
}
struct mode
{
enum type {
taking, // capture messages
triggered, // ignore messages
errored, // error occured
stopped // observable completed
};
};
template<class Subscriber>
void on_subscribe(const Subscriber& s) const {
typedef Subscriber output_type;
struct state_type
: public std::enable_shared_from_this<state_type>
, public values
{
state_type(const values& i, const output_type& oarg)
: values(i)
, mode_value(mode::taking)
, out(oarg)
{
}
typename mode::type mode_value;
output_type out;
};
// take a copy of the values for each subscription
auto state = std::make_shared<state_type>(initial, s);
composite_subscription source_lifetime;
s.add(source_lifetime);
state->source.subscribe(
// split subscription lifetime
source_lifetime,
// on_next
[state, source_lifetime](T t) {
if (state->mode_value < mode::triggered) {
if (--state->count > 0) {
state->out.on_next(t);
} else {
state->mode_value = mode::triggered;
state->out.on_next(t);
// must shutdown source before signaling completion
source_lifetime.unsubscribe();
state->out.on_completed();
}
}
},
// on_error
[state](rxu::error_ptr e) {
state->mode_value = mode::errored;
state->out.on_error(e);
},
// on_completed
[state]() {
state->mode_value = mode::stopped;
state->out.on_completed();
}
);
}
};
}
/*! @copydoc rx-take.hpp
*/
template<class... AN>
auto take(AN&&... an)
-> operator_factory<take_tag, AN...> {
return operator_factory<take_tag, AN...>(std::make_tuple(std::forward<AN>(an)...));
}
}
template<>
struct member_overload<take_tag>
{
template<class Observable,
class Count,
class Enabled = rxu::enable_if_all_true_type_t<
is_observable<Observable>>,
class SourceValue = rxu::value_type_t<Observable>,
class Take = rxo::detail::take<SourceValue, rxu::decay_t<Observable>, rxu::decay_t<Count>>,
class Value = rxu::value_type_t<Take>,
class Result = observable<Value, Take>>
static Result member(Observable&& o, Count&& c) {
return Result(Take(std::forward<Observable>(o), std::forward<Count>(c)));
}
template<class... AN>
static operators::detail::take_invalid_t<AN...> member(AN...) {
std::terminate();
return {};
static_assert(sizeof...(AN) == 10000, "take takes (optional Count)");
}
};
}
#endif
|
