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
|
// 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_GROUP_BY_HPP)
#define RXCPP_OPERATORS_RX_GROUP_BY_HPP
#include "../rx-includes.hpp"
namespace rxcpp {
namespace operators {
namespace detail {
template<class T, class Selector>
struct is_group_by_selector_for {
typedef rxu::decay_t<Selector> selector_type;
typedef T source_value_type;
struct tag_not_valid {};
template<class CV, class CS>
static auto check(int) -> decltype((*(CS*)nullptr)(*(CV*)nullptr));
template<class CV, class CS>
static tag_not_valid check(...);
typedef decltype(check<source_value_type, selector_type>(0)) type;
static const bool value = !std::is_same<type, tag_not_valid>::value;
};
template<class T, class Observable, class KeySelector, class MarbleSelector, class BinaryPredicate>
struct group_by_traits
{
typedef T source_value_type;
typedef rxu::decay_t<Observable> source_type;
typedef rxu::decay_t<KeySelector> key_selector_type;
typedef rxu::decay_t<MarbleSelector> marble_selector_type;
typedef rxu::decay_t<BinaryPredicate> predicate_type;
static_assert(is_group_by_selector_for<source_value_type, key_selector_type>::value, "group_by KeySelector must be a function with the signature key_type(source_value_type)");
typedef typename is_group_by_selector_for<source_value_type, key_selector_type>::type key_type;
static_assert(is_group_by_selector_for<source_value_type, marble_selector_type>::value, "group_by MarbleSelector must be a function with the signature marble_type(source_value_type)");
typedef typename is_group_by_selector_for<source_value_type, marble_selector_type>::type marble_type;
typedef rxsub::subject<marble_type> subject_type;
typedef std::map<key_type, typename subject_type::subscriber_type, predicate_type> key_subscriber_map_type;
typedef grouped_observable<key_type, source_value_type> grouped_observable_type;
};
template<class T, class Observable, class KeySelector, class MarbleSelector, class BinaryPredicate>
struct group_by
{
typedef group_by_traits<T, Observable, KeySelector, MarbleSelector, BinaryPredicate> traits_type;
typedef typename traits_type::key_selector_type key_selector_type;
typedef typename traits_type::marble_selector_type marble_selector_type;
typedef typename traits_type::marble_type marble_type;
typedef typename traits_type::predicate_type predicate_type;
typedef typename traits_type::subject_type subject_type;
typedef typename traits_type::key_type key_type;
struct group_by_values
{
group_by_values(key_selector_type ks, marble_selector_type ms, predicate_type p)
: keySelector(std::move(ks))
, marbleSelector(std::move(ms))
, predicate(std::move(p))
{
}
mutable key_selector_type keySelector;
mutable marble_selector_type marbleSelector;
mutable predicate_type predicate;
};
group_by_values initial;
group_by(key_selector_type ks, marble_selector_type ms, predicate_type p)
: initial(std::move(ks), std::move(ms), std::move(p))
{
}
struct group_by_observable : public rxs::source_base<marble_type>
{
subject_type subject;
key_type key;
group_by_observable(subject_type s, key_type k)
: subject(std::move(s))
, key(k)
{
}
template<class Subscriber>
void on_subscribe(Subscriber&& o) const {
subject.get_observable().subscribe(std::forward<Subscriber>(o));
}
key_type on_get_key() {
return key;
}
};
template<class Subscriber>
struct group_by_observer : public group_by_values
{
typedef group_by_observer<Subscriber> this_type;
typedef typename traits_type::grouped_observable_type value_type;
typedef rxu::decay_t<Subscriber> dest_type;
typedef observer<T, this_type> observer_type;
dest_type dest;
mutable typename traits_type::key_subscriber_map_type groups;
group_by_observer(dest_type d, group_by_values v)
: group_by_values(v)
, dest(std::move(d))
, groups(group_by_values::predicate)
{
}
void on_next(T v) const {
auto selectedKey = on_exception(
[&](){
return this->keySelector(v);},
[this](std::exception_ptr e){on_error(e);});
if (selectedKey.empty()) {
return;
}
auto g = groups.find(selectedKey.get());
if (g == groups.end()) {
auto sub = subject_type();
g = groups.insert(std::make_pair(selectedKey.get(), sub.get_subscriber())).first;
dest.on_next(make_dynamic_grouped_observable<key_type, marble_type>(group_by_observable(sub, selectedKey.get())));
}
auto selectedMarble = on_exception(
[&](){
return this->marbleSelector(v);},
[this](std::exception_ptr e){on_error(e);});
if (selectedMarble.empty()) {
return;
}
g->second.on_next(std::move(selectedMarble.get()));
}
void on_error(std::exception_ptr e) const {
for(auto& g : groups) {
g.second.on_error(e);
}
dest.on_error(e);
}
void on_completed() const {
for(auto& g : groups) {
g.second.on_completed();
}
dest.on_completed();
}
static subscriber<T, observer_type> make(dest_type d, group_by_values v) {
auto cs = d.get_subscription();
return make_subscriber<T>(std::move(cs), observer_type(this_type(std::move(d), std::move(v))));
}
};
template<class Subscriber>
auto operator()(Subscriber dest) const
-> decltype(group_by_observer<Subscriber>::make(std::move(dest), initial)) {
return group_by_observer<Subscriber>::make(std::move(dest), initial);
}
};
template<class KeySelector, class MarbleSelector, class BinaryPredicate>
class group_by_factory
{
typedef rxu::decay_t<KeySelector> key_selector_type;
typedef rxu::decay_t<MarbleSelector> marble_selector_type;
typedef rxu::decay_t<BinaryPredicate> predicate_type;
key_selector_type keySelector;
marble_selector_type marbleSelector;
predicate_type predicate;
public:
group_by_factory(key_selector_type ks, marble_selector_type ms, predicate_type p)
: keySelector(std::move(ks))
, marbleSelector(std::move(ms))
, predicate(std::move(p))
{
}
template<class Observable>
struct group_by_factory_traits
{
typedef rxu::value_type_t<Observable> value_type;
typedef detail::group_by_traits<value_type, Observable, KeySelector, MarbleSelector, BinaryPredicate> traits_type;
typedef detail::group_by<value_type, Observable, KeySelector, MarbleSelector, BinaryPredicate> group_by_type;
};
template<class Observable>
auto operator()(Observable&& source)
-> decltype(source.template lift<typename group_by_factory_traits<Observable>::traits_type::grouped_observable_type>(typename group_by_factory_traits<Observable>::group_by_type(std::move(keySelector), std::move(marbleSelector), std::move(predicate)))) {
return source.template lift<typename group_by_factory_traits<Observable>::traits_type::grouped_observable_type>(typename group_by_factory_traits<Observable>::group_by_type(std::move(keySelector), std::move(marbleSelector), std::move(predicate)));
}
};
}
template<class KeySelector, class MarbleSelector, class BinaryPredicate>
inline auto group_by(KeySelector ks, MarbleSelector ms, BinaryPredicate p)
-> detail::group_by_factory<KeySelector, MarbleSelector, BinaryPredicate> {
return detail::group_by_factory<KeySelector, MarbleSelector, BinaryPredicate>(std::move(ks), std::move(ms), std::move(p));
}
}
}
#endif
|
