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
|
// Copyright (c) Microsoft Open Technologies, Inc. All rights reserved. See License.txt in the project root for license information.
#pragma once
#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 T, class Observable, class Count>
struct take : public operator_base<T>
{
typedef typename std::decay<Observable>::type source_type;
typedef typename std::decay<Count>::type 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::shared_ptr<state_type>(new 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](std::exception_ptr e) {
state->mode_value = mode::errored;
state->out.on_error(e);
},
// on_completed
[state]() {
state->mode_value = mode::stopped;
state->out.on_completed();
}
);
}
};
template<class T>
class take_factory
{
typedef typename std::decay<T>::type count_type;
count_type count;
public:
take_factory(count_type t) : count(std::move(t)) {}
template<class Observable>
auto operator()(Observable&& source)
-> observable<typename std::decay<Observable>::type::value_type, take<typename std::decay<Observable>::type::value_type, Observable, count_type>> {
return observable<typename std::decay<Observable>::type::value_type, take<typename std::decay<Observable>::type::value_type, Observable, count_type>>(
take<typename std::decay<Observable>::type::value_type, Observable, count_type>(std::forward<Observable>(source), count));
}
};
}
template<class T>
auto take(T&& t)
-> detail::take_factory<T> {
return detail::take_factory<T>(std::forward<T>(t));
}
}
}
#endif
|
