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
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
|
// Copyright (c) Microsoft Open Technologies, Inc. All rights reserved. See License.txt in the project root for license information.
#pragma once
/*! \file rx-take_until.hpp
\brief For each item from this observable until on_next occurs on the trigger observable or until the specified time, emit them from the new observable that is returned.
take_until takes (TriggerObservable, optional Coordination) or (TimePoint, optional Coordination)
\tparam TriggerSource the type of the trigger observable.
\tparam TimePoint the type of the time interval.
\tparam Coordination the type of the scheduler (optional).
\param t an observable whose first emitted item will stop emitting items from the source observable.
\param when a time point when the returned observable will stop emitting items.
\param cn the scheduler to use for scheduling the items (optional).
\return An observable that emits the items emitted by the source observable until trigger observable emitted or the time runs out.
\sample
\snippet take_until.cpp take_until sample
\snippet output.txt take_until sample
\sample
\snippet take_until.cpp threaded take_until sample
\snippet output.txt threaded take_until sample
\sample
\snippet take_until.cpp take_until time sample
\snippet output.txt take_until time sample
\sample
\snippet take_until.cpp threaded take_until time sample
\snippet output.txt threaded take_until time sample
*/
#if !defined(RXCPP_OPERATORS_RX_TAKE_UNTIL_HPP)
#define RXCPP_OPERATORS_RX_TAKE_UNTIL_HPP
#include "../rx-includes.hpp"
namespace rxcpp {
namespace operators {
namespace detail {
template<class... AN>
struct take_until_invalid_arguments {};
template<class... AN>
struct take_until_invalid : public rxo::operator_base<take_until_invalid_arguments<AN...>> {
using type = observable<take_until_invalid_arguments<AN...>, take_until_invalid<AN...>>;
};
template<class... AN>
using take_until_invalid_t = typename take_until_invalid<AN...>::type;
template<class T, class Observable, class TriggerObservable, class Coordination>
struct take_until : public operator_base<T>
{
typedef rxu::decay_t<Observable> source_type;
typedef rxu::decay_t<TriggerObservable> trigger_source_type;
typedef rxu::decay_t<Coordination> coordination_type;
typedef typename coordination_type::coordinator_type coordinator_type;
struct values
{
values(source_type s, trigger_source_type t, coordination_type sf)
: source(std::move(s))
, trigger(std::move(t))
, coordination(std::move(sf))
{
}
source_type source;
trigger_source_type trigger;
coordination_type coordination;
};
values initial;
take_until(source_type s, trigger_source_type t, coordination_type sf)
: initial(std::move(s), std::move(t), std::move(sf))
{
}
struct mode
{
enum type {
taking, // no messages from trigger
clear, // trigger completed
triggered, // trigger sent on_next
errored, // error either on trigger or on observable
stopped // observable completed
};
};
template<class Subscriber>
void on_subscribe(Subscriber s) const {
typedef Subscriber output_type;
struct take_until_state_type
: public std::enable_shared_from_this<take_until_state_type>
, public values
{
take_until_state_type(const values& i, coordinator_type coor, const output_type& oarg)
: values(i)
, mode_value(mode::taking)
, coordinator(std::move(coor))
, out(oarg)
{
out.add(trigger_lifetime);
out.add(source_lifetime);
}
typename mode::type mode_value;
composite_subscription trigger_lifetime;
composite_subscription source_lifetime;
coordinator_type coordinator;
output_type out;
};
auto coordinator = initial.coordination.create_coordinator(s.get_subscription());
// take a copy of the values for each subscription
auto state = std::make_shared<take_until_state_type>(initial, std::move(coordinator), std::move(s));
auto trigger = on_exception(
[&](){return state->coordinator.in(state->trigger);},
state->out);
if (trigger.empty()) {
return;
}
auto source = on_exception(
[&](){return state->coordinator.in(state->source);},
state->out);
if (source.empty()) {
return;
}
auto sinkTrigger = make_subscriber<typename trigger_source_type::value_type>(
// share parts of subscription
state->out,
// new lifetime
state->trigger_lifetime,
// on_next
[state](const typename trigger_source_type::value_type&) {
if (state->mode_value != mode::taking) {return;}
state->mode_value = mode::triggered;
state->out.on_completed();
},
// on_error
[state](std::exception_ptr e) {
if (state->mode_value != mode::taking) {return;}
state->mode_value = mode::errored;
state->out.on_error(e);
},
// on_completed
[state]() {
if (state->mode_value != mode::taking) {return;}
state->mode_value = mode::clear;
}
);
auto selectedSinkTrigger = on_exception(
[&](){return state->coordinator.out(sinkTrigger);},
state->out);
if (selectedSinkTrigger.empty()) {
return;
}
trigger->subscribe(std::move(selectedSinkTrigger.get()));
auto sinkSource = make_subscriber<T>(
// split subscription lifetime
state->source_lifetime,
// on_next
[state](T t) {
//
// everything is crafted to minimize the overhead of this function.
//
if (state->mode_value < mode::triggered) {
state->out.on_next(t);
}
},
// on_error
[state](std::exception_ptr e) {
if (state->mode_value > mode::clear) {return;}
state->mode_value = mode::errored;
state->out.on_error(e);
},
// on_completed
[state]() {
if (state->mode_value > mode::clear) {return;}
state->mode_value = mode::stopped;
state->out.on_completed();
}
);
auto selectedSinkSource = on_exception(
[&](){return state->coordinator.out(sinkSource);},
state->out);
if (selectedSinkSource.empty()) {
return;
}
source->subscribe(std::move(selectedSinkSource.get()));
}
};
}
/*! @copydoc rx-take_until.hpp
*/
template<class... AN>
auto take_until(AN&&... an)
-> operator_factory<take_until_tag, AN...> {
return operator_factory<take_until_tag, AN...>(std::make_tuple(std::forward<AN>(an)...));
}
}
template<>
struct member_overload<take_until_tag>
{
template<class Observable, class TimePoint,
class Enabled = rxu::enable_if_all_true_type_t<
is_observable<Observable>,
std::is_convertible<TimePoint, rxsc::scheduler::clock_type::time_point>>,
class SourceValue = rxu::value_type_t<Observable>,
class Timer = typename rxu::defer_type<rxs::detail::timer, identity_one_worker>::type,
class TimerValue = rxu::value_type_t<Timer>,
class TriggerObservable = observable<TimerValue, Timer>,
class TakeUntil = rxo::detail::take_until<SourceValue, rxu::decay_t<Observable>, TriggerObservable, identity_one_worker>,
class Value = rxu::value_type_t<TakeUntil>,
class Result = observable<Value, TakeUntil>>
static Result member(Observable&& o, TimePoint&& when) {
auto cn = identity_current_thread();
return Result(TakeUntil(std::forward<Observable>(o), rxs::timer(std::forward<TimePoint>(when), cn), cn));
}
template<class Observable, class TimePoint, class Coordination,
class Enabled = rxu::enable_if_all_true_type_t<
is_observable<Observable>,
is_coordination<Coordination>,
std::is_convertible<TimePoint, rxsc::scheduler::clock_type::time_point>>,
class SourceValue = rxu::value_type_t<Observable>,
class Timer = typename rxu::defer_type<rxs::detail::timer, rxu::decay_t<Coordination>>::type,
class TimerValue = rxu::value_type_t<Timer>,
class TriggerObservable = observable<TimerValue, Timer>,
class TakeUntil = rxo::detail::take_until<SourceValue, rxu::decay_t<Observable>, TriggerObservable, rxu::decay_t<Coordination>>,
class Value = rxu::value_type_t<TakeUntil>,
class Result = observable<Value, TakeUntil>>
static Result member(Observable&& o, TimePoint&& when, Coordination cn) {
return Result(TakeUntil(std::forward<Observable>(o), rxs::timer(std::forward<TimePoint>(when), cn), cn));
}
template<class Observable, class TriggerObservable,
class Enabled = rxu::enable_if_all_true_type_t<
all_observables<Observable, TriggerObservable>>,
class SourceValue = rxu::value_type_t<Observable>,
class TakeUntil = rxo::detail::take_until<SourceValue, rxu::decay_t<Observable>, rxu::decay_t<TriggerObservable>, identity_one_worker>,
class Value = rxu::value_type_t<TakeUntil>,
class Result = observable<Value, TakeUntil>>
static Result member(Observable&& o, TriggerObservable&& t) {
return Result(TakeUntil(std::forward<Observable>(o), std::forward<TriggerObservable>(t), identity_current_thread()));
}
template<class Observable, class TriggerObservable, class Coordination,
class Enabled = rxu::enable_if_all_true_type_t<
all_observables<Observable, TriggerObservable>,
is_coordination<Coordination>>,
class SourceValue = rxu::value_type_t<Observable>,
class TakeUntil = rxo::detail::take_until<SourceValue, rxu::decay_t<Observable>, rxu::decay_t<TriggerObservable>, rxu::decay_t<Coordination>>,
class Value = rxu::value_type_t<TakeUntil>,
class Result = observable<Value, TakeUntil>>
static Result member(Observable&& o, TriggerObservable&& t, Coordination&& cn) {
return Result(TakeUntil(std::forward<Observable>(o), std::forward<TriggerObservable>(t), std::forward<Coordination>(cn)));
}
template<class... AN>
static operators::detail::take_until_invalid_t<AN...> member(AN...) {
std::terminate();
return {};
static_assert(sizeof...(AN) == 10000, "take_until takes (TriggerObservable, optional Coordination) or (TimePoint, optional Coordination)");
}
};
}
#endif
|
