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
|
// 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_UNTIL_HPP)
#define RXCPP_OPERATORS_RX_TAKE_UNTIL_HPP
#include "../rx-includes.hpp"
namespace rxcpp {
namespace operators {
namespace detail {
template<class T, class Observable, class TriggerObservable, class Coordination>
struct take_until : public operator_base<T>
{
typedef typename std::decay<Observable>::type source_type;
typedef typename std::decay<TriggerObservable>::type trigger_source_type;
typedef typename std::decay<Coordination>::type 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::shared_ptr<take_until_state_type>(new 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()));
}
};
template<class TriggerObservable, class Coordination>
class take_until_factory
{
typedef typename std::decay<TriggerObservable>::type trigger_source_type;
typedef typename std::decay<Coordination>::type coordination_type;
trigger_source_type trigger_source;
coordination_type coordination;
public:
take_until_factory(trigger_source_type t, coordination_type sf)
: trigger_source(std::move(t))
, coordination(std::move(sf))
{
}
template<class Observable>
auto operator()(Observable&& source)
-> observable<typename std::decay<Observable>::type::value_type, take_until<typename std::decay<Observable>::type::value_type, Observable, trigger_source_type, Coordination>> {
return observable<typename std::decay<Observable>::type::value_type, take_until<typename std::decay<Observable>::type::value_type, Observable, trigger_source_type, Coordination>>(
take_until<typename std::decay<Observable>::type::value_type, Observable, trigger_source_type, Coordination>(std::forward<Observable>(source), trigger_source, coordination));
}
};
}
template<class TriggerObservable, class Coordination>
auto take_until(TriggerObservable t, Coordination sf)
-> detail::take_until_factory<TriggerObservable, Coordination> {
return detail::take_until_factory<TriggerObservable, Coordination>(std::move(t), std::move(sf));
}
}
}
#endif
|
