line |
stmt |
bran |
cond |
sub |
pod |
time |
code |
1
|
|
|
|
|
|
|
package RxPerl::Operators::Pipeable; |
2
|
5
|
|
|
5
|
|
36
|
use strict; |
|
5
|
|
|
|
|
17
|
|
|
5
|
|
|
|
|
147
|
|
3
|
5
|
|
|
5
|
|
27
|
use warnings; |
|
5
|
|
|
|
|
11
|
|
|
5
|
|
|
|
|
217
|
|
4
|
|
|
|
|
|
|
|
5
|
5
|
|
|
|
|
528
|
use RxPerl::Operators::Creation qw/ |
6
|
|
|
|
|
|
|
rx_observable rx_subject rx_concat rx_of rx_interval rx_combine_latest rx_concat |
7
|
|
|
|
|
|
|
rx_throw_error rx_zip rx_merge rx_on_error_resume_next rx_race rx_timer |
8
|
|
|
|
|
|
|
rx_behavior_subject |
9
|
5
|
|
|
5
|
|
36
|
/; |
|
5
|
|
|
|
|
10
|
|
10
|
5
|
|
|
5
|
|
2030
|
use RxPerl::ConnectableObservable; |
|
5
|
|
|
|
|
11
|
|
|
5
|
|
|
|
|
143
|
|
11
|
5
|
|
|
5
|
|
33
|
use RxPerl::Utils qw/ get_timer_subs /; |
|
5
|
|
|
|
|
9
|
|
|
5
|
|
|
|
|
231
|
|
12
|
5
|
|
|
5
|
|
30
|
use RxPerl::Subscription; |
|
5
|
|
|
|
|
17
|
|
|
5
|
|
|
|
|
95
|
|
13
|
|
|
|
|
|
|
|
14
|
5
|
|
|
5
|
|
24
|
use Carp 'croak'; |
|
5
|
|
|
|
|
10
|
|
|
5
|
|
|
|
|
216
|
|
15
|
5
|
|
|
5
|
|
29
|
use Scalar::Util 'reftype', 'refaddr', 'blessed', 'weaken'; |
|
5
|
|
|
|
|
9
|
|
|
5
|
|
|
|
|
220
|
|
16
|
5
|
|
|
5
|
|
25
|
use Time::HiRes (); |
|
5
|
|
|
|
|
11
|
|
|
5
|
|
|
|
|
105
|
|
17
|
|
|
|
|
|
|
|
18
|
5
|
|
|
5
|
|
23
|
use Exporter 'import'; |
|
5
|
|
|
|
|
10
|
|
|
5
|
|
|
|
|
69587
|
|
19
|
|
|
|
|
|
|
our @EXPORT_OK = qw/ |
20
|
|
|
|
|
|
|
op_audit op_audit_time op_buffer op_buffer_count op_buffer_time op_catch_error op_combine_latest_with op_concat_all |
21
|
|
|
|
|
|
|
op_concat_map op_concat_with op_count op_debounce op_debounce_time op_default_if_empty op_delay op_delay_when |
22
|
|
|
|
|
|
|
op_distinct op_distinct_until_changed op_distinct_until_key_changed op_element_at op_end_with op_every |
23
|
|
|
|
|
|
|
op_exhaust_all op_exhaust_map op_filter op_finalize op_find op_find_index op_first op_group_by op_ignore_elements |
24
|
|
|
|
|
|
|
op_is_empty op_last op_map op_map_to op_max op_merge_all op_merge_map op_merge_with op_min op_multicast |
25
|
|
|
|
|
|
|
op_on_error_resume_next_with op_pairwise op_pluck op_race_with op_reduce op_ref_count op_repeat op_retry op_sample |
26
|
|
|
|
|
|
|
op_sample_time op_scan op_share op_single op_skip op_skip_last op_skip_until op_skip_while op_start_with |
27
|
|
|
|
|
|
|
op_switch_all op_switch_map op_take op_take_last op_take_until op_take_while op_tap op_throttle op_throttle_time |
28
|
|
|
|
|
|
|
op_throw_if_empty op_time_interval op_timeout op_timestamp op_to_array op_with_latest_from op_zip_with |
29
|
|
|
|
|
|
|
/; |
30
|
|
|
|
|
|
|
our %EXPORT_TAGS = (all => \@EXPORT_OK); |
31
|
|
|
|
|
|
|
|
32
|
|
|
|
|
|
|
our $VERSION = "v6.27.0"; |
33
|
|
|
|
|
|
|
|
34
|
|
|
|
|
|
|
sub op_audit { |
35
|
3
|
|
|
3
|
|
8
|
my ($duration_selector) = @_; |
36
|
|
|
|
|
|
|
|
37
|
|
|
|
|
|
|
return sub { |
38
|
3
|
|
|
3
|
|
7
|
my ($source) = @_; |
39
|
|
|
|
|
|
|
|
40
|
|
|
|
|
|
|
return rx_observable->new(sub { |
41
|
3
|
|
|
|
|
6
|
my ($subscriber) = @_; |
42
|
|
|
|
|
|
|
|
43
|
3
|
|
|
|
|
9
|
my $last_val; |
44
|
|
|
|
|
|
|
my $mini_subscription; |
45
|
3
|
|
|
|
|
0
|
my $main_is_complete; |
46
|
|
|
|
|
|
|
|
47
|
|
|
|
|
|
|
my $mini_subscriber = { |
48
|
|
|
|
|
|
|
next => sub { |
49
|
5
|
100
|
|
|
|
18
|
$subscriber->{next}->($last_val) if defined $subscriber->{next}; |
50
|
|
|
|
|
|
|
}, |
51
|
|
|
|
|
|
|
error => sub { |
52
|
1
|
50
|
|
|
|
6
|
$subscriber->{error}->(@_) if defined $subscriber->{error}; |
53
|
|
|
|
|
|
|
}, |
54
|
|
|
|
|
|
|
complete => sub { |
55
|
5
|
|
|
|
|
12
|
undef $mini_subscription; |
56
|
5
|
|
|
|
|
10
|
undef $last_val; |
57
|
5
|
100
|
|
|
|
14
|
if ($main_is_complete) { |
58
|
1
|
50
|
|
|
|
6
|
$subscriber->{complete}->() if defined $subscriber->{complete}; |
59
|
|
|
|
|
|
|
} |
60
|
|
|
|
|
|
|
}, |
61
|
3
|
|
|
|
|
17
|
}; |
62
|
|
|
|
|
|
|
|
63
|
|
|
|
|
|
|
my $own_subscriber = { |
64
|
|
|
|
|
|
|
next => sub { |
65
|
11
|
|
|
|
|
24
|
my ($v) = @_; |
66
|
|
|
|
|
|
|
|
67
|
11
|
|
|
|
|
17
|
$last_val = $v; |
68
|
11
|
100
|
|
|
|
27
|
if (!defined $mini_subscription) { |
69
|
6
|
|
|
|
|
11
|
my $o = do { local $_ = $v; $duration_selector->($v) }; |
|
6
|
|
|
|
|
9
|
|
|
6
|
|
|
|
|
17
|
|
70
|
6
|
|
|
|
|
16
|
$mini_subscription = $o->pipe( |
71
|
|
|
|
|
|
|
op_take(1), |
72
|
|
|
|
|
|
|
)->subscribe($mini_subscriber); |
73
|
|
|
|
|
|
|
} |
74
|
|
|
|
|
|
|
}, |
75
|
|
|
|
|
|
|
error => sub { |
76
|
1
|
50
|
|
|
|
5
|
$subscriber->{error}->(@_) if defined $subscriber->{error}; |
77
|
|
|
|
|
|
|
}, |
78
|
|
|
|
|
|
|
complete => sub { |
79
|
1
|
|
|
|
|
3
|
$main_is_complete = 1; |
80
|
1
|
50
|
|
|
|
6
|
if (! defined $mini_subscription) { |
81
|
0
|
0
|
|
|
|
0
|
$subscriber->{complete}->() if defined $subscriber->{complete}; |
82
|
|
|
|
|
|
|
} |
83
|
|
|
|
|
|
|
} |
84
|
3
|
|
|
|
|
17
|
}; |
85
|
|
|
|
|
|
|
|
86
|
3
|
|
|
|
|
11
|
my $own_subscription = $source->subscribe($own_subscriber); |
87
|
|
|
|
|
|
|
|
88
|
3
|
|
|
|
|
10
|
return $mini_subscription, $own_subscription; |
89
|
3
|
|
|
|
|
8
|
}); |
90
|
3
|
|
|
|
|
15
|
}; |
91
|
|
|
|
|
|
|
} |
92
|
|
|
|
|
|
|
|
93
|
|
|
|
|
|
|
sub op_audit_time { |
94
|
0
|
|
|
0
|
|
0
|
my ($duration) = @_; |
95
|
|
|
|
|
|
|
|
96
|
0
|
|
|
0
|
|
0
|
return op_audit(sub { rx_timer($duration) }); |
|
0
|
|
|
|
|
0
|
|
97
|
|
|
|
|
|
|
} |
98
|
|
|
|
|
|
|
|
99
|
|
|
|
|
|
|
sub op_buffer { |
100
|
2
|
|
|
2
|
|
31
|
my ($notifier) = @_; |
101
|
|
|
|
|
|
|
|
102
|
|
|
|
|
|
|
return sub { |
103
|
2
|
|
|
2
|
|
4
|
my ($source) = @_; |
104
|
|
|
|
|
|
|
|
105
|
|
|
|
|
|
|
return rx_observable->new(sub { |
106
|
2
|
|
|
|
|
5
|
my ($subscriber) = @_; |
107
|
|
|
|
|
|
|
|
108
|
2
|
|
|
|
|
3
|
my @buffer; |
109
|
|
|
|
|
|
|
|
110
|
|
|
|
|
|
|
my $own_subscriber = { |
111
|
|
|
|
|
|
|
%$subscriber, |
112
|
|
|
|
|
|
|
next => sub { |
113
|
16
|
|
|
|
|
37
|
push @buffer, $_[0]; |
114
|
|
|
|
|
|
|
}, |
115
|
|
|
|
|
|
|
error => sub { |
116
|
0
|
0
|
|
|
|
0
|
$subscriber->{error}->($_[0]) if defined $subscriber->{error}; |
117
|
|
|
|
|
|
|
}, |
118
|
|
|
|
|
|
|
complete => sub { |
119
|
2
|
50
|
33
|
|
|
47
|
$subscriber->{next}->([@buffer]) if @buffer and defined $subscriber->{next}; |
120
|
2
|
|
|
|
|
5
|
undef @buffer; |
121
|
2
|
50
|
|
|
|
10
|
$subscriber->{complete}->() if defined $subscriber->{complete}; |
122
|
|
|
|
|
|
|
}, |
123
|
2
|
|
|
|
|
17
|
}; |
124
|
|
|
|
|
|
|
|
125
|
|
|
|
|
|
|
my $notifier_subscriber = { |
126
|
|
|
|
|
|
|
next => sub { |
127
|
5
|
50
|
|
|
|
24
|
$subscriber->{next}->([@buffer]) if defined $subscriber->{next}; |
128
|
5
|
|
|
|
|
15
|
undef @buffer; |
129
|
|
|
|
|
|
|
}, |
130
|
|
|
|
|
|
|
error => sub { |
131
|
0
|
0
|
|
|
|
0
|
$subscriber->{error}->($_[0]) if defined $subscriber->{error}; |
132
|
|
|
|
|
|
|
}, |
133
|
2
|
|
|
|
|
13
|
}; |
134
|
|
|
|
|
|
|
|
135
|
2
|
|
|
|
|
6
|
my $s1 = $source->subscribe($own_subscriber); |
136
|
2
|
|
|
|
|
7
|
my $s2 = $notifier->subscribe($notifier_subscriber); |
137
|
|
|
|
|
|
|
|
138
|
2
|
|
|
|
|
17
|
return [$s1, $s2], sub { undef @buffer }; |
|
2
|
|
|
|
|
5
|
|
139
|
|
|
|
|
|
|
}) |
140
|
2
|
|
|
|
|
5
|
} |
141
|
2
|
|
|
|
|
14
|
} |
142
|
|
|
|
|
|
|
|
143
|
|
|
|
|
|
|
sub op_buffer_count { |
144
|
3
|
|
|
3
|
|
8
|
my ($buffer_size, $start_buffer_every) = @_; |
145
|
|
|
|
|
|
|
|
146
|
3
|
|
66
|
|
|
12
|
$start_buffer_every //= $buffer_size; |
147
|
|
|
|
|
|
|
|
148
|
|
|
|
|
|
|
return sub { |
149
|
3
|
|
|
3
|
|
8
|
my ($source) = @_; |
150
|
|
|
|
|
|
|
|
151
|
|
|
|
|
|
|
return rx_observable->new(sub { |
152
|
3
|
|
|
|
|
8
|
my ($subscriber) = @_; |
153
|
|
|
|
|
|
|
|
154
|
3
|
|
|
|
|
8
|
my @buffers; |
155
|
3
|
|
|
|
|
4
|
my $count = 0; |
156
|
|
|
|
|
|
|
my $own_subscriber = { |
157
|
|
|
|
|
|
|
%$subscriber, |
158
|
|
|
|
|
|
|
next => sub { |
159
|
18
|
|
|
|
|
37
|
my ($value) = @_; |
160
|
|
|
|
|
|
|
|
161
|
18
|
100
|
|
|
|
53
|
if ($count++ % $start_buffer_every == 0) { |
162
|
13
|
|
|
|
|
26
|
push @buffers, []; |
163
|
|
|
|
|
|
|
} |
164
|
|
|
|
|
|
|
|
165
|
18
|
|
|
|
|
43
|
for (my $i = 0; $i < @buffers; $i++) { |
166
|
25
|
|
|
|
|
42
|
my $buffer = $buffers[$i]; |
167
|
|
|
|
|
|
|
|
168
|
25
|
|
|
|
|
43
|
push @$buffer, $value; |
169
|
|
|
|
|
|
|
|
170
|
25
|
100
|
|
|
|
66
|
if (@$buffer == $buffer_size) { |
171
|
10
|
50
|
|
|
|
49
|
$subscriber->{next}->($buffer) if defined $subscriber->{next}; |
172
|
10
|
|
|
|
|
20
|
splice @buffers, $i, 1; |
173
|
10
|
|
|
|
|
15
|
$i--; |
174
|
10
|
|
|
|
|
27
|
next; |
175
|
|
|
|
|
|
|
} |
176
|
|
|
|
|
|
|
} |
177
|
|
|
|
|
|
|
}, |
178
|
|
|
|
|
|
|
complete => sub { |
179
|
3
|
50
|
|
|
|
18
|
if (defined $subscriber->{next}) { |
180
|
3
|
|
|
|
|
14
|
$subscriber->{next}->($_) foreach @buffers; |
181
|
|
|
|
|
|
|
} |
182
|
|
|
|
|
|
|
|
183
|
3
|
50
|
|
|
|
13
|
$subscriber->{complete}->() if defined $subscriber->{complete}; |
184
|
|
|
|
|
|
|
}, |
185
|
3
|
|
|
|
|
24
|
}; |
186
|
|
|
|
|
|
|
|
187
|
3
|
|
|
|
|
12
|
$source->subscribe($own_subscriber); |
188
|
|
|
|
|
|
|
|
189
|
3
|
|
|
|
|
28
|
return; |
190
|
3
|
|
|
|
|
8
|
}); |
191
|
3
|
|
|
|
|
20
|
}; |
192
|
|
|
|
|
|
|
} |
193
|
|
|
|
|
|
|
|
194
|
|
|
|
|
|
|
sub op_buffer_time { |
195
|
1
|
|
|
1
|
|
15
|
my ($buffer_time_span) = @_; |
196
|
|
|
|
|
|
|
|
197
|
1
|
|
|
|
|
8
|
return op_buffer(rx_interval($buffer_time_span)); |
198
|
|
|
|
|
|
|
} |
199
|
|
|
|
|
|
|
|
200
|
|
|
|
|
|
|
sub _op_catch_error_helper { |
201
|
0
|
|
|
0
|
|
0
|
my ($source, $selector, $subscriber, $dependents, $error) = @_; |
202
|
|
|
|
|
|
|
|
203
|
0
|
|
|
|
|
0
|
my $new_observable; |
204
|
0
|
0
|
|
|
|
0
|
if (@_ == 4) { |
205
|
0
|
|
|
|
|
0
|
$new_observable = $source; |
206
|
|
|
|
|
|
|
} else { |
207
|
0
|
|
|
|
|
0
|
eval { $new_observable = $selector->($error, $source) }; |
|
0
|
|
|
|
|
0
|
|
208
|
0
|
0
|
|
|
|
0
|
if (my $e = $@) { |
209
|
0
|
0
|
|
|
|
0
|
$subscriber->{error}->($e) if defined $subscriber->{error}; |
210
|
0
|
|
|
|
|
0
|
return; |
211
|
|
|
|
|
|
|
} |
212
|
|
|
|
|
|
|
} |
213
|
|
|
|
|
|
|
|
214
|
0
|
|
|
|
|
0
|
my $own_subscription = RxPerl::Subscription->new; |
215
|
0
|
|
|
|
|
0
|
@$dependents = ($own_subscription); |
216
|
|
|
|
|
|
|
my $own_subscriber = { |
217
|
|
|
|
|
|
|
new_subscription => $own_subscription, |
218
|
|
|
|
|
|
|
next => $subscriber->{next}, |
219
|
|
|
|
|
|
|
error => sub { |
220
|
0
|
|
|
0
|
|
0
|
_op_catch_error_helper($source, $selector, $subscriber, $dependents, $_[0]); |
221
|
|
|
|
|
|
|
}, |
222
|
|
|
|
|
|
|
complete => $subscriber->{complete}, |
223
|
0
|
|
|
|
|
0
|
}; |
224
|
|
|
|
|
|
|
|
225
|
0
|
|
|
|
|
0
|
$new_observable->subscribe($own_subscriber); |
226
|
|
|
|
|
|
|
} |
227
|
|
|
|
|
|
|
|
228
|
|
|
|
|
|
|
sub op_catch_error { |
229
|
0
|
|
|
0
|
|
0
|
my ($selector) = @_; |
230
|
|
|
|
|
|
|
|
231
|
|
|
|
|
|
|
return sub { |
232
|
0
|
|
|
0
|
|
0
|
my ($source) = @_; |
233
|
|
|
|
|
|
|
|
234
|
|
|
|
|
|
|
return rx_observable->new(sub { |
235
|
0
|
|
|
|
|
0
|
my ($subscriber) = @_; |
236
|
|
|
|
|
|
|
|
237
|
0
|
|
|
|
|
0
|
my $dependents = []; |
238
|
0
|
|
|
|
|
0
|
$subscriber->subscription->add($dependents); |
239
|
|
|
|
|
|
|
|
240
|
0
|
|
|
|
|
0
|
_op_catch_error_helper( |
241
|
|
|
|
|
|
|
$source, $selector, $subscriber, $dependents, |
242
|
|
|
|
|
|
|
); |
243
|
|
|
|
|
|
|
|
244
|
0
|
|
|
|
|
0
|
return; |
245
|
0
|
|
|
|
|
0
|
}); |
246
|
0
|
|
|
|
|
0
|
}; |
247
|
|
|
|
|
|
|
} |
248
|
|
|
|
|
|
|
|
249
|
|
|
|
|
|
|
sub op_combine_latest_with { |
250
|
1
|
|
|
1
|
|
21
|
my (@other_observables) = @_; |
251
|
|
|
|
|
|
|
|
252
|
|
|
|
|
|
|
return sub { |
253
|
1
|
|
|
1
|
|
3
|
my $source = shift; |
254
|
|
|
|
|
|
|
|
255
|
1
|
|
|
|
|
4
|
return rx_combine_latest([$source, @other_observables]); |
256
|
|
|
|
|
|
|
} |
257
|
1
|
|
|
|
|
10
|
} |
258
|
|
|
|
|
|
|
|
259
|
|
|
|
|
|
|
sub op_concat_all { |
260
|
1
|
|
|
1
|
|
4
|
return op_merge_all(1); |
261
|
|
|
|
|
|
|
} |
262
|
|
|
|
|
|
|
|
263
|
|
|
|
|
|
|
sub op_concat_map { |
264
|
0
|
|
|
0
|
|
0
|
my ($observable_factory) = @_; |
265
|
|
|
|
|
|
|
|
266
|
|
|
|
|
|
|
return sub { |
267
|
0
|
|
|
0
|
|
0
|
my ($source) = @_; |
268
|
|
|
|
|
|
|
|
269
|
0
|
|
|
|
|
0
|
return $source->pipe( |
270
|
|
|
|
|
|
|
op_map($observable_factory), |
271
|
|
|
|
|
|
|
op_concat_all(), |
272
|
|
|
|
|
|
|
); |
273
|
0
|
|
|
|
|
0
|
}; |
274
|
|
|
|
|
|
|
} |
275
|
|
|
|
|
|
|
|
276
|
|
|
|
|
|
|
sub op_concat_with { |
277
|
3
|
|
|
3
|
|
8
|
my @other_observables = @_; |
278
|
|
|
|
|
|
|
|
279
|
|
|
|
|
|
|
return sub { |
280
|
3
|
|
|
3
|
|
10
|
my ($source) = @_; |
281
|
|
|
|
|
|
|
|
282
|
3
|
|
|
|
|
25
|
return rx_concat( |
283
|
|
|
|
|
|
|
$source, |
284
|
|
|
|
|
|
|
@other_observables, |
285
|
|
|
|
|
|
|
); |
286
|
3
|
|
|
|
|
21
|
}; |
287
|
|
|
|
|
|
|
} |
288
|
|
|
|
|
|
|
|
289
|
|
|
|
|
|
|
sub op_count { |
290
|
3
|
|
|
3
|
|
7
|
my ($predicate) = @_; |
291
|
|
|
|
|
|
|
|
292
|
|
|
|
|
|
|
return sub { |
293
|
3
|
|
|
3
|
|
9
|
my ($source) = @_; |
294
|
|
|
|
|
|
|
|
295
|
|
|
|
|
|
|
return rx_observable->new(sub { |
296
|
3
|
|
|
|
|
6
|
my ($subscriber) = @_; |
297
|
|
|
|
|
|
|
|
298
|
3
|
|
|
|
|
7
|
my $count = 0; |
299
|
3
|
|
|
|
|
4
|
my $idx = 0; |
300
|
|
|
|
|
|
|
|
301
|
|
|
|
|
|
|
my $own_subscriber = { |
302
|
|
|
|
|
|
|
%$subscriber, |
303
|
|
|
|
|
|
|
next => sub { |
304
|
14
|
|
|
|
|
23
|
my ($v) = @_; |
305
|
14
|
|
|
|
|
23
|
local $_ = $v; |
306
|
14
|
100
|
66
|
|
|
50
|
if (!$predicate or $predicate->($v, $idx++)) { |
307
|
7
|
|
|
|
|
47
|
$count++; |
308
|
|
|
|
|
|
|
} |
309
|
|
|
|
|
|
|
}, |
310
|
|
|
|
|
|
|
complete => sub { |
311
|
3
|
50
|
|
|
|
19
|
$subscriber->{next}->($count) if defined $subscriber->{next}; |
312
|
3
|
50
|
|
|
|
15
|
$subscriber->{complete}->() if defined $subscriber->{complete}; |
313
|
|
|
|
|
|
|
}, |
314
|
3
|
|
|
|
|
21
|
}; |
315
|
|
|
|
|
|
|
|
316
|
3
|
|
|
|
|
11
|
$source->subscribe($own_subscriber); |
317
|
|
|
|
|
|
|
|
318
|
3
|
|
|
|
|
25
|
return; |
319
|
3
|
|
|
|
|
9
|
}); |
320
|
3
|
|
|
|
|
16
|
}; |
321
|
|
|
|
|
|
|
} |
322
|
|
|
|
|
|
|
|
323
|
|
|
|
|
|
|
sub op_debounce { |
324
|
2
|
|
|
2
|
|
5
|
my ($duration_selector) = @_; |
325
|
|
|
|
|
|
|
|
326
|
|
|
|
|
|
|
return sub { |
327
|
2
|
|
|
2
|
|
7
|
my ($source) = @_; |
328
|
|
|
|
|
|
|
|
329
|
|
|
|
|
|
|
return rx_observable->new(sub { |
330
|
2
|
|
|
|
|
8
|
my ($subscriber) = @_; |
331
|
|
|
|
|
|
|
|
332
|
2
|
|
|
|
|
5
|
my $mini_subscription; |
333
|
|
|
|
|
|
|
my $last_val; |
334
|
2
|
|
|
|
|
0
|
my $has_last_val; |
335
|
|
|
|
|
|
|
|
336
|
|
|
|
|
|
|
my $mini_subscriber = { |
337
|
|
|
|
|
|
|
next => sub { |
338
|
5
|
50
|
|
|
|
22
|
$subscriber->{next}->($last_val) if defined $subscriber->{next}; |
339
|
5
|
|
|
|
|
11
|
undef $has_last_val; |
340
|
|
|
|
|
|
|
}, |
341
|
|
|
|
|
|
|
error => sub { |
342
|
0
|
0
|
|
|
|
0
|
$subscriber->{error}->(@_) if defined $subscriber->{error}; |
343
|
|
|
|
|
|
|
}, |
344
|
|
|
|
|
|
|
complete => sub { |
345
|
5
|
|
|
|
|
10
|
undef $mini_subscription; |
346
|
|
|
|
|
|
|
}, |
347
|
2
|
|
|
|
|
12
|
}; |
348
|
|
|
|
|
|
|
|
349
|
|
|
|
|
|
|
my $own_subscriber = { |
350
|
|
|
|
|
|
|
next => sub { |
351
|
12
|
|
|
|
|
27
|
my ($v) = @_; |
352
|
|
|
|
|
|
|
|
353
|
12
|
100
|
|
|
|
28
|
if (defined $mini_subscription) { |
354
|
6
|
|
|
|
|
15
|
$mini_subscription->unsubscribe(); |
355
|
|
|
|
|
|
|
} |
356
|
|
|
|
|
|
|
|
357
|
12
|
|
|
|
|
22
|
$last_val = $v; |
358
|
12
|
|
|
|
|
17
|
$has_last_val = 1; |
359
|
|
|
|
|
|
|
|
360
|
12
|
|
|
|
|
19
|
my $o = do { local $_ = $v; $duration_selector->($v) }; |
|
12
|
|
|
|
|
21
|
|
|
12
|
|
|
|
|
23
|
|
361
|
12
|
|
|
|
|
36
|
$mini_subscription = $o->pipe( |
362
|
|
|
|
|
|
|
op_take(1), |
363
|
|
|
|
|
|
|
)->subscribe($mini_subscriber); |
364
|
|
|
|
|
|
|
}, |
365
|
|
|
|
|
|
|
error => sub { |
366
|
0
|
0
|
|
|
|
0
|
$subscriber->{error}->(@_) if defined $subscriber->{error}; |
367
|
|
|
|
|
|
|
}, |
368
|
|
|
|
|
|
|
complete => sub { |
369
|
2
|
100
|
|
|
|
8
|
if ($has_last_val) { |
370
|
1
|
50
|
|
|
|
17
|
$subscriber->{next}->($last_val) if defined $subscriber->{next}; |
371
|
|
|
|
|
|
|
} |
372
|
2
|
50
|
|
|
|
8
|
$subscriber->{complete}->() if defined $subscriber->{complete}; |
373
|
|
|
|
|
|
|
} |
374
|
2
|
|
|
|
|
13
|
}; |
375
|
|
|
|
|
|
|
|
376
|
2
|
|
|
|
|
8
|
my $main_subscription = $source->subscribe($own_subscriber); |
377
|
2
|
|
|
|
|
19
|
$main_subscription->add(\$mini_subscription); |
378
|
|
|
|
|
|
|
|
379
|
2
|
|
|
|
|
6
|
return $main_subscription, $mini_subscription; |
380
|
2
|
|
|
|
|
3
|
}); |
381
|
2
|
|
|
|
|
10
|
}; |
382
|
|
|
|
|
|
|
} |
383
|
|
|
|
|
|
|
|
384
|
|
|
|
|
|
|
sub op_debounce_time { |
385
|
2
|
|
|
2
|
|
4
|
my ($due_time) = @_; |
386
|
|
|
|
|
|
|
|
387
|
2
|
|
|
12
|
|
9
|
return op_debounce(sub { rx_timer($due_time) }); |
|
12
|
|
|
|
|
28
|
|
388
|
|
|
|
|
|
|
} |
389
|
|
|
|
|
|
|
|
390
|
|
|
|
|
|
|
sub op_default_if_empty { |
391
|
7
|
|
|
7
|
|
21
|
my ($default_value) = @_; |
392
|
|
|
|
|
|
|
|
393
|
|
|
|
|
|
|
return sub { |
394
|
7
|
|
|
7
|
|
27
|
my ($source) = @_; |
395
|
|
|
|
|
|
|
|
396
|
|
|
|
|
|
|
return rx_observable->new(sub { |
397
|
7
|
|
|
|
|
33
|
my ($subscriber) = @_; |
398
|
|
|
|
|
|
|
|
399
|
7
|
|
|
|
|
11
|
my $source_emitted = 0; |
400
|
|
|
|
|
|
|
|
401
|
|
|
|
|
|
|
my $own_subscriber = { |
402
|
|
|
|
|
|
|
%$subscriber, |
403
|
|
|
|
|
|
|
next => sub { |
404
|
3
|
|
|
|
|
6
|
$source_emitted = 1; |
405
|
3
|
50
|
|
|
|
26
|
$subscriber->{next}->(@_) if exists $subscriber->{next}; |
406
|
|
|
|
|
|
|
}, |
407
|
|
|
|
|
|
|
complete => sub { |
408
|
7
|
50
|
66
|
|
|
35
|
$subscriber->{next}->($default_value) if ! $source_emitted and exists $subscriber->{next}; |
409
|
7
|
50
|
|
|
|
26
|
$subscriber->{complete}->() if exists $subscriber->{complete}; |
410
|
|
|
|
|
|
|
}, |
411
|
7
|
|
|
|
|
46
|
}; |
412
|
|
|
|
|
|
|
|
413
|
7
|
|
|
|
|
31
|
$source->subscribe($own_subscriber); |
414
|
|
|
|
|
|
|
|
415
|
7
|
|
|
|
|
32
|
return; |
416
|
7
|
|
|
|
|
18
|
}); |
417
|
7
|
|
|
|
|
37
|
}; |
418
|
|
|
|
|
|
|
} |
419
|
|
|
|
|
|
|
|
420
|
|
|
|
|
|
|
sub op_delay { |
421
|
674
|
|
|
674
|
|
1167
|
my ($delay) = @_; |
422
|
|
|
|
|
|
|
|
423
|
674
|
|
|
|
|
1367
|
my ($timer_sub, $cancel_timer_sub) = get_timer_subs; |
424
|
|
|
|
|
|
|
|
425
|
|
|
|
|
|
|
return sub { |
426
|
674
|
|
|
674
|
|
1019
|
my ($source) = @_; |
427
|
|
|
|
|
|
|
|
428
|
|
|
|
|
|
|
return rx_observable->new(sub { |
429
|
724
|
|
|
|
|
1193
|
my ($subscriber) = @_; |
430
|
|
|
|
|
|
|
|
431
|
724
|
|
|
|
|
1539
|
my %timers; |
432
|
|
|
|
|
|
|
my $queue; |
433
|
724
|
|
|
|
|
0
|
my $completed; |
434
|
|
|
|
|
|
|
my $own_subscriber = { |
435
|
|
|
|
|
|
|
error => sub { |
436
|
1
|
|
|
|
|
3
|
my @value = @_; |
437
|
1
|
50
|
|
|
|
5
|
$subscriber->{error}->(@value) if defined $subscriber->{error}; |
438
|
|
|
|
|
|
|
}, |
439
|
|
|
|
|
|
|
next => sub { |
440
|
730
|
|
|
|
|
1545
|
my @value = @_; |
441
|
|
|
|
|
|
|
|
442
|
730
|
100
|
|
|
|
1477
|
if (!defined $queue) { |
443
|
729
|
|
|
|
|
1166
|
$queue = []; |
444
|
729
|
|
|
|
|
1110
|
my ($timer1, $timer2); |
445
|
|
|
|
|
|
|
$timer1 = $timer_sub->(0, sub { |
446
|
729
|
|
|
|
|
1216
|
delete $timers{$timer1}; |
447
|
729
|
|
|
|
|
1395
|
my @queue_copy = @$queue; |
448
|
729
|
|
|
|
|
1170
|
undef $queue; |
449
|
|
|
|
|
|
|
$timer2 = $timer_sub->($delay, sub { |
450
|
694
|
|
|
|
|
1205
|
delete $timers{$timer2}; |
451
|
694
|
|
|
|
|
1188
|
foreach my $item (@queue_copy) { |
452
|
695
|
100
|
|
|
|
2022
|
$subscriber->{next}->(@$item) if defined $subscriber->{next}; |
453
|
|
|
|
|
|
|
} |
454
|
694
|
100
|
100
|
|
|
2462
|
if ($completed and ! %timers) { |
455
|
689
|
100
|
|
|
|
2097
|
$subscriber->{complete}->() if defined $subscriber->{complete}; |
456
|
|
|
|
|
|
|
} |
457
|
729
|
|
|
|
|
2593
|
}); |
458
|
729
|
|
|
|
|
2097
|
$timers{$timer2} = $timer2; |
459
|
729
|
|
|
|
|
3063
|
}); |
460
|
729
|
|
|
|
|
1751
|
$timers{$timer1} = $timer1; |
461
|
|
|
|
|
|
|
} |
462
|
730
|
|
|
|
|
2708
|
push @$queue, \@value; |
463
|
|
|
|
|
|
|
}, |
464
|
|
|
|
|
|
|
complete => sub { |
465
|
723
|
|
|
|
|
1026
|
$completed = 1; |
466
|
723
|
100
|
|
|
|
1575
|
if (! %timers) { |
467
|
1
|
50
|
|
|
|
5
|
$subscriber->{complete}->() if defined $subscriber->{complete}; |
468
|
|
|
|
|
|
|
} |
469
|
|
|
|
|
|
|
}, |
470
|
724
|
|
|
|
|
4406
|
}; |
471
|
|
|
|
|
|
|
|
472
|
|
|
|
|
|
|
return [ |
473
|
|
|
|
|
|
|
$source->subscribe($own_subscriber), |
474
|
|
|
|
|
|
|
sub { |
475
|
724
|
|
|
|
|
1623
|
$cancel_timer_sub->($_) foreach values %timers; |
476
|
724
|
|
|
|
|
1648
|
%timers = (); |
477
|
|
|
|
|
|
|
}, |
478
|
724
|
|
|
|
|
1817
|
]; |
479
|
674
|
|
|
|
|
1255
|
}); |
480
|
674
|
|
|
|
|
2875
|
}; |
481
|
|
|
|
|
|
|
} |
482
|
|
|
|
|
|
|
|
483
|
|
|
|
|
|
|
sub op_delay_when { |
484
|
1
|
|
|
1
|
|
5
|
my ($delay_duration_selector) = @_; |
485
|
|
|
|
|
|
|
|
486
|
|
|
|
|
|
|
return sub { |
487
|
1
|
|
|
1
|
|
5
|
my ($source) = @_; |
488
|
|
|
|
|
|
|
|
489
|
|
|
|
|
|
|
return rx_observable->new(sub { |
490
|
1
|
|
|
|
|
5
|
my ($subscriber) = @_; |
491
|
|
|
|
|
|
|
|
492
|
1
|
|
|
|
|
3
|
my $idx = 0; |
493
|
1
|
|
|
|
|
2
|
my %mini_subscriptions; |
494
|
|
|
|
|
|
|
my $main_finished; |
495
|
|
|
|
|
|
|
|
496
|
|
|
|
|
|
|
my $make_mini_subscriber = sub { |
497
|
3
|
|
|
|
|
7
|
my ($v) = @_; |
498
|
|
|
|
|
|
|
|
499
|
3
|
|
|
|
|
9
|
my $mini_subscription = RxPerl::Subscription->new; |
500
|
3
|
|
|
|
|
12
|
$mini_subscriptions{$mini_subscription} = $mini_subscription; |
501
|
|
|
|
|
|
|
|
502
|
|
|
|
|
|
|
return { |
503
|
|
|
|
|
|
|
new_subscription => $mini_subscription, |
504
|
|
|
|
|
|
|
next => sub { |
505
|
3
|
50
|
|
|
|
17
|
$subscriber->{next}->($v) if defined $subscriber->{next}; |
506
|
|
|
|
|
|
|
}, |
507
|
|
|
|
|
|
|
error => sub { |
508
|
0
|
0
|
|
|
|
0
|
$subscriber->{error}->(@_) if defined $subscriber->{error}; |
509
|
|
|
|
|
|
|
}, |
510
|
|
|
|
|
|
|
complete => sub { |
511
|
3
|
|
|
|
|
9
|
delete $mini_subscriptions{$mini_subscription}; |
512
|
3
|
100
|
66
|
|
|
15
|
if ($main_finished and ! %mini_subscriptions) { |
513
|
1
|
50
|
|
|
|
8
|
$subscriber->{complete}->() if defined $subscriber->{complete}; |
514
|
|
|
|
|
|
|
} |
515
|
|
|
|
|
|
|
}, |
516
|
3
|
|
|
|
|
24
|
}; |
517
|
1
|
|
|
|
|
4
|
}; |
518
|
|
|
|
|
|
|
|
519
|
1
|
|
|
|
|
4
|
my $own_subscription = RxPerl::Subscription->new; |
520
|
|
|
|
|
|
|
my $own_subscriber = { |
521
|
|
|
|
|
|
|
new_subscription => $own_subscription, |
522
|
|
|
|
|
|
|
next => sub { |
523
|
3
|
|
|
|
|
6
|
my ($v) = @_; |
524
|
|
|
|
|
|
|
|
525
|
3
|
|
|
|
|
5
|
local $_ = $v; |
526
|
3
|
|
|
|
|
9
|
my $mini_obs = $delay_duration_selector->($v, $idx++); |
527
|
3
|
|
|
|
|
11
|
$mini_obs->subscribe($make_mini_subscriber->($v)); |
528
|
|
|
|
|
|
|
}, |
529
|
|
|
|
|
|
|
error => sub { |
530
|
0
|
0
|
|
|
|
0
|
$subscriber->{error}->(@_) if defined $subscriber->{error}; |
531
|
|
|
|
|
|
|
}, |
532
|
|
|
|
|
|
|
complete => sub { |
533
|
1
|
|
|
|
|
2
|
$main_finished = 1; |
534
|
1
|
50
|
|
|
|
4
|
if (!%mini_subscriptions) { |
535
|
0
|
0
|
|
|
|
0
|
$subscriber->{complete}->() if defined $subscriber->{complete}; |
536
|
|
|
|
|
|
|
} |
537
|
|
|
|
|
|
|
}, |
538
|
1
|
|
|
|
|
11
|
}; |
539
|
|
|
|
|
|
|
|
540
|
1
|
|
|
|
|
5
|
$subscriber->subscription->add($own_subscription, \%mini_subscriptions); |
541
|
|
|
|
|
|
|
|
542
|
1
|
|
|
|
|
6
|
$source->subscribe($own_subscriber); |
543
|
|
|
|
|
|
|
|
544
|
1
|
|
|
|
|
11
|
return; |
545
|
1
|
|
|
|
|
3
|
}); |
546
|
1
|
|
|
|
|
6
|
}; |
547
|
|
|
|
|
|
|
} |
548
|
|
|
|
|
|
|
|
549
|
|
|
|
|
|
|
sub op_distinct { |
550
|
2
|
|
|
2
|
|
6
|
my ($key_selector) = @_; |
551
|
|
|
|
|
|
|
|
552
|
|
|
|
|
|
|
return sub { |
553
|
2
|
|
|
2
|
|
6
|
my ($source) = @_; |
554
|
|
|
|
|
|
|
|
555
|
|
|
|
|
|
|
return rx_observable->new(sub { |
556
|
2
|
|
|
|
|
5
|
my ($subscriber) = @_; |
557
|
|
|
|
|
|
|
|
558
|
2
|
|
|
|
|
3
|
my %keys_passed; |
559
|
2
|
|
|
|
|
8
|
$subscriber->subscription->add(sub { %keys_passed = () }); |
|
2
|
|
|
|
|
14
|
|
560
|
|
|
|
|
|
|
|
561
|
|
|
|
|
|
|
my $own_subscriber = { |
562
|
|
|
|
|
|
|
%$subscriber, |
563
|
|
|
|
|
|
|
next => sub { |
564
|
15
|
|
|
|
|
20
|
my ($v) = @_; |
565
|
|
|
|
|
|
|
|
566
|
15
|
|
|
|
|
20
|
my $k; |
567
|
15
|
100
|
|
|
|
26
|
if ($key_selector) { |
568
|
3
|
|
|
|
|
5
|
my $ok = eval { local $_ = $v; $k = $key_selector->($v); 1 }; |
|
3
|
|
|
|
|
5
|
|
|
3
|
|
|
|
|
8
|
|
|
3
|
|
|
|
|
12
|
|
569
|
3
|
50
|
|
|
|
7
|
if (! $ok) { |
570
|
0
|
0
|
|
|
|
0
|
$subscriber->{error}->($@) if defined $subscriber->{error}; |
571
|
0
|
|
|
|
|
0
|
return; |
572
|
|
|
|
|
|
|
} |
573
|
|
|
|
|
|
|
} else { |
574
|
12
|
|
|
|
|
18
|
$k = $v; |
575
|
|
|
|
|
|
|
} |
576
|
15
|
100
|
|
|
|
40
|
if (! exists $keys_passed{$k}) { |
577
|
6
|
|
|
|
|
12
|
$keys_passed{$k} = 1; |
578
|
6
|
50
|
|
|
|
18
|
$subscriber->{next}->($v) if defined $subscriber->{next}; |
579
|
|
|
|
|
|
|
} |
580
|
|
|
|
|
|
|
}, |
581
|
2
|
|
|
|
|
13
|
}; |
582
|
|
|
|
|
|
|
|
583
|
2
|
|
|
|
|
11
|
$source->subscribe($own_subscriber); |
584
|
|
|
|
|
|
|
|
585
|
2
|
|
|
|
|
19
|
return; |
586
|
2
|
|
|
|
|
5
|
}); |
587
|
2
|
|
|
|
|
13
|
}; |
588
|
|
|
|
|
|
|
} |
589
|
|
|
|
|
|
|
|
590
|
|
|
|
|
|
|
sub op_distinct_until_changed { |
591
|
4
|
|
|
4
|
|
10
|
my ($comparison_function) = @_; |
592
|
|
|
|
|
|
|
|
593
|
4
|
|
100
|
|
|
18
|
$comparison_function //= \&_eqq; |
594
|
|
|
|
|
|
|
|
595
|
|
|
|
|
|
|
return sub { |
596
|
4
|
|
|
4
|
|
9
|
my ($source) = @_; |
597
|
|
|
|
|
|
|
|
598
|
|
|
|
|
|
|
return rx_observable->new(sub { |
599
|
4
|
|
|
|
|
21
|
my ($subscriber) = @_; |
600
|
|
|
|
|
|
|
|
601
|
4
|
|
|
|
|
9
|
my $prev_value; |
602
|
4
|
|
|
|
|
6
|
my $have_prev_value = 0; |
603
|
|
|
|
|
|
|
|
604
|
|
|
|
|
|
|
my $own_subscriber = { |
605
|
|
|
|
|
|
|
%$subscriber, |
606
|
|
|
|
|
|
|
next => sub { |
607
|
27
|
|
|
|
|
47
|
my @value = @_; |
608
|
|
|
|
|
|
|
|
609
|
27
|
100
|
100
|
|
|
96
|
if (! $have_prev_value or ! $comparison_function->($prev_value, $value[0])) { |
610
|
15
|
50
|
|
|
|
55
|
$subscriber->{next}->(@value) if defined $subscriber->{next}; |
611
|
15
|
|
|
|
|
23
|
$have_prev_value = 1; |
612
|
15
|
|
|
|
|
40
|
$prev_value = $value[0]; |
613
|
|
|
|
|
|
|
} |
614
|
|
|
|
|
|
|
}, |
615
|
4
|
|
|
|
|
32
|
}; |
616
|
|
|
|
|
|
|
|
617
|
4
|
|
|
|
|
16
|
$source->subscribe($own_subscriber); |
618
|
|
|
|
|
|
|
|
619
|
4
|
|
|
|
|
24
|
return; |
620
|
4
|
|
|
|
|
10
|
}); |
621
|
4
|
|
|
|
|
22
|
}; |
622
|
|
|
|
|
|
|
} |
623
|
|
|
|
|
|
|
|
624
|
|
|
|
|
|
|
sub op_distinct_until_key_changed { |
625
|
1
|
|
|
1
|
|
3
|
my ($key) = @_; |
626
|
|
|
|
|
|
|
|
627
|
|
|
|
|
|
|
return op_distinct_until_changed(sub { |
628
|
5
|
|
|
5
|
|
14
|
_eqq($_[0]->{$key}, $_[1]->{$key}); |
629
|
1
|
|
|
|
|
5
|
}), |
630
|
|
|
|
|
|
|
} |
631
|
|
|
|
|
|
|
|
632
|
|
|
|
|
|
|
sub op_element_at { |
633
|
2
|
|
|
2
|
|
7
|
my ($index, $default) = @_; |
634
|
2
|
|
|
|
|
6
|
my $has_default = @_ >= 2; |
635
|
2
|
|
|
|
|
6
|
$index = int $index; |
636
|
|
|
|
|
|
|
|
637
|
|
|
|
|
|
|
return sub { |
638
|
2
|
|
|
2
|
|
6
|
my ($source) = @_; |
639
|
|
|
|
|
|
|
|
640
|
2
|
50
|
|
|
|
13
|
$index >= 0 or return rx_throw_error('ArgumentOutOfRangeError'); |
641
|
|
|
|
|
|
|
|
642
|
|
|
|
|
|
|
return rx_observable->new(sub { |
643
|
2
|
|
|
|
|
8
|
my ($subscriber) = @_; |
644
|
|
|
|
|
|
|
|
645
|
2
|
|
|
|
|
5
|
my $i = 0; |
646
|
|
|
|
|
|
|
my $own_subscriber = { |
647
|
|
|
|
|
|
|
%$subscriber, |
648
|
|
|
|
|
|
|
next => sub { |
649
|
6
|
100
|
|
|
|
23
|
if ($i++ == $index) { |
650
|
1
|
50
|
|
|
|
7
|
$subscriber->{next}->(@_) if defined $subscriber->{next}; |
651
|
1
|
50
|
|
|
|
7
|
$subscriber->{complete}->() if defined $subscriber->{complete}; |
652
|
|
|
|
|
|
|
} |
653
|
|
|
|
|
|
|
}, |
654
|
|
|
|
|
|
|
complete => sub { |
655
|
1
|
50
|
|
|
|
7
|
if ($has_default) { |
656
|
1
|
50
|
|
|
|
9
|
$subscriber->{next}->($default) if defined $subscriber->{next}; |
657
|
1
|
50
|
|
|
|
5
|
$subscriber->{complete}->() if defined $subscriber->{complete}; |
658
|
|
|
|
|
|
|
} else { |
659
|
0
|
0
|
|
|
|
0
|
$subscriber->{error}->('ArgumentOutOfRangeError') if defined $subscriber->{error}; |
660
|
|
|
|
|
|
|
} |
661
|
|
|
|
|
|
|
}, |
662
|
2
|
|
|
|
|
16
|
}; |
663
|
|
|
|
|
|
|
|
664
|
2
|
|
|
|
|
10
|
$source->subscribe($own_subscriber); |
665
|
|
|
|
|
|
|
|
666
|
2
|
|
|
|
|
10
|
return; |
667
|
2
|
|
|
|
|
6
|
}); |
668
|
2
|
|
|
|
|
10
|
}; |
669
|
|
|
|
|
|
|
} |
670
|
|
|
|
|
|
|
|
671
|
|
|
|
|
|
|
sub op_end_with { |
672
|
0
|
|
|
0
|
|
0
|
my (@values) = @_; |
673
|
|
|
|
|
|
|
|
674
|
|
|
|
|
|
|
return sub { |
675
|
0
|
|
|
0
|
|
0
|
my ($source) = @_; |
676
|
|
|
|
|
|
|
|
677
|
0
|
|
|
|
|
0
|
return rx_concat( |
678
|
|
|
|
|
|
|
$source, |
679
|
|
|
|
|
|
|
rx_of(@values), |
680
|
|
|
|
|
|
|
); |
681
|
|
|
|
|
|
|
} |
682
|
0
|
|
|
|
|
0
|
} |
683
|
|
|
|
|
|
|
|
684
|
|
|
|
|
|
|
sub op_every { |
685
|
2
|
|
|
2
|
|
8
|
my ($predicate) = @_; |
686
|
|
|
|
|
|
|
|
687
|
|
|
|
|
|
|
return sub { |
688
|
2
|
|
|
2
|
|
8
|
my ($source) = @_; |
689
|
|
|
|
|
|
|
|
690
|
|
|
|
|
|
|
return rx_observable->new(sub { |
691
|
2
|
|
|
|
|
5
|
my ($subscriber) = @_; |
692
|
|
|
|
|
|
|
|
693
|
2
|
|
|
|
|
4
|
my $idx = 0; |
694
|
|
|
|
|
|
|
my $own_subscriber = { |
695
|
|
|
|
|
|
|
%$subscriber, |
696
|
|
|
|
|
|
|
next => sub { |
697
|
8
|
|
|
|
|
16
|
my ($v) = @_; |
698
|
8
|
|
|
|
|
12
|
local $_ = $v; |
699
|
8
|
100
|
|
|
|
19
|
if (! $predicate->($v, $idx++)) { |
700
|
1
|
50
|
|
|
|
12
|
$subscriber->{next}->(0) if defined $subscriber->{next}; |
701
|
1
|
50
|
|
|
|
9
|
$subscriber->{complete}->() if defined $subscriber->{complete}; |
702
|
|
|
|
|
|
|
} |
703
|
|
|
|
|
|
|
}, |
704
|
|
|
|
|
|
|
complete => sub { |
705
|
1
|
50
|
|
|
|
7
|
$subscriber->{next}->(1) if defined $subscriber->{next}; |
706
|
1
|
50
|
|
|
|
4
|
$subscriber->{complete}->() if defined $subscriber->{complete}; |
707
|
|
|
|
|
|
|
}, |
708
|
2
|
|
|
|
|
15
|
}; |
709
|
|
|
|
|
|
|
|
710
|
2
|
|
|
|
|
9
|
$source->subscribe($own_subscriber); |
711
|
|
|
|
|
|
|
|
712
|
2
|
|
|
|
|
6
|
return; |
713
|
2
|
|
|
|
|
5
|
}); |
714
|
2
|
|
|
|
|
18
|
}; |
715
|
|
|
|
|
|
|
} |
716
|
|
|
|
|
|
|
|
717
|
|
|
|
|
|
|
sub op_exhaust_all { |
718
|
|
|
|
|
|
|
return sub { |
719
|
1
|
|
|
1
|
|
5
|
my ($source) = @_; |
720
|
|
|
|
|
|
|
|
721
|
|
|
|
|
|
|
return rx_observable->new(sub { |
722
|
1
|
|
|
|
|
3
|
my ($subscriber) = @_; |
723
|
|
|
|
|
|
|
|
724
|
1
|
|
|
|
|
2
|
my $active_subscription; |
725
|
|
|
|
|
|
|
my $big_completed; |
726
|
1
|
|
|
|
|
4
|
my $own_subscription = RxPerl::Subscription->new; |
727
|
|
|
|
|
|
|
|
728
|
1
|
|
|
|
|
3
|
$subscriber->subscription->add( |
729
|
|
|
|
|
|
|
\$active_subscription, |
730
|
|
|
|
|
|
|
$own_subscription, |
731
|
|
|
|
|
|
|
); |
732
|
|
|
|
|
|
|
|
733
|
|
|
|
|
|
|
my $own_subscriber = { |
734
|
|
|
|
|
|
|
new_subscription => $own_subscription, |
735
|
|
|
|
|
|
|
next => sub { |
736
|
6
|
|
|
|
|
12
|
my ($new_obs) = @_; |
737
|
|
|
|
|
|
|
|
738
|
6
|
100
|
|
|
|
46
|
!$active_subscription or return; |
739
|
3
|
|
|
|
|
13
|
$active_subscription = RxPerl::Subscription->new; |
740
|
|
|
|
|
|
|
my $small_subscriber = { |
741
|
|
|
|
|
|
|
new_subscription => $active_subscription, |
742
|
|
|
|
|
|
|
next => sub { |
743
|
12
|
50
|
|
|
|
31
|
$subscriber->{next}->(@_) if defined $subscriber->{next}; |
744
|
|
|
|
|
|
|
}, |
745
|
|
|
|
|
|
|
error => sub { |
746
|
0
|
0
|
|
|
|
0
|
$subscriber->{error}->(@_) if defined $subscriber->{error}; |
747
|
|
|
|
|
|
|
}, |
748
|
|
|
|
|
|
|
complete => sub { |
749
|
2
|
|
|
|
|
4
|
undef $active_subscription; |
750
|
2
|
50
|
33
|
|
|
9
|
$subscriber->{complete}->() if $big_completed and defined $subscriber->{complete}; |
751
|
|
|
|
|
|
|
}, |
752
|
3
|
|
|
|
|
21
|
}; |
753
|
3
|
|
|
|
|
10
|
$new_obs->subscribe($small_subscriber); |
754
|
|
|
|
|
|
|
}, |
755
|
|
|
|
|
|
|
error => sub { |
756
|
0
|
0
|
|
|
|
0
|
$subscriber->{error}->(@_) if defined $subscriber->{error}; |
757
|
|
|
|
|
|
|
}, |
758
|
|
|
|
|
|
|
complete => sub { |
759
|
0
|
|
|
|
|
0
|
$big_completed = 1; |
760
|
0
|
0
|
0
|
|
|
0
|
$subscriber->{complete}->() if !$active_subscription and defined $subscriber->{complete}; |
761
|
|
|
|
|
|
|
}, |
762
|
1
|
|
|
|
|
11
|
}; |
763
|
|
|
|
|
|
|
|
764
|
1
|
|
|
|
|
5
|
$source->subscribe($own_subscriber); |
765
|
|
|
|
|
|
|
|
766
|
1
|
|
|
|
|
3
|
return; |
767
|
1
|
|
|
|
|
2
|
}); |
768
|
1
|
|
|
1
|
|
6
|
}; |
769
|
|
|
|
|
|
|
} |
770
|
|
|
|
|
|
|
|
771
|
|
|
|
|
|
|
sub op_exhaust_map { |
772
|
0
|
|
|
0
|
|
0
|
my ($observable_factory) = @_; |
773
|
|
|
|
|
|
|
|
774
|
|
|
|
|
|
|
return sub { |
775
|
0
|
|
|
0
|
|
0
|
my ($source) = @_; |
776
|
|
|
|
|
|
|
|
777
|
0
|
|
|
|
|
0
|
return $source->pipe( |
778
|
|
|
|
|
|
|
op_map($observable_factory), |
779
|
|
|
|
|
|
|
op_exhaust_all(), |
780
|
|
|
|
|
|
|
); |
781
|
0
|
|
|
|
|
0
|
}; |
782
|
|
|
|
|
|
|
} |
783
|
|
|
|
|
|
|
|
784
|
|
|
|
|
|
|
sub op_filter { |
785
|
9
|
|
|
9
|
|
21
|
my ($filtering_sub) = @_; |
786
|
|
|
|
|
|
|
|
787
|
|
|
|
|
|
|
return sub { |
788
|
9
|
|
|
9
|
|
20
|
my ($source) = @_; |
789
|
|
|
|
|
|
|
|
790
|
|
|
|
|
|
|
return rx_observable->new(sub { |
791
|
9
|
|
|
|
|
20
|
my ($subscriber) = @_; |
792
|
|
|
|
|
|
|
|
793
|
9
|
|
|
|
|
31
|
my $own_subscriber = { %$subscriber }; |
794
|
9
|
|
|
|
|
21
|
my $idx = 0; |
795
|
|
|
|
|
|
|
$own_subscriber->{next} = sub { |
796
|
75
|
|
|
|
|
126
|
my ($value) = @_; |
797
|
75
|
|
|
|
|
446
|
my $passes = eval { |
798
|
75
|
|
|
|
|
141
|
local $_ = $value; |
799
|
75
|
|
|
|
|
181
|
$filtering_sub->($value, $idx++); |
800
|
|
|
|
|
|
|
}; |
801
|
75
|
50
|
|
|
|
358
|
if (my $error = $@) { |
802
|
0
|
|
|
|
|
0
|
$subscriber->{error}->($error); |
803
|
|
|
|
|
|
|
} else { |
804
|
75
|
100
|
66
|
|
|
296
|
$subscriber->{next}->(@_) if $passes and defined $subscriber->{next}; |
805
|
|
|
|
|
|
|
} |
806
|
9
|
|
|
|
|
35
|
}; |
807
|
|
|
|
|
|
|
|
808
|
9
|
|
|
|
|
30
|
$source->subscribe($own_subscriber); |
809
|
|
|
|
|
|
|
|
810
|
9
|
|
|
|
|
27
|
return; |
811
|
9
|
|
|
|
|
20
|
}); |
812
|
9
|
|
|
|
|
37
|
}; |
813
|
|
|
|
|
|
|
} |
814
|
|
|
|
|
|
|
|
815
|
|
|
|
|
|
|
sub op_finalize { |
816
|
8
|
|
|
8
|
|
15
|
my ($fn) = @_; |
817
|
|
|
|
|
|
|
|
818
|
|
|
|
|
|
|
return sub { |
819
|
8
|
|
|
8
|
|
14
|
my ($source) = @_; |
820
|
|
|
|
|
|
|
|
821
|
|
|
|
|
|
|
return rx_observable->new(sub { |
822
|
8
|
|
|
|
|
15
|
my ($subscriber) = @_; |
823
|
|
|
|
|
|
|
|
824
|
8
|
|
100
|
|
|
26
|
my $arr = $subscriber->{_subscription}{_finalize_cbs} //= []; |
825
|
8
|
|
|
|
|
17
|
unshift @$arr, $fn; |
826
|
8
|
|
|
|
|
22
|
$subscriber->{_subscription}->add( $arr ); |
827
|
|
|
|
|
|
|
|
828
|
8
|
|
|
|
|
26
|
$source->subscribe($subscriber); |
829
|
|
|
|
|
|
|
|
830
|
8
|
|
|
|
|
15
|
return; |
831
|
8
|
|
|
|
|
16
|
}); |
832
|
8
|
|
|
|
|
42
|
}; |
833
|
|
|
|
|
|
|
} |
834
|
|
|
|
|
|
|
|
835
|
|
|
|
|
|
|
sub op_find { |
836
|
2
|
|
|
2
|
|
4
|
my ($predicate) = @_; |
837
|
|
|
|
|
|
|
|
838
|
|
|
|
|
|
|
return sub { |
839
|
2
|
|
|
2
|
|
6
|
my ($source) = @_; |
840
|
|
|
|
|
|
|
|
841
|
2
|
50
|
|
|
|
7
|
$predicate or return rx_throw_error('missing predicate in op_find'); |
842
|
|
|
|
|
|
|
|
843
|
2
|
|
|
|
|
7
|
return $source->pipe( |
844
|
|
|
|
|
|
|
op_first($predicate), |
845
|
|
|
|
|
|
|
op_default_if_empty(undef), |
846
|
|
|
|
|
|
|
); |
847
|
2
|
|
|
|
|
11
|
}; |
848
|
|
|
|
|
|
|
} |
849
|
|
|
|
|
|
|
|
850
|
|
|
|
|
|
|
sub op_find_index { |
851
|
2
|
|
|
2
|
|
5
|
my ($predicate) = @_; |
852
|
|
|
|
|
|
|
|
853
|
|
|
|
|
|
|
return sub { |
854
|
2
|
|
|
2
|
|
5
|
my ($source) = @_; |
855
|
|
|
|
|
|
|
|
856
|
2
|
50
|
|
|
|
5
|
$predicate or return rx_throw_error('missing predicate in op_find_index'); |
857
|
|
|
|
|
|
|
|
858
|
|
|
|
|
|
|
return rx_observable->new(sub { |
859
|
2
|
|
|
|
|
11
|
my ($subscriber) = @_; |
860
|
|
|
|
|
|
|
|
861
|
2
|
|
|
|
|
8
|
my $idx = 0; |
862
|
|
|
|
|
|
|
my $own_subscriber = { |
863
|
|
|
|
|
|
|
%$subscriber, |
864
|
|
|
|
|
|
|
next => sub { |
865
|
13
|
|
|
|
|
23
|
my ($val) = @_; |
866
|
13
|
|
|
|
|
17
|
my $truth; |
867
|
13
|
|
|
|
|
18
|
my $ok = eval { |
868
|
13
|
|
|
|
|
20
|
local $_ = $val; |
869
|
13
|
|
|
|
|
27
|
$truth = $predicate->($val, $idx++); |
870
|
13
|
|
|
|
|
46
|
1 |
871
|
|
|
|
|
|
|
}; |
872
|
13
|
50
|
|
|
|
25
|
if (!$ok) { |
873
|
0
|
0
|
|
|
|
0
|
$subscriber->{error}->($@) if defined $subscriber->{error}; |
874
|
|
|
|
|
|
|
} |
875
|
13
|
100
|
|
|
|
48
|
if ($truth) { |
876
|
1
|
50
|
|
|
|
14
|
$subscriber->{next}->($idx - 1) if defined $subscriber->{next}; |
877
|
1
|
50
|
|
|
|
6
|
$subscriber->{complete}->() if defined $subscriber->{complete}; |
878
|
|
|
|
|
|
|
} |
879
|
|
|
|
|
|
|
}, |
880
|
|
|
|
|
|
|
complete => sub { |
881
|
1
|
50
|
|
|
|
6
|
$subscriber->{next}->(-1) if defined $subscriber->{next}; |
882
|
1
|
50
|
|
|
|
5
|
$subscriber->{complete}->() if defined $subscriber->{complete}; |
883
|
|
|
|
|
|
|
}, |
884
|
2
|
|
|
|
|
15
|
}; |
885
|
|
|
|
|
|
|
|
886
|
2
|
|
|
|
|
10
|
$source->subscribe($own_subscriber); |
887
|
|
|
|
|
|
|
|
888
|
2
|
|
|
|
|
5
|
return; |
889
|
2
|
|
|
|
|
6
|
}); |
890
|
2
|
|
|
|
|
10
|
}; |
891
|
|
|
|
|
|
|
} |
892
|
|
|
|
|
|
|
|
893
|
|
|
|
|
|
|
sub op_first { |
894
|
7
|
|
|
7
|
|
15
|
my ($condition) = @_; |
895
|
|
|
|
|
|
|
|
896
|
|
|
|
|
|
|
return sub { |
897
|
7
|
|
|
7
|
|
15
|
my ($source) = @_; |
898
|
|
|
|
|
|
|
|
899
|
7
|
|
|
|
|
18
|
my @pipes = (op_take(1)); |
900
|
7
|
100
|
|
|
|
41
|
unshift @pipes, op_filter($condition) if defined $condition; |
901
|
|
|
|
|
|
|
|
902
|
7
|
|
|
|
|
25
|
return $source->pipe(@pipes); |
903
|
7
|
|
|
|
|
39
|
}; |
904
|
|
|
|
|
|
|
} |
905
|
|
|
|
|
|
|
|
906
|
|
|
|
|
|
|
sub op_group_by { |
907
|
1
|
|
|
1
|
|
3
|
my ($key_fn) = @_; |
908
|
|
|
|
|
|
|
|
909
|
|
|
|
|
|
|
return sub { |
910
|
1
|
|
|
1
|
|
3
|
my ($source) = @_; |
911
|
|
|
|
|
|
|
|
912
|
|
|
|
|
|
|
return rx_observable->new(sub { |
913
|
1
|
|
|
|
|
2
|
my ($subscriber) = @_; |
914
|
|
|
|
|
|
|
|
915
|
1
|
|
|
|
|
4
|
my %observables; |
916
|
|
|
|
|
|
|
my @observables; |
917
|
1
|
|
|
|
|
0
|
my $stop_producing_observables; |
918
|
|
|
|
|
|
|
|
919
|
|
|
|
|
|
|
$subscriber->subscription->add(sub { |
920
|
1
|
|
|
|
|
3
|
$stop_producing_observables = 1; |
921
|
1
|
|
|
|
|
4
|
}); |
922
|
|
|
|
|
|
|
|
923
|
|
|
|
|
|
|
my $own_subscriber = { |
924
|
|
|
|
|
|
|
# %$subscriber, |
925
|
|
|
|
|
|
|
next => sub { |
926
|
5
|
|
|
|
|
9
|
my ($v) = @_; |
927
|
|
|
|
|
|
|
|
928
|
5
|
|
|
|
|
9
|
my $key = do { local $_ = $v; $key_fn->($v); }; |
|
5
|
|
|
|
|
8
|
|
|
5
|
|
|
|
|
12
|
|
929
|
5
|
50
|
66
|
|
|
32
|
$observables{$key} //= do { |
930
|
2
|
|
|
|
|
8
|
my $new_obs = rx_subject->new; |
931
|
2
|
|
|
|
|
6
|
push @observables, $new_obs; |
932
|
2
|
|
|
|
|
7
|
$subscriber->{next}->($new_obs); |
933
|
2
|
|
|
|
|
8
|
$new_obs; |
934
|
|
|
|
|
|
|
} unless $stop_producing_observables; |
935
|
5
|
50
|
33
|
|
|
28
|
$observables{$key}{next}->($v) if exists $observables{$key} and defined $observables{$key}{next}; |
936
|
|
|
|
|
|
|
}, |
937
|
|
|
|
|
|
|
error => sub { |
938
|
0
|
0
|
|
|
|
0
|
$subscriber->{error}->() if defined $subscriber->{error}; |
939
|
|
|
|
|
|
|
}, |
940
|
|
|
|
|
|
|
complete => sub { |
941
|
1
|
|
|
|
|
3
|
for my $val (@observables) { |
942
|
2
|
50
|
|
|
|
26
|
$val->{complete}->() if defined $val->{complete}; |
943
|
|
|
|
|
|
|
} |
944
|
1
|
50
|
|
|
|
13
|
$subscriber->{complete}->() if defined $subscriber->{complete}; |
945
|
1
|
|
|
|
|
2
|
undef @observables; |
946
|
1
|
|
|
|
|
3
|
%observables = (); |
947
|
|
|
|
|
|
|
}, |
948
|
1
|
|
|
|
|
9
|
}; |
949
|
|
|
|
|
|
|
|
950
|
1
|
|
|
|
|
4
|
$source->subscribe($own_subscriber); |
951
|
|
|
|
|
|
|
|
952
|
1
|
|
|
|
|
3
|
return; |
953
|
1
|
|
|
|
|
4
|
}); |
954
|
1
|
|
|
|
|
9
|
}; |
955
|
|
|
|
|
|
|
} |
956
|
|
|
|
|
|
|
|
957
|
|
|
|
|
|
|
sub op_ignore_elements { |
958
|
|
|
|
|
|
|
return sub { |
959
|
68
|
|
|
68
|
|
120
|
my ($source) = @_; |
960
|
|
|
|
|
|
|
|
961
|
|
|
|
|
|
|
return rx_observable->new(sub { |
962
|
76
|
|
|
|
|
126
|
my ($subscriber) = @_; |
963
|
|
|
|
|
|
|
|
964
|
76
|
|
|
|
|
299
|
my %own_subscriber = %$subscriber; |
965
|
76
|
|
|
|
|
166
|
delete $own_subscriber{next}; |
966
|
|
|
|
|
|
|
|
967
|
76
|
|
|
|
|
216
|
$source->subscribe(\%own_subscriber); |
968
|
|
|
|
|
|
|
|
969
|
76
|
|
|
|
|
192
|
return; |
970
|
68
|
|
|
|
|
133
|
}); |
971
|
|
|
|
|
|
|
} |
972
|
68
|
|
|
68
|
|
336
|
} |
973
|
|
|
|
|
|
|
|
974
|
|
|
|
|
|
|
sub op_is_empty { |
975
|
|
|
|
|
|
|
return sub { |
976
|
2
|
|
|
2
|
|
6
|
my ($source) = @_; |
977
|
|
|
|
|
|
|
|
978
|
2
|
|
|
|
|
5
|
return $source->pipe( |
979
|
|
|
|
|
|
|
op_first(), |
980
|
|
|
|
|
|
|
op_map_to(0), |
981
|
|
|
|
|
|
|
op_default_if_empty(1), |
982
|
|
|
|
|
|
|
); |
983
|
2
|
|
|
2
|
|
12
|
}; |
984
|
|
|
|
|
|
|
} |
985
|
|
|
|
|
|
|
|
986
|
|
|
|
|
|
|
sub op_last { |
987
|
3
|
|
|
3
|
|
7
|
my ($predicate, $default) = @_; |
988
|
3
|
|
|
|
|
7
|
my $has_default = @_ >= 2; |
989
|
|
|
|
|
|
|
|
990
|
|
|
|
|
|
|
return sub { |
991
|
3
|
|
|
3
|
|
6
|
my ($source) = @_; |
992
|
|
|
|
|
|
|
|
993
|
|
|
|
|
|
|
return rx_observable->new(sub { |
994
|
3
|
|
|
|
|
6
|
my ($subscriber) = @_; |
995
|
|
|
|
|
|
|
|
996
|
3
|
|
|
|
|
4
|
my $last_val; |
997
|
|
|
|
|
|
|
my $last_val_obtained; |
998
|
|
|
|
|
|
|
|
999
|
3
|
|
|
|
|
7
|
my $idx = 0; |
1000
|
|
|
|
|
|
|
my $own_subscriber = { |
1001
|
|
|
|
|
|
|
%$subscriber, |
1002
|
|
|
|
|
|
|
next => sub { |
1003
|
5
|
|
|
|
|
12
|
my ($v) = @_; |
1004
|
|
|
|
|
|
|
|
1005
|
5
|
100
|
|
|
|
12
|
if ($predicate) { |
1006
|
3
|
|
|
|
|
4
|
my $passes; |
1007
|
3
|
|
|
|
|
5
|
my $ok = eval { local $_ = $v; $passes = $predicate->($v, $idx++); 1 }; |
|
3
|
|
|
|
|
5
|
|
|
3
|
|
|
|
|
9
|
|
|
3
|
|
|
|
|
16
|
|
1008
|
3
|
50
|
|
|
|
8
|
$ok or do { |
1009
|
0
|
0
|
|
|
|
0
|
$subscriber->{error}->($@) if defined $subscriber->{error}; |
1010
|
0
|
|
|
|
|
0
|
return; |
1011
|
|
|
|
|
|
|
}; |
1012
|
3
|
100
|
|
|
|
8
|
if ($passes) { |
1013
|
1
|
|
|
|
|
3
|
$last_val = $v; |
1014
|
1
|
|
|
|
|
3
|
$last_val_obtained = 1; |
1015
|
|
|
|
|
|
|
} |
1016
|
|
|
|
|
|
|
} else { |
1017
|
2
|
|
|
|
|
4
|
$last_val = $v; |
1018
|
2
|
|
|
|
|
5
|
$last_val_obtained = 1; |
1019
|
|
|
|
|
|
|
} |
1020
|
|
|
|
|
|
|
}, |
1021
|
|
|
|
|
|
|
complete => sub { |
1022
|
3
|
100
|
|
|
|
9
|
if (! $last_val_obtained) { |
1023
|
1
|
50
|
|
|
|
4
|
if ($has_default) { |
1024
|
1
|
50
|
|
|
|
13
|
$subscriber->{next}->($default) if defined $subscriber->{next}; |
1025
|
1
|
50
|
|
|
|
4
|
$subscriber->{complete}->() if defined $subscriber->{complete}; |
1026
|
|
|
|
|
|
|
} else { |
1027
|
0
|
|
|
|
|
0
|
$subscriber->{error}->("no last value found"); |
1028
|
|
|
|
|
|
|
} |
1029
|
|
|
|
|
|
|
} else { |
1030
|
2
|
50
|
|
|
|
11
|
$subscriber->{next}->($last_val) if defined $subscriber->{next}; |
1031
|
2
|
50
|
|
|
|
11
|
$subscriber->{complete}->() if defined $subscriber->{complete}; |
1032
|
|
|
|
|
|
|
} |
1033
|
|
|
|
|
|
|
}, |
1034
|
3
|
|
|
|
|
21
|
}; |
1035
|
|
|
|
|
|
|
|
1036
|
3
|
|
|
|
|
13
|
$source->subscribe($own_subscriber); |
1037
|
|
|
|
|
|
|
|
1038
|
3
|
|
|
|
|
8
|
return; |
1039
|
3
|
|
|
|
|
8
|
}); |
1040
|
3
|
|
|
|
|
15
|
}; |
1041
|
|
|
|
|
|
|
} |
1042
|
|
|
|
|
|
|
|
1043
|
|
|
|
|
|
|
sub op_map { |
1044
|
28
|
|
|
28
|
|
63
|
my ($mapping_sub) = @_; |
1045
|
|
|
|
|
|
|
|
1046
|
|
|
|
|
|
|
return sub { |
1047
|
28
|
|
|
28
|
|
49
|
my ($source) = @_; |
1048
|
|
|
|
|
|
|
|
1049
|
|
|
|
|
|
|
return rx_observable->new(sub { |
1050
|
19
|
|
|
|
|
36
|
my ($subscriber) = @_; |
1051
|
|
|
|
|
|
|
|
1052
|
19
|
|
|
|
|
71
|
my $own_subscriber = { %$subscriber }; |
1053
|
19
|
|
|
|
|
41
|
my $idx = 0; |
1054
|
|
|
|
|
|
|
$own_subscriber->{next} = sub { |
1055
|
96
|
|
|
|
|
183
|
my ($value) = @_; |
1056
|
96
|
|
|
|
|
147
|
my $result = eval { |
1057
|
96
|
|
|
|
|
146
|
local $_ = $value; |
1058
|
96
|
|
|
|
|
233
|
$mapping_sub->($value, $idx++); |
1059
|
|
|
|
|
|
|
}; |
1060
|
96
|
50
|
|
|
|
377
|
if (my $error = $@) { |
1061
|
0
|
0
|
|
|
|
0
|
$subscriber->{error}->($error) if defined $subscriber->{error}; |
1062
|
|
|
|
|
|
|
} else { |
1063
|
96
|
50
|
|
|
|
283
|
$subscriber->{next}->($result) if defined $subscriber->{next}; |
1064
|
|
|
|
|
|
|
} |
1065
|
19
|
|
|
|
|
75
|
}; |
1066
|
|
|
|
|
|
|
|
1067
|
19
|
|
|
|
|
66
|
$source->subscribe($own_subscriber); |
1068
|
|
|
|
|
|
|
|
1069
|
19
|
|
|
|
|
58
|
return; |
1070
|
28
|
|
|
|
|
58
|
}); |
1071
|
28
|
|
|
|
|
132
|
}; |
1072
|
|
|
|
|
|
|
} |
1073
|
|
|
|
|
|
|
|
1074
|
|
|
|
|
|
|
sub op_map_to { |
1075
|
2
|
|
|
2
|
|
5
|
my ($mapping_value) = @_; |
1076
|
|
|
|
|
|
|
|
1077
|
|
|
|
|
|
|
return sub { |
1078
|
2
|
|
|
2
|
|
5
|
my ($source) = @_; |
1079
|
|
|
|
|
|
|
|
1080
|
|
|
|
|
|
|
return rx_observable->new(sub { |
1081
|
2
|
|
|
|
|
5
|
my ($subscriber) = @_; |
1082
|
|
|
|
|
|
|
|
1083
|
2
|
|
|
|
|
8
|
my $own_subscriber = { %$subscriber }; |
1084
|
|
|
|
|
|
|
$own_subscriber->{next} &&= sub { |
1085
|
1
|
50
|
|
|
|
7
|
$subscriber->{next}->($mapping_value) if defined $subscriber->{next}; |
1086
|
2
|
|
50
|
|
|
16
|
}; |
1087
|
|
|
|
|
|
|
|
1088
|
2
|
|
|
|
|
7
|
$source->subscribe($own_subscriber); |
1089
|
|
|
|
|
|
|
|
1090
|
2
|
|
|
|
|
7
|
return; |
1091
|
2
|
|
|
|
|
6
|
}); |
1092
|
2
|
|
|
|
|
8
|
}; |
1093
|
|
|
|
|
|
|
} |
1094
|
|
|
|
|
|
|
|
1095
|
|
|
|
|
|
|
sub op_max { |
1096
|
0
|
|
|
0
|
|
0
|
my ($comparer) = @_; |
1097
|
|
|
|
|
|
|
|
1098
|
|
|
|
|
|
|
return sub { |
1099
|
0
|
|
|
0
|
|
0
|
my ($source) = @_; |
1100
|
|
|
|
|
|
|
|
1101
|
|
|
|
|
|
|
return rx_observable->new(sub { |
1102
|
0
|
|
|
|
|
0
|
my ($subscriber) = @_; |
1103
|
|
|
|
|
|
|
|
1104
|
0
|
|
|
|
|
0
|
my $curr_max; |
1105
|
|
|
|
|
|
|
my $has_curr_max; |
1106
|
|
|
|
|
|
|
|
1107
|
|
|
|
|
|
|
my $own_subscriber = { |
1108
|
|
|
|
|
|
|
%$subscriber, |
1109
|
|
|
|
|
|
|
next => sub { |
1110
|
0
|
|
|
|
|
0
|
my ($v) = @_; |
1111
|
|
|
|
|
|
|
|
1112
|
0
|
0
|
|
|
|
0
|
if (!$has_curr_max) { |
1113
|
0
|
|
|
|
|
0
|
$curr_max = $v; |
1114
|
0
|
|
|
|
|
0
|
$has_curr_max = 1; |
1115
|
|
|
|
|
|
|
} |
1116
|
|
|
|
|
|
|
else { |
1117
|
0
|
0
|
|
|
|
0
|
if (!$comparer) { |
1118
|
0
|
0
|
|
|
|
0
|
if ($v > $curr_max) { |
1119
|
0
|
|
|
|
|
0
|
$curr_max = $v; |
1120
|
|
|
|
|
|
|
} |
1121
|
|
|
|
|
|
|
} |
1122
|
|
|
|
|
|
|
else { |
1123
|
0
|
0
|
|
|
|
0
|
if ($comparer->($v, $curr_max) > 0) { |
1124
|
0
|
|
|
|
|
0
|
$curr_max = $v; |
1125
|
|
|
|
|
|
|
} |
1126
|
|
|
|
|
|
|
} |
1127
|
|
|
|
|
|
|
} |
1128
|
|
|
|
|
|
|
}, |
1129
|
|
|
|
|
|
|
complete => sub { |
1130
|
0
|
0
|
|
|
|
0
|
if ($has_curr_max) { |
1131
|
0
|
0
|
|
|
|
0
|
$subscriber->{next}->($curr_max) if defined $subscriber->{next}; |
1132
|
|
|
|
|
|
|
} |
1133
|
0
|
0
|
|
|
|
0
|
$subscriber->{complete}->() if defined $subscriber->{complete}; |
1134
|
|
|
|
|
|
|
}, |
1135
|
0
|
|
|
|
|
0
|
}; |
1136
|
|
|
|
|
|
|
|
1137
|
0
|
|
|
|
|
0
|
$source->subscribe($own_subscriber); |
1138
|
|
|
|
|
|
|
|
1139
|
0
|
|
|
|
|
0
|
return; |
1140
|
0
|
|
|
|
|
0
|
}); |
1141
|
0
|
|
|
|
|
0
|
}; |
1142
|
|
|
|
|
|
|
} |
1143
|
|
|
|
|
|
|
|
1144
|
|
|
|
|
|
|
sub _op_merge_all_make_subscriber { |
1145
|
13
|
|
|
13
|
|
29
|
my ($small_subscriptions, $subscriber, $stored_observables, $big_completed_ref) = @_; |
1146
|
|
|
|
|
|
|
|
1147
|
13
|
|
|
|
|
37
|
my $small_subscription = RxPerl::Subscription->new; |
1148
|
13
|
|
|
|
|
39
|
$small_subscriptions->{$small_subscription} = $small_subscription; |
1149
|
|
|
|
|
|
|
return { |
1150
|
|
|
|
|
|
|
new_subscription => $small_subscription, |
1151
|
|
|
|
|
|
|
next => sub { |
1152
|
28
|
50
|
|
28
|
|
97
|
$subscriber->{next}->(@_) if defined $subscriber->{next}; |
1153
|
|
|
|
|
|
|
}, |
1154
|
|
|
|
|
|
|
error => sub { |
1155
|
0
|
0
|
|
0
|
|
0
|
$subscriber->{error}->(@_) if defined $subscriber->{error}; |
1156
|
|
|
|
|
|
|
}, |
1157
|
|
|
|
|
|
|
complete => sub { |
1158
|
11
|
|
|
11
|
|
65
|
delete $small_subscriptions->{$small_subscription}; |
1159
|
11
|
100
|
66
|
|
|
46
|
if (@$stored_observables) { |
|
|
100
|
|
|
|
|
|
1160
|
8
|
|
|
|
|
16
|
my $new_obs = shift @$stored_observables; |
1161
|
8
|
|
|
|
|
22
|
$new_obs->subscribe( |
1162
|
|
|
|
|
|
|
_op_merge_all_make_subscriber( |
1163
|
|
|
|
|
|
|
$small_subscriptions, |
1164
|
|
|
|
|
|
|
$subscriber, |
1165
|
|
|
|
|
|
|
$stored_observables, |
1166
|
|
|
|
|
|
|
$big_completed_ref, |
1167
|
|
|
|
|
|
|
), |
1168
|
|
|
|
|
|
|
); |
1169
|
|
|
|
|
|
|
} elsif ($$big_completed_ref and !%$small_subscriptions) { |
1170
|
1
|
50
|
|
|
|
5
|
$subscriber->{complete}->() if defined $subscriber->{complete}; |
1171
|
|
|
|
|
|
|
} |
1172
|
|
|
|
|
|
|
}, |
1173
|
13
|
|
|
|
|
113
|
}; |
1174
|
|
|
|
|
|
|
} |
1175
|
|
|
|
|
|
|
|
1176
|
|
|
|
|
|
|
sub op_merge_all { |
1177
|
4
|
|
|
4
|
|
9
|
my ($concurrent) = @_; |
1178
|
|
|
|
|
|
|
|
1179
|
|
|
|
|
|
|
return sub { |
1180
|
4
|
|
|
4
|
|
20
|
my ($source) = @_; |
1181
|
|
|
|
|
|
|
|
1182
|
|
|
|
|
|
|
return rx_observable->new(sub { |
1183
|
4
|
|
|
|
|
14
|
my ($subscriber) = @_; |
1184
|
|
|
|
|
|
|
|
1185
|
4
|
|
|
|
|
16
|
my @stored_observables; |
1186
|
|
|
|
|
|
|
my %small_subscriptions; |
1187
|
4
|
|
|
|
|
0
|
my $big_completed; |
1188
|
|
|
|
|
|
|
|
1189
|
4
|
|
|
|
|
11
|
my $own_subscription = RxPerl::Subscription->new; |
1190
|
4
|
|
|
|
|
15
|
$subscriber->subscription->add( |
1191
|
|
|
|
|
|
|
$own_subscription, |
1192
|
|
|
|
|
|
|
\%small_subscriptions, |
1193
|
|
|
|
|
|
|
); |
1194
|
|
|
|
|
|
|
|
1195
|
|
|
|
|
|
|
my $own_subscriber = { |
1196
|
|
|
|
|
|
|
new_subscription => $own_subscription, |
1197
|
|
|
|
|
|
|
next => sub { |
1198
|
26
|
|
|
|
|
51
|
my ($new_observable) = @_; |
1199
|
|
|
|
|
|
|
|
1200
|
26
|
|
|
|
|
47
|
push @stored_observables, $new_observable; |
1201
|
26
|
100
|
100
|
|
|
181
|
if (!defined $concurrent or keys(%small_subscriptions) < $concurrent) { |
1202
|
5
|
|
|
|
|
10
|
my $new_obs = shift @stored_observables; |
1203
|
5
|
|
|
|
|
22
|
$new_obs->subscribe( |
1204
|
|
|
|
|
|
|
_op_merge_all_make_subscriber( |
1205
|
|
|
|
|
|
|
\%small_subscriptions, |
1206
|
|
|
|
|
|
|
$subscriber, |
1207
|
|
|
|
|
|
|
\@stored_observables, |
1208
|
|
|
|
|
|
|
\$big_completed, |
1209
|
|
|
|
|
|
|
), |
1210
|
|
|
|
|
|
|
); |
1211
|
|
|
|
|
|
|
} |
1212
|
|
|
|
|
|
|
}, |
1213
|
|
|
|
|
|
|
error => sub { |
1214
|
0
|
0
|
|
|
|
0
|
$subscriber->{error}->(@_) if defined $subscriber->{error}; |
1215
|
|
|
|
|
|
|
}, |
1216
|
|
|
|
|
|
|
complete => sub { |
1217
|
2
|
|
|
|
|
7
|
$big_completed = 1; |
1218
|
2
|
100
|
66
|
|
|
11
|
$subscriber->{complete}->() if !%small_subscriptions and defined $subscriber->{complete}; |
1219
|
|
|
|
|
|
|
}, |
1220
|
4
|
|
|
|
|
37
|
}; |
1221
|
|
|
|
|
|
|
|
1222
|
4
|
|
|
|
|
18
|
$source->subscribe($own_subscriber); |
1223
|
|
|
|
|
|
|
|
1224
|
4
|
|
|
|
|
18
|
return; |
1225
|
4
|
|
|
|
|
10
|
}); |
1226
|
4
|
|
|
|
|
23
|
}; |
1227
|
|
|
|
|
|
|
} |
1228
|
|
|
|
|
|
|
|
1229
|
|
|
|
|
|
|
sub op_merge_map { |
1230
|
1
|
|
|
1
|
|
3
|
my ($observable_factory) = @_; |
1231
|
|
|
|
|
|
|
|
1232
|
|
|
|
|
|
|
return sub { |
1233
|
1
|
|
|
1
|
|
3
|
my ($source) = @_; |
1234
|
|
|
|
|
|
|
|
1235
|
1
|
|
|
|
|
5
|
return $source->pipe( |
1236
|
|
|
|
|
|
|
op_map($observable_factory), |
1237
|
|
|
|
|
|
|
op_merge_all(), |
1238
|
|
|
|
|
|
|
); |
1239
|
1
|
|
|
|
|
6
|
}; |
1240
|
|
|
|
|
|
|
} |
1241
|
|
|
|
|
|
|
|
1242
|
|
|
|
|
|
|
sub op_merge_with { |
1243
|
1
|
|
|
1
|
|
5
|
my @other_sources = @_; |
1244
|
|
|
|
|
|
|
|
1245
|
|
|
|
|
|
|
return sub { |
1246
|
1
|
|
|
1
|
|
16
|
my ($source) = @_; |
1247
|
|
|
|
|
|
|
|
1248
|
1
|
|
|
|
|
7
|
return rx_merge( |
1249
|
|
|
|
|
|
|
$source, |
1250
|
|
|
|
|
|
|
@other_sources, |
1251
|
|
|
|
|
|
|
); |
1252
|
1
|
|
|
|
|
8
|
}; |
1253
|
|
|
|
|
|
|
} |
1254
|
|
|
|
|
|
|
|
1255
|
|
|
|
|
|
|
sub op_min { |
1256
|
0
|
|
|
0
|
|
0
|
my ($comparer) = @_; |
1257
|
|
|
|
|
|
|
|
1258
|
|
|
|
|
|
|
return sub { |
1259
|
0
|
|
|
0
|
|
0
|
my ($source) = @_; |
1260
|
|
|
|
|
|
|
|
1261
|
|
|
|
|
|
|
return rx_observable->new(sub { |
1262
|
0
|
|
|
|
|
0
|
my ($subscriber) = @_; |
1263
|
|
|
|
|
|
|
|
1264
|
0
|
|
|
|
|
0
|
my $curr_min; |
1265
|
|
|
|
|
|
|
my $has_curr_min; |
1266
|
|
|
|
|
|
|
|
1267
|
|
|
|
|
|
|
my $own_subscriber = { |
1268
|
|
|
|
|
|
|
%$subscriber, |
1269
|
|
|
|
|
|
|
next => sub { |
1270
|
0
|
|
|
|
|
0
|
my ($v) = @_; |
1271
|
|
|
|
|
|
|
|
1272
|
0
|
0
|
|
|
|
0
|
if (!$has_curr_min) { |
1273
|
0
|
|
|
|
|
0
|
$curr_min = $v; |
1274
|
0
|
|
|
|
|
0
|
$has_curr_min = 1; |
1275
|
|
|
|
|
|
|
} |
1276
|
|
|
|
|
|
|
else { |
1277
|
0
|
0
|
|
|
|
0
|
if (!$comparer) { |
1278
|
0
|
0
|
|
|
|
0
|
if ($v < $curr_min) { |
1279
|
0
|
|
|
|
|
0
|
$curr_min = $v; |
1280
|
|
|
|
|
|
|
} |
1281
|
|
|
|
|
|
|
} |
1282
|
|
|
|
|
|
|
else { |
1283
|
0
|
0
|
|
|
|
0
|
if ($comparer->($v, $curr_min) < 0) { |
1284
|
0
|
|
|
|
|
0
|
$curr_min = $v; |
1285
|
|
|
|
|
|
|
} |
1286
|
|
|
|
|
|
|
} |
1287
|
|
|
|
|
|
|
} |
1288
|
|
|
|
|
|
|
}, |
1289
|
|
|
|
|
|
|
complete => sub { |
1290
|
0
|
0
|
|
|
|
0
|
if ($has_curr_min) { |
1291
|
0
|
0
|
|
|
|
0
|
$subscriber->{next}->($curr_min) if defined $subscriber->{next}; |
1292
|
|
|
|
|
|
|
} |
1293
|
0
|
0
|
|
|
|
0
|
$subscriber->{complete}->() if defined $subscriber->{complete}; |
1294
|
|
|
|
|
|
|
}, |
1295
|
0
|
|
|
|
|
0
|
}; |
1296
|
|
|
|
|
|
|
|
1297
|
0
|
|
|
|
|
0
|
$source->subscribe($own_subscriber); |
1298
|
|
|
|
|
|
|
|
1299
|
0
|
|
|
|
|
0
|
return; |
1300
|
0
|
|
|
|
|
0
|
}); |
1301
|
0
|
|
|
|
|
0
|
}; |
1302
|
|
|
|
|
|
|
} |
1303
|
|
|
|
|
|
|
|
1304
|
|
|
|
|
|
|
sub op_multicast { |
1305
|
0
|
|
|
0
|
|
0
|
my ($subject_factory) = @_; |
1306
|
|
|
|
|
|
|
|
1307
|
|
|
|
|
|
|
return sub { |
1308
|
0
|
|
|
0
|
|
0
|
my ($source) = @_; |
1309
|
|
|
|
|
|
|
|
1310
|
0
|
|
|
|
|
0
|
return RxPerl::ConnectableObservable->new($source, $subject_factory); |
1311
|
0
|
|
|
|
|
0
|
}; |
1312
|
|
|
|
|
|
|
} |
1313
|
|
|
|
|
|
|
|
1314
|
|
|
|
|
|
|
sub op_on_error_resume_next_with { |
1315
|
1
|
|
|
1
|
|
9
|
my @other_sources = @_; |
1316
|
|
|
|
|
|
|
|
1317
|
|
|
|
|
|
|
return sub { |
1318
|
1
|
|
|
1
|
|
6
|
my ($source) = @_; |
1319
|
|
|
|
|
|
|
|
1320
|
1
|
|
|
|
|
15
|
return rx_on_error_resume_next( |
1321
|
|
|
|
|
|
|
$source, |
1322
|
|
|
|
|
|
|
@other_sources, |
1323
|
|
|
|
|
|
|
); |
1324
|
1
|
|
|
|
|
8
|
}; |
1325
|
|
|
|
|
|
|
} |
1326
|
|
|
|
|
|
|
|
1327
|
|
|
|
|
|
|
sub op_pairwise { |
1328
|
|
|
|
|
|
|
return sub { |
1329
|
0
|
|
|
0
|
|
0
|
my ($source) = @_; |
1330
|
|
|
|
|
|
|
|
1331
|
|
|
|
|
|
|
return rx_observable->new(sub { |
1332
|
0
|
|
|
|
|
0
|
my ($subscriber) = @_; |
1333
|
|
|
|
|
|
|
|
1334
|
0
|
|
|
|
|
0
|
my $prev_value; |
1335
|
0
|
|
|
|
|
0
|
my $have_prev_value = 0; |
1336
|
|
|
|
|
|
|
|
1337
|
|
|
|
|
|
|
my $own_subscriber = { |
1338
|
|
|
|
|
|
|
%$subscriber, |
1339
|
|
|
|
|
|
|
( |
1340
|
|
|
|
|
|
|
next => sub { |
1341
|
0
|
|
|
|
|
0
|
my ($value) = @_; |
1342
|
|
|
|
|
|
|
|
1343
|
0
|
0
|
|
|
|
0
|
if ($have_prev_value) { |
1344
|
0
|
0
|
|
|
|
0
|
$subscriber->{next}->([$prev_value, $value]) if defined $subscriber->{next}; |
1345
|
|
|
|
|
|
|
} else { |
1346
|
0
|
|
|
|
|
0
|
$have_prev_value = 1; |
1347
|
|
|
|
|
|
|
} |
1348
|
|
|
|
|
|
|
|
1349
|
0
|
|
|
|
|
0
|
$prev_value = $value; |
1350
|
|
|
|
|
|
|
} |
1351
|
|
|
|
|
|
|
) x!! defined $subscriber->{next}, |
1352
|
0
|
|
|
|
|
0
|
}; |
1353
|
|
|
|
|
|
|
|
1354
|
0
|
|
|
|
|
0
|
$source->subscribe($own_subscriber); |
1355
|
|
|
|
|
|
|
|
1356
|
0
|
|
|
|
|
0
|
return; |
1357
|
0
|
|
|
|
|
0
|
}); |
1358
|
0
|
|
|
0
|
|
0
|
}; |
1359
|
|
|
|
|
|
|
} |
1360
|
|
|
|
|
|
|
|
1361
|
|
|
|
|
|
|
sub op_pluck { |
1362
|
1
|
|
|
1
|
|
4
|
my (@keys) = @_; |
1363
|
|
|
|
|
|
|
|
1364
|
1
|
50
|
|
|
|
6
|
croak 'List of properties cannot be empty,' unless @keys; |
1365
|
|
|
|
|
|
|
|
1366
|
|
|
|
|
|
|
return sub { |
1367
|
1
|
|
|
1
|
|
4
|
my ($source) = @_; |
1368
|
|
|
|
|
|
|
|
1369
|
|
|
|
|
|
|
return rx_observable->new(sub { |
1370
|
1
|
|
|
|
|
6
|
my ($subscriber) = @_; |
1371
|
|
|
|
|
|
|
|
1372
|
|
|
|
|
|
|
my $own_subscriber = { |
1373
|
|
|
|
|
|
|
%$subscriber, |
1374
|
|
|
|
|
|
|
next => sub { |
1375
|
5
|
|
|
|
|
14
|
my (@value) = @_; |
1376
|
|
|
|
|
|
|
|
1377
|
5
|
50
|
|
|
|
11
|
if (! @value) { |
1378
|
0
|
0
|
|
|
|
0
|
$subscriber->{next}->() if defined $subscriber->{next}; |
1379
|
0
|
|
|
|
|
0
|
return; |
1380
|
|
|
|
|
|
|
} |
1381
|
|
|
|
|
|
|
|
1382
|
5
|
|
|
|
|
8
|
my $cursor = $value[0]; |
1383
|
5
|
|
|
|
|
9
|
foreach my $key (@keys) { |
1384
|
7
|
100
|
100
|
|
|
35
|
if ((reftype($cursor) // '') eq 'HASH' and exists $cursor->{$key}) { |
|
|
|
100
|
|
|
|
|
1385
|
4
|
|
|
|
|
12
|
$cursor = $cursor->{$key}; |
1386
|
|
|
|
|
|
|
} else { |
1387
|
3
|
50
|
|
|
|
14
|
$subscriber->{next}->(undef) if defined $subscriber->{next}; |
1388
|
3
|
|
|
|
|
10
|
return; |
1389
|
|
|
|
|
|
|
} |
1390
|
|
|
|
|
|
|
} |
1391
|
|
|
|
|
|
|
|
1392
|
2
|
50
|
|
|
|
11
|
$subscriber->{next}->($cursor) if defined $subscriber->{next}; |
1393
|
|
|
|
|
|
|
}, |
1394
|
1
|
|
|
|
|
10
|
}; |
1395
|
|
|
|
|
|
|
|
1396
|
1
|
|
|
|
|
5
|
$source->subscribe($own_subscriber); |
1397
|
1
|
|
|
|
|
4
|
}); |
1398
|
1
|
|
|
|
|
14
|
}; |
1399
|
|
|
|
|
|
|
} |
1400
|
|
|
|
|
|
|
|
1401
|
|
|
|
|
|
|
sub op_race_with { |
1402
|
1
|
|
|
1
|
|
4
|
my @other_sources = @_; |
1403
|
|
|
|
|
|
|
|
1404
|
|
|
|
|
|
|
return sub { |
1405
|
1
|
|
|
1
|
|
2
|
my ($source) = @_; |
1406
|
|
|
|
|
|
|
|
1407
|
1
|
|
|
|
|
5
|
return rx_race( |
1408
|
|
|
|
|
|
|
$source, |
1409
|
|
|
|
|
|
|
@other_sources, |
1410
|
|
|
|
|
|
|
); |
1411
|
1
|
|
|
|
|
7
|
}; |
1412
|
|
|
|
|
|
|
} |
1413
|
|
|
|
|
|
|
|
1414
|
|
|
|
|
|
|
sub op_reduce { |
1415
|
4
|
|
|
4
|
|
23
|
my ($accumulator, @seed) = @_; |
1416
|
|
|
|
|
|
|
|
1417
|
|
|
|
|
|
|
return sub { |
1418
|
4
|
|
|
4
|
|
8
|
my ($source) = @_; |
1419
|
|
|
|
|
|
|
|
1420
|
|
|
|
|
|
|
return rx_observable->new(sub { |
1421
|
4
|
|
|
|
|
13
|
my ($subscriber) = @_; |
1422
|
|
|
|
|
|
|
|
1423
|
4
|
|
|
|
|
9
|
my $got_first = @seed; |
1424
|
4
|
|
|
|
|
10
|
my $acc; |
1425
|
4
|
100
|
|
|
|
11
|
$acc = $seed[0] if $got_first; |
1426
|
|
|
|
|
|
|
|
1427
|
4
|
|
|
|
|
8
|
my $idx = 0; |
1428
|
|
|
|
|
|
|
my $own_subscriber = { |
1429
|
|
|
|
|
|
|
%$subscriber, |
1430
|
|
|
|
|
|
|
next => sub { |
1431
|
15
|
|
|
|
|
29
|
my ($v) = @_; |
1432
|
|
|
|
|
|
|
|
1433
|
15
|
100
|
|
|
|
34
|
if ($got_first) { |
1434
|
14
|
|
|
|
|
26
|
my $ok = eval { $acc = $accumulator->($acc, $v, $idx++); 1 }; |
|
14
|
|
|
|
|
37
|
|
|
14
|
|
|
|
|
59
|
|
1435
|
14
|
50
|
33
|
|
|
58
|
$ok or $subscriber->{error}->($@) if defined $subscriber->{error}; |
1436
|
|
|
|
|
|
|
} else { |
1437
|
1
|
|
|
|
|
2
|
$acc = $v; |
1438
|
1
|
|
|
|
|
4
|
$got_first = 1; |
1439
|
|
|
|
|
|
|
} |
1440
|
|
|
|
|
|
|
}, |
1441
|
|
|
|
|
|
|
complete => sub { |
1442
|
4
|
50
|
33
|
|
|
36
|
$subscriber->{next}->($acc) if $got_first and defined $subscriber->{next}; |
1443
|
4
|
50
|
|
|
|
17
|
$subscriber->{complete}->() if defined $subscriber->{complete}; |
1444
|
|
|
|
|
|
|
}, |
1445
|
4
|
|
|
|
|
32
|
}; |
1446
|
|
|
|
|
|
|
|
1447
|
4
|
|
|
|
|
19
|
$source->subscribe($own_subscriber); |
1448
|
|
|
|
|
|
|
|
1449
|
4
|
|
|
|
|
14
|
return; |
1450
|
4
|
|
|
|
|
14
|
}); |
1451
|
4
|
|
|
|
|
24
|
}; |
1452
|
|
|
|
|
|
|
} |
1453
|
|
|
|
|
|
|
|
1454
|
|
|
|
|
|
|
sub op_ref_count { |
1455
|
|
|
|
|
|
|
return sub { |
1456
|
0
|
|
|
0
|
|
0
|
my ($source) = @_; |
1457
|
|
|
|
|
|
|
|
1458
|
0
|
0
|
|
|
|
0
|
croak 'op_ref_count() was not applied to a connectable observable' |
1459
|
|
|
|
|
|
|
unless $source->isa('RxPerl::ConnectableObservable'); |
1460
|
|
|
|
|
|
|
|
1461
|
0
|
|
|
|
|
0
|
my $count = 0; |
1462
|
|
|
|
|
|
|
|
1463
|
0
|
|
|
|
|
0
|
my $connection_subscription; |
1464
|
|
|
|
|
|
|
my $typical_unsubscription_fn = sub { |
1465
|
0
|
0
|
|
|
|
0
|
if (--$count == 0) { |
1466
|
0
|
|
|
|
|
0
|
$connection_subscription->unsubscribe; |
1467
|
|
|
|
|
|
|
} |
1468
|
0
|
|
|
|
|
0
|
}; |
1469
|
|
|
|
|
|
|
|
1470
|
|
|
|
|
|
|
return rx_observable->new(sub { |
1471
|
0
|
|
|
|
|
0
|
my ($subscriber) = @_; |
1472
|
|
|
|
|
|
|
|
1473
|
0
|
|
|
|
|
0
|
my $count_was = $count++; |
1474
|
|
|
|
|
|
|
|
1475
|
0
|
0
|
|
|
|
0
|
if ($count_was == 0) { |
1476
|
0
|
|
|
|
|
0
|
$connection_subscription = RxPerl::Subscription->new; |
1477
|
|
|
|
|
|
|
|
1478
|
0
|
|
|
|
|
0
|
$subscriber->subscription->add($typical_unsubscription_fn); |
1479
|
0
|
|
|
|
|
0
|
$source->subscribe($subscriber); |
1480
|
|
|
|
|
|
|
|
1481
|
0
|
|
|
|
|
0
|
$connection_subscription = $source->connect; |
1482
|
|
|
|
|
|
|
} else { |
1483
|
0
|
|
|
|
|
0
|
$subscriber->subscription->add($typical_unsubscription_fn); |
1484
|
0
|
|
|
|
|
0
|
$source->subscribe($subscriber); |
1485
|
|
|
|
|
|
|
} |
1486
|
|
|
|
|
|
|
|
1487
|
0
|
|
|
|
|
0
|
return; |
1488
|
0
|
|
|
|
|
0
|
}); |
1489
|
0
|
|
|
0
|
|
0
|
}; |
1490
|
|
|
|
|
|
|
} |
1491
|
|
|
|
|
|
|
|
1492
|
|
|
|
|
|
|
sub _op_repeat_helper { |
1493
|
8
|
|
|
8
|
|
19
|
my ($subscriber, $source, $count_ref, $own_subscription_ref) = @_; |
1494
|
|
|
|
|
|
|
|
1495
|
8
|
|
|
|
|
20
|
my $own_subscription = RxPerl::Subscription->new; |
1496
|
8
|
|
|
|
|
12
|
$$own_subscription_ref = $own_subscription; |
1497
|
|
|
|
|
|
|
my $own_subscriber = { |
1498
|
|
|
|
|
|
|
new_subscription => $own_subscription, |
1499
|
|
|
|
|
|
|
next => $subscriber->{next}, |
1500
|
|
|
|
|
|
|
error => $subscriber->{error}, |
1501
|
|
|
|
|
|
|
complete => sub { |
1502
|
6
|
100
|
|
6
|
|
17
|
if (--$$count_ref) { |
1503
|
5
|
|
|
|
|
14
|
_op_repeat_helper( |
1504
|
|
|
|
|
|
|
$subscriber, $source, $count_ref, $own_subscription_ref, |
1505
|
|
|
|
|
|
|
); |
1506
|
|
|
|
|
|
|
} else { |
1507
|
1
|
50
|
|
|
|
17
|
$subscriber->{complete}->() if defined $subscriber->{complete}; |
1508
|
|
|
|
|
|
|
} |
1509
|
|
|
|
|
|
|
}, |
1510
|
8
|
|
|
|
|
49
|
}; |
1511
|
|
|
|
|
|
|
|
1512
|
8
|
|
|
|
|
23
|
$source->subscribe($own_subscriber); |
1513
|
|
|
|
|
|
|
} |
1514
|
|
|
|
|
|
|
|
1515
|
|
|
|
|
|
|
sub op_repeat { |
1516
|
3
|
|
|
3
|
|
6
|
my ($count) = @_; |
1517
|
|
|
|
|
|
|
|
1518
|
|
|
|
|
|
|
return sub { |
1519
|
3
|
|
|
3
|
|
10
|
my ($source) = @_; |
1520
|
|
|
|
|
|
|
|
1521
|
|
|
|
|
|
|
return rx_observable->new(sub { |
1522
|
3
|
|
|
|
|
8
|
my ($subscriber) = @_; |
1523
|
|
|
|
|
|
|
|
1524
|
3
|
|
|
|
|
8
|
my $count = $count; |
1525
|
|
|
|
|
|
|
|
1526
|
3
|
50
|
|
|
|
7
|
$count = -1 if ! defined $count; |
1527
|
3
|
50
|
|
|
|
8
|
if ($count == 0) { |
1528
|
0
|
0
|
|
|
|
0
|
$subscriber->{complete}->() if defined $subscriber->{complete}; |
1529
|
0
|
|
|
|
|
0
|
return; |
1530
|
|
|
|
|
|
|
} |
1531
|
|
|
|
|
|
|
|
1532
|
3
|
|
|
|
|
4
|
my $own_subscription; |
1533
|
3
|
|
|
|
|
7
|
my $own_subscription_ref = \$own_subscription; |
1534
|
|
|
|
|
|
|
|
1535
|
3
|
|
|
|
|
8
|
$subscriber->subscription->add( |
1536
|
|
|
|
|
|
|
$own_subscription_ref, |
1537
|
|
|
|
|
|
|
); |
1538
|
|
|
|
|
|
|
|
1539
|
3
|
|
|
|
|
21
|
_op_repeat_helper( |
1540
|
|
|
|
|
|
|
$subscriber, $source, \$count, $own_subscription_ref, |
1541
|
|
|
|
|
|
|
); |
1542
|
|
|
|
|
|
|
|
1543
|
3
|
|
|
|
|
7
|
return; |
1544
|
3
|
|
|
|
|
8
|
}); |
1545
|
3
|
|
|
|
|
20
|
}; |
1546
|
|
|
|
|
|
|
} |
1547
|
|
|
|
|
|
|
|
1548
|
|
|
|
|
|
|
sub _op_retry_helper { |
1549
|
8
|
|
|
8
|
|
19
|
my ($subscriber, $source, $count_ref, $own_subscription_ref) = @_; |
1550
|
|
|
|
|
|
|
|
1551
|
8
|
|
|
|
|
22
|
my $own_subscription = RxPerl::Subscription->new; |
1552
|
8
|
|
|
|
|
17
|
$$own_subscription_ref = $own_subscription; |
1553
|
|
|
|
|
|
|
my $own_subscriber = { |
1554
|
|
|
|
|
|
|
new_subscription => $own_subscription, |
1555
|
|
|
|
|
|
|
next => $subscriber->{next}, |
1556
|
|
|
|
|
|
|
error => sub { |
1557
|
7
|
100
|
|
7
|
|
25
|
if ($$count_ref--) { |
1558
|
6
|
|
|
|
|
14
|
_op_retry_helper( |
1559
|
|
|
|
|
|
|
$subscriber, $source, $count_ref, $own_subscription_ref, |
1560
|
|
|
|
|
|
|
); |
1561
|
|
|
|
|
|
|
} else { |
1562
|
1
|
50
|
|
|
|
21
|
$subscriber->{error}->(@_) if defined $subscriber->{error}; |
1563
|
|
|
|
|
|
|
} |
1564
|
|
|
|
|
|
|
}, |
1565
|
|
|
|
|
|
|
complete => $subscriber->{complete}, |
1566
|
8
|
|
|
|
|
36
|
}; |
1567
|
|
|
|
|
|
|
|
1568
|
8
|
|
|
|
|
20
|
$source->subscribe($own_subscriber); |
1569
|
|
|
|
|
|
|
} |
1570
|
|
|
|
|
|
|
|
1571
|
|
|
|
|
|
|
sub op_retry { |
1572
|
2
|
|
|
2
|
|
6
|
my ($count) = @_; |
1573
|
|
|
|
|
|
|
|
1574
|
|
|
|
|
|
|
return sub { |
1575
|
2
|
|
|
2
|
|
7
|
my ($source) = @_; |
1576
|
|
|
|
|
|
|
|
1577
|
|
|
|
|
|
|
return rx_observable->new(sub { |
1578
|
2
|
|
|
|
|
11
|
my ($subscriber) = @_; |
1579
|
|
|
|
|
|
|
|
1580
|
2
|
|
|
|
|
3
|
my $count = $count; |
1581
|
|
|
|
|
|
|
|
1582
|
2
|
50
|
|
|
|
9
|
$count = -1 if ! defined $count; |
1583
|
|
|
|
|
|
|
|
1584
|
2
|
|
|
|
|
2
|
my $own_subscription; |
1585
|
2
|
|
|
|
|
4
|
my $own_subscription_ref = \$own_subscription; |
1586
|
|
|
|
|
|
|
|
1587
|
2
|
|
|
|
|
6
|
$subscriber->subscription->add( |
1588
|
|
|
|
|
|
|
$own_subscription_ref, |
1589
|
|
|
|
|
|
|
); |
1590
|
|
|
|
|
|
|
|
1591
|
2
|
|
|
|
|
14
|
_op_retry_helper( |
1592
|
|
|
|
|
|
|
$subscriber, $source, \$count, $own_subscription_ref, |
1593
|
|
|
|
|
|
|
); |
1594
|
|
|
|
|
|
|
|
1595
|
2
|
|
|
|
|
5
|
return; |
1596
|
2
|
|
|
|
|
5
|
}); |
1597
|
2
|
|
|
|
|
14
|
}; |
1598
|
|
|
|
|
|
|
} |
1599
|
|
|
|
|
|
|
|
1600
|
|
|
|
|
|
|
sub op_sample { |
1601
|
0
|
|
|
0
|
|
0
|
my ($notifier) = @_; |
1602
|
|
|
|
|
|
|
|
1603
|
|
|
|
|
|
|
return sub { |
1604
|
0
|
|
|
0
|
|
0
|
my ($source) = @_; |
1605
|
|
|
|
|
|
|
|
1606
|
|
|
|
|
|
|
return rx_observable->new(sub { |
1607
|
0
|
|
|
|
|
0
|
my ($subscriber) = @_; |
1608
|
|
|
|
|
|
|
|
1609
|
0
|
|
|
|
|
0
|
my $last_val; |
1610
|
|
|
|
|
|
|
my $has_last_val; |
1611
|
|
|
|
|
|
|
|
1612
|
0
|
|
|
|
|
0
|
my $notifier_subscription = RxPerl::Subscription->new; |
1613
|
|
|
|
|
|
|
my $notifier_subscriber = { |
1614
|
|
|
|
|
|
|
new_subscription => $notifier_subscription, |
1615
|
|
|
|
|
|
|
next => sub { |
1616
|
0
|
0
|
|
|
|
0
|
if ($has_last_val) { |
1617
|
0
|
0
|
|
|
|
0
|
$subscriber->{next}->($last_val) if defined $subscriber->{next}; |
1618
|
0
|
|
|
|
|
0
|
undef $has_last_val; |
1619
|
|
|
|
|
|
|
} |
1620
|
|
|
|
|
|
|
}, |
1621
|
|
|
|
|
|
|
error => sub { |
1622
|
0
|
0
|
|
|
|
0
|
$subscriber->{error}->(@_) if defined $subscriber->{error}; |
1623
|
|
|
|
|
|
|
}, |
1624
|
0
|
|
|
|
|
0
|
}; |
1625
|
|
|
|
|
|
|
|
1626
|
0
|
|
|
|
|
0
|
$subscriber->subscription->add($notifier_subscription); |
1627
|
|
|
|
|
|
|
|
1628
|
|
|
|
|
|
|
my $own_subscriber = { |
1629
|
|
|
|
|
|
|
%$subscriber, |
1630
|
|
|
|
|
|
|
next => sub { |
1631
|
0
|
|
|
|
|
0
|
my ($v) = @_; |
1632
|
|
|
|
|
|
|
|
1633
|
0
|
|
|
|
|
0
|
$last_val = $v; |
1634
|
0
|
|
|
|
|
0
|
$has_last_val = 1; |
1635
|
|
|
|
|
|
|
}, |
1636
|
0
|
|
|
|
|
0
|
}; |
1637
|
|
|
|
|
|
|
|
1638
|
0
|
|
|
|
|
0
|
$notifier->subscribe($notifier_subscriber); |
1639
|
0
|
|
|
|
|
0
|
$source->subscribe($own_subscriber); |
1640
|
|
|
|
|
|
|
|
1641
|
0
|
|
|
|
|
0
|
return; |
1642
|
0
|
|
|
|
|
0
|
}); |
1643
|
0
|
|
|
|
|
0
|
}; |
1644
|
|
|
|
|
|
|
} |
1645
|
|
|
|
|
|
|
|
1646
|
|
|
|
|
|
|
sub op_sample_time { |
1647
|
0
|
|
|
0
|
|
0
|
my ($period) = @_; |
1648
|
|
|
|
|
|
|
|
1649
|
0
|
|
|
|
|
0
|
return op_sample(rx_interval($period)); |
1650
|
|
|
|
|
|
|
} |
1651
|
|
|
|
|
|
|
|
1652
|
|
|
|
|
|
|
sub op_scan { |
1653
|
0
|
|
|
0
|
|
0
|
my ($accumulator_function, $seed) = @_; |
1654
|
0
|
|
|
|
|
0
|
my $has_seed = @_ >= 2; |
1655
|
|
|
|
|
|
|
|
1656
|
|
|
|
|
|
|
return sub { |
1657
|
0
|
|
|
0
|
|
0
|
my ($source) = @_; |
1658
|
|
|
|
|
|
|
|
1659
|
|
|
|
|
|
|
return rx_observable->new(sub { |
1660
|
0
|
|
|
|
|
0
|
my ($subscriber) = @_; |
1661
|
|
|
|
|
|
|
|
1662
|
0
|
|
|
|
|
0
|
my $has_seed = $has_seed; |
1663
|
|
|
|
|
|
|
|
1664
|
0
|
0
|
|
|
|
0
|
my $acc; $acc = $seed if $has_seed; |
|
0
|
|
|
|
|
0
|
|
1665
|
0
|
|
|
|
|
0
|
my $index = -1; |
1666
|
|
|
|
|
|
|
my $own_subscriber = { |
1667
|
|
|
|
|
|
|
%$subscriber, |
1668
|
|
|
|
|
|
|
( |
1669
|
|
|
|
|
|
|
next => sub { |
1670
|
0
|
|
|
|
|
0
|
my ($value) = @_; |
1671
|
|
|
|
|
|
|
|
1672
|
0
|
0
|
|
|
|
0
|
if (! $has_seed) { |
1673
|
0
|
|
|
|
|
0
|
$acc = $value; |
1674
|
0
|
|
|
|
|
0
|
$has_seed = 1; |
1675
|
|
|
|
|
|
|
} else { |
1676
|
0
|
|
|
|
|
0
|
++$index; |
1677
|
0
|
|
|
|
|
0
|
$acc = $accumulator_function->($acc, $value, $index); |
1678
|
|
|
|
|
|
|
} |
1679
|
|
|
|
|
|
|
|
1680
|
0
|
0
|
|
|
|
0
|
$subscriber->{next}->($acc) if defined $subscriber->{next}; |
1681
|
|
|
|
|
|
|
}, |
1682
|
|
|
|
|
|
|
) x!! defined $subscriber->{next}, |
1683
|
0
|
|
|
|
|
0
|
}; |
1684
|
|
|
|
|
|
|
|
1685
|
0
|
|
|
|
|
0
|
$source->subscribe($own_subscriber); |
1686
|
|
|
|
|
|
|
|
1687
|
0
|
|
|
|
|
0
|
return; |
1688
|
0
|
|
|
|
|
0
|
}); |
1689
|
0
|
|
|
|
|
0
|
}; |
1690
|
|
|
|
|
|
|
} |
1691
|
|
|
|
|
|
|
|
1692
|
|
|
|
|
|
|
sub op_share { |
1693
|
|
|
|
|
|
|
return ( |
1694
|
0
|
|
|
0
|
|
0
|
op_multicast(sub { rx_subject->new }), |
|
0
|
|
|
0
|
|
0
|
|
1695
|
|
|
|
|
|
|
op_ref_count(), |
1696
|
|
|
|
|
|
|
); |
1697
|
|
|
|
|
|
|
} |
1698
|
|
|
|
|
|
|
|
1699
|
|
|
|
|
|
|
sub op_single { |
1700
|
2
|
|
|
2
|
|
5
|
my ($predicate) = @_; |
1701
|
|
|
|
|
|
|
|
1702
|
|
|
|
|
|
|
return sub { |
1703
|
2
|
|
|
2
|
|
4
|
my ($source) = @_; |
1704
|
|
|
|
|
|
|
|
1705
|
|
|
|
|
|
|
return rx_observable->new(sub { |
1706
|
2
|
|
|
|
|
5
|
my ($subscriber) = @_; |
1707
|
|
|
|
|
|
|
|
1708
|
2
|
|
|
|
|
4
|
my @found; |
1709
|
|
|
|
|
|
|
|
1710
|
2
|
|
|
|
|
3
|
my $idx = 0; |
1711
|
|
|
|
|
|
|
my $own_subscriber = { |
1712
|
|
|
|
|
|
|
%$subscriber, |
1713
|
|
|
|
|
|
|
next => sub { |
1714
|
3
|
|
|
|
|
7
|
my ($v) = @_; |
1715
|
|
|
|
|
|
|
|
1716
|
3
|
100
|
|
|
|
9
|
if (!$predicate) { |
1717
|
1
|
|
|
|
|
3
|
push @found, $v; |
1718
|
|
|
|
|
|
|
} else { |
1719
|
2
|
|
|
|
|
3
|
my $found; |
1720
|
2
|
|
|
|
|
4
|
my $ok = eval { local $_ = $v; $found = $predicate->($v, $idx++); 1 }; |
|
2
|
|
|
|
|
5
|
|
|
2
|
|
|
|
|
6
|
|
|
2
|
|
|
|
|
12
|
|
1721
|
2
|
50
|
|
|
|
7
|
if (! $ok) { |
1722
|
0
|
0
|
|
|
|
0
|
$subscriber->{error}->($@) if defined $subscriber->{error}; |
1723
|
0
|
|
|
|
|
0
|
return; |
1724
|
|
|
|
|
|
|
} |
1725
|
2
|
100
|
|
|
|
6
|
push @found, $v if $found; |
1726
|
|
|
|
|
|
|
} |
1727
|
|
|
|
|
|
|
|
1728
|
3
|
50
|
33
|
|
|
65
|
$subscriber->{error}->('Too many values match') if @found > 1 and defined $subscriber->{error}; |
1729
|
|
|
|
|
|
|
}, |
1730
|
|
|
|
|
|
|
complete => sub { |
1731
|
2
|
50
|
|
|
|
7
|
if (! @found) { |
1732
|
0
|
0
|
|
|
|
0
|
$subscriber->{error}->('No values match') if defined $subscriber->{error}; |
1733
|
|
|
|
|
|
|
} else { |
1734
|
2
|
50
|
|
|
|
12
|
$subscriber->{next}->($found[0]) if defined $subscriber->{next}; |
1735
|
2
|
50
|
|
|
|
9
|
$subscriber->{complete}->() if defined $subscriber->{complete}; |
1736
|
|
|
|
|
|
|
} |
1737
|
|
|
|
|
|
|
}, |
1738
|
2
|
|
|
|
|
16
|
}; |
1739
|
|
|
|
|
|
|
|
1740
|
2
|
|
|
|
|
9
|
$source->subscribe($own_subscriber); |
1741
|
|
|
|
|
|
|
|
1742
|
2
|
|
|
|
|
6
|
return; |
1743
|
2
|
|
|
|
|
6
|
}); |
1744
|
2
|
|
|
|
|
11
|
}; |
1745
|
|
|
|
|
|
|
} |
1746
|
|
|
|
|
|
|
|
1747
|
|
|
|
|
|
|
sub op_skip { |
1748
|
1
|
|
|
1
|
|
9
|
my ($count) = @_; |
1749
|
|
|
|
|
|
|
|
1750
|
|
|
|
|
|
|
return sub { |
1751
|
1
|
|
|
1
|
|
5
|
my ($source) = @_; |
1752
|
|
|
|
|
|
|
|
1753
|
|
|
|
|
|
|
return rx_observable->new(sub { |
1754
|
1
|
|
|
|
|
3
|
my ($subscriber) = @_; |
1755
|
|
|
|
|
|
|
|
1756
|
1
|
|
|
|
|
3
|
my $count = int $count; |
1757
|
|
|
|
|
|
|
|
1758
|
1
|
|
|
|
|
3
|
my $own_subscriber; |
1759
|
1
|
50
|
|
|
|
7
|
$own_subscriber = $subscriber if $count <= 0; |
1760
|
|
|
|
|
|
|
$own_subscriber //= { |
1761
|
|
|
|
|
|
|
%$subscriber, |
1762
|
|
|
|
|
|
|
next => sub { |
1763
|
5
|
100
|
|
|
|
17
|
if ($count-- <= 0) { |
1764
|
2
|
50
|
|
|
|
12
|
$subscriber->{next}->(@_) if defined $subscriber->{next}; |
1765
|
|
|
|
|
|
|
} |
1766
|
|
|
|
|
|
|
}, |
1767
|
1
|
|
50
|
|
|
13
|
}; |
1768
|
|
|
|
|
|
|
|
1769
|
1
|
|
|
|
|
6
|
$source->subscribe($own_subscriber); |
1770
|
|
|
|
|
|
|
|
1771
|
1
|
|
|
|
|
8
|
return; |
1772
|
1
|
|
|
|
|
2
|
}); |
1773
|
1
|
|
|
|
|
11
|
}; |
1774
|
|
|
|
|
|
|
} |
1775
|
|
|
|
|
|
|
|
1776
|
|
|
|
|
|
|
sub op_skip_last { |
1777
|
1
|
|
|
1
|
|
3
|
my ($skip_count) = @_; |
1778
|
|
|
|
|
|
|
|
1779
|
|
|
|
|
|
|
return sub { |
1780
|
1
|
|
|
1
|
|
3
|
my ($source) = @_; |
1781
|
|
|
|
|
|
|
|
1782
|
|
|
|
|
|
|
return rx_observable->new(sub { |
1783
|
1
|
|
|
|
|
3
|
my ($subscriber) = @_; |
1784
|
|
|
|
|
|
|
|
1785
|
1
|
|
|
|
|
2
|
my @skipped; |
1786
|
1
|
|
|
|
|
4
|
$subscriber->subscription->add(sub { undef @skipped }); |
|
1
|
|
|
|
|
5
|
|
1787
|
|
|
|
|
|
|
|
1788
|
1
|
|
|
|
|
4
|
my $own_subscriber = { %$subscriber }; |
1789
|
|
|
|
|
|
|
$own_subscriber->{next} &&= sub { |
1790
|
4
|
|
|
|
|
10
|
my ($v) = @_; |
1791
|
|
|
|
|
|
|
|
1792
|
4
|
|
|
|
|
8
|
push @skipped, $v; |
1793
|
4
|
100
|
|
|
|
24
|
if (@skipped > $skip_count) { |
1794
|
2
|
|
|
|
|
5
|
my $new_v = shift @skipped; |
1795
|
2
|
50
|
|
|
|
9
|
$subscriber->{next}->($new_v) if defined $subscriber->{next}; |
1796
|
|
|
|
|
|
|
} |
1797
|
1
|
|
50
|
|
|
15
|
}; |
1798
|
|
|
|
|
|
|
|
1799
|
1
|
|
|
|
|
5
|
$source->subscribe($own_subscriber); |
1800
|
|
|
|
|
|
|
|
1801
|
1
|
|
|
|
|
3
|
return; |
1802
|
1
|
|
|
|
|
4
|
}); |
1803
|
1
|
|
|
|
|
6
|
}; |
1804
|
|
|
|
|
|
|
} |
1805
|
|
|
|
|
|
|
|
1806
|
|
|
|
|
|
|
sub op_skip_until { |
1807
|
6
|
|
|
6
|
|
13
|
my ($notifier) = @_; |
1808
|
|
|
|
|
|
|
|
1809
|
|
|
|
|
|
|
# FUTURE TODO: allow notifier to be a promise |
1810
|
6
|
50
|
|
|
|
18
|
croak q"You provided 'undef' where a stream was expected. You can provide an observable." |
1811
|
|
|
|
|
|
|
unless defined $notifier; |
1812
|
6
|
50
|
33
|
|
|
46
|
croak q"The notifier of 'op_skip_until' needs to be an observable." |
1813
|
|
|
|
|
|
|
unless blessed $notifier and $notifier->isa('RxPerl::Observable'); |
1814
|
|
|
|
|
|
|
|
1815
|
|
|
|
|
|
|
return sub { |
1816
|
6
|
|
|
6
|
|
16
|
my ($source) = @_; |
1817
|
|
|
|
|
|
|
|
1818
|
|
|
|
|
|
|
return rx_observable->new(sub { |
1819
|
6
|
|
|
|
|
29
|
my ($subscriber) = @_; |
1820
|
|
|
|
|
|
|
|
1821
|
6
|
|
|
|
|
8
|
my $notifier_has_emitted; |
1822
|
|
|
|
|
|
|
my $n_s = $notifier->pipe( |
1823
|
|
|
|
|
|
|
op_take(1), |
1824
|
|
|
|
|
|
|
)->subscribe( |
1825
|
|
|
|
|
|
|
sub { |
1826
|
3
|
|
|
|
|
5
|
$notifier_has_emitted = 1; |
1827
|
|
|
|
|
|
|
}, |
1828
|
|
|
|
|
|
|
sub { |
1829
|
1
|
50
|
|
|
|
5
|
$subscriber->{error}->(@_) if defined $subscriber->{error}; |
1830
|
|
|
|
|
|
|
}, |
1831
|
6
|
|
|
|
|
19
|
); |
1832
|
|
|
|
|
|
|
|
1833
|
|
|
|
|
|
|
my $own_subscriber = { |
1834
|
|
|
|
|
|
|
%$subscriber, |
1835
|
|
|
|
|
|
|
next => sub { |
1836
|
|
|
|
|
|
|
$subscriber->{next}->(@_) if defined $subscriber->{next} |
1837
|
32
|
100
|
66
|
|
|
139
|
and $notifier_has_emitted; |
1838
|
|
|
|
|
|
|
}, |
1839
|
6
|
|
|
|
|
54
|
}; |
1840
|
|
|
|
|
|
|
|
1841
|
6
|
|
|
|
|
22
|
$source->subscribe($own_subscriber); |
1842
|
|
|
|
|
|
|
|
1843
|
6
|
|
|
|
|
21
|
return $n_s; |
1844
|
6
|
|
|
|
|
14
|
}); |
1845
|
6
|
|
|
|
|
43
|
}; |
1846
|
|
|
|
|
|
|
} |
1847
|
|
|
|
|
|
|
|
1848
|
|
|
|
|
|
|
sub op_skip_while { |
1849
|
1
|
|
|
1
|
|
3
|
my ($predicate) = @_; |
1850
|
|
|
|
|
|
|
|
1851
|
|
|
|
|
|
|
return sub { |
1852
|
1
|
|
|
1
|
|
2
|
my ($source) = @_; |
1853
|
|
|
|
|
|
|
|
1854
|
|
|
|
|
|
|
return rx_observable->new(sub { |
1855
|
1
|
|
|
|
|
3
|
my ($subscriber) = @_; |
1856
|
|
|
|
|
|
|
|
1857
|
1
|
|
|
|
|
2
|
my $finished_skipping = 0; |
1858
|
|
|
|
|
|
|
|
1859
|
1
|
|
|
|
|
3
|
my $idx = 0; |
1860
|
|
|
|
|
|
|
my $own_subscriber = { |
1861
|
|
|
|
|
|
|
%$subscriber, |
1862
|
|
|
|
|
|
|
next => sub { |
1863
|
6
|
|
|
|
|
8
|
my ($v) = @_; |
1864
|
|
|
|
|
|
|
|
1865
|
6
|
|
|
|
|
9
|
my $should_display; |
1866
|
6
|
100
|
|
|
|
12
|
if ($finished_skipping) { |
1867
|
3
|
|
|
|
|
6
|
$should_display = 1; |
1868
|
|
|
|
|
|
|
} else { |
1869
|
3
|
|
|
|
|
4
|
my $satisfies_predicate; |
1870
|
3
|
|
|
|
|
5
|
my $ok = eval { local $_ = $v; $satisfies_predicate = $predicate->($v, $idx++); 1 }; |
|
3
|
|
|
|
|
5
|
|
|
3
|
|
|
|
|
8
|
|
|
3
|
|
|
|
|
12
|
|
1871
|
3
|
50
|
|
|
|
8
|
$ok or do { |
1872
|
0
|
0
|
|
|
|
0
|
$subscriber->{error}->($@) if defined $subscriber->{error}; |
1873
|
0
|
|
|
|
|
0
|
return; |
1874
|
|
|
|
|
|
|
}; |
1875
|
3
|
100
|
|
|
|
18
|
if (! $satisfies_predicate) { |
1876
|
1
|
|
|
|
|
2
|
$finished_skipping = 1; |
1877
|
1
|
|
|
|
|
2
|
$should_display = 1; |
1878
|
|
|
|
|
|
|
} |
1879
|
|
|
|
|
|
|
} |
1880
|
|
|
|
|
|
|
|
1881
|
6
|
100
|
66
|
|
|
31
|
$subscriber->{next}->(@_) if $should_display and defined $subscriber->{next}; |
1882
|
|
|
|
|
|
|
} |
1883
|
1
|
|
|
|
|
9
|
}; |
1884
|
|
|
|
|
|
|
|
1885
|
1
|
|
|
|
|
6
|
$source->subscribe($own_subscriber); |
1886
|
|
|
|
|
|
|
|
1887
|
1
|
|
|
|
|
12
|
return; |
1888
|
1
|
|
|
|
|
3
|
}); |
1889
|
1
|
|
|
|
|
6
|
}; |
1890
|
|
|
|
|
|
|
} |
1891
|
|
|
|
|
|
|
|
1892
|
|
|
|
|
|
|
sub op_start_with { |
1893
|
3
|
|
|
3
|
|
9
|
my (@values) = @_; |
1894
|
|
|
|
|
|
|
|
1895
|
|
|
|
|
|
|
return sub { |
1896
|
3
|
|
|
3
|
|
4
|
my ($source) = @_; |
1897
|
|
|
|
|
|
|
|
1898
|
3
|
|
|
|
|
28
|
return rx_concat( |
1899
|
|
|
|
|
|
|
rx_of(@values), |
1900
|
|
|
|
|
|
|
$source, |
1901
|
|
|
|
|
|
|
); |
1902
|
3
|
|
|
|
|
17
|
}; |
1903
|
|
|
|
|
|
|
} |
1904
|
|
|
|
|
|
|
|
1905
|
|
|
|
|
|
|
sub op_switch_all { |
1906
|
|
|
|
|
|
|
return sub { |
1907
|
1
|
|
|
1
|
|
9
|
my ($source) = @_; |
1908
|
|
|
|
|
|
|
|
1909
|
|
|
|
|
|
|
return rx_observable->new(sub { |
1910
|
1
|
|
|
|
|
19
|
my ($subscriber) = @_; |
1911
|
|
|
|
|
|
|
|
1912
|
1
|
|
|
|
|
3
|
my $old_subscription; |
1913
|
|
|
|
|
|
|
|
1914
|
|
|
|
|
|
|
my $chief_is_complete; |
1915
|
1
|
|
|
|
|
0
|
my $sub_is_complete; |
1916
|
|
|
|
|
|
|
|
1917
|
|
|
|
|
|
|
my $obs_subscriber = { |
1918
|
|
|
|
|
|
|
next => sub { |
1919
|
6
|
50
|
|
|
|
33
|
$subscriber->{next}->(@_) if defined $subscriber->{next}; |
1920
|
|
|
|
|
|
|
}, |
1921
|
|
|
|
|
|
|
error => sub { |
1922
|
0
|
0
|
|
|
|
0
|
$subscriber->{error}->(@_) if defined $subscriber->{error}; |
1923
|
|
|
|
|
|
|
}, |
1924
|
|
|
|
|
|
|
complete => sub { |
1925
|
0
|
|
|
|
|
0
|
$sub_is_complete = 1; |
1926
|
0
|
0
|
0
|
|
|
0
|
$subscriber->{complete}->() if $chief_is_complete and defined $subscriber->{complete}; |
1927
|
|
|
|
|
|
|
}, |
1928
|
1
|
|
|
|
|
11
|
}; |
1929
|
|
|
|
|
|
|
|
1930
|
1
|
|
|
|
|
5
|
my $own_subscription = RxPerl::Subscription->new; |
1931
|
1
|
|
|
|
|
4
|
$subscriber->subscription->add(\$old_subscription, $own_subscription); |
1932
|
|
|
|
|
|
|
|
1933
|
|
|
|
|
|
|
my $own_subscriber = { |
1934
|
|
|
|
|
|
|
new_subscription => $own_subscription, |
1935
|
|
|
|
|
|
|
next => sub { |
1936
|
3
|
|
|
|
|
6
|
my ($new_observable) = @_; |
1937
|
|
|
|
|
|
|
|
1938
|
3
|
|
|
|
|
6
|
$sub_is_complete = 0; |
1939
|
3
|
100
|
|
|
|
11
|
$old_subscription->unsubscribe() if $old_subscription; |
1940
|
3
|
|
|
|
|
10
|
$old_subscription = $new_observable->subscribe($obs_subscriber); |
1941
|
|
|
|
|
|
|
}, |
1942
|
|
|
|
|
|
|
error => sub { |
1943
|
0
|
0
|
|
|
|
0
|
$subscriber->{error}->(@_) if defined $subscriber->{error}; |
1944
|
|
|
|
|
|
|
}, |
1945
|
|
|
|
|
|
|
complete => sub { |
1946
|
1
|
|
|
|
|
6
|
$chief_is_complete = 1; |
1947
|
1
|
50
|
33
|
|
|
10
|
$subscriber->{complete}->() if $sub_is_complete and defined $subscriber->{complete}; |
1948
|
|
|
|
|
|
|
}, |
1949
|
1
|
|
|
|
|
9
|
}; |
1950
|
|
|
|
|
|
|
|
1951
|
1
|
|
|
|
|
5
|
$source->subscribe($own_subscriber); |
1952
|
|
|
|
|
|
|
|
1953
|
1
|
|
|
|
|
4
|
return; |
1954
|
1
|
|
|
|
|
7
|
}); |
1955
|
1
|
|
|
1
|
|
8
|
}; |
1956
|
|
|
|
|
|
|
} |
1957
|
|
|
|
|
|
|
|
1958
|
|
|
|
|
|
|
sub op_switch_map { |
1959
|
0
|
|
|
0
|
|
0
|
my ($observable_factory) = @_; |
1960
|
|
|
|
|
|
|
|
1961
|
|
|
|
|
|
|
return sub { |
1962
|
0
|
|
|
0
|
|
0
|
my ($source) = @_; |
1963
|
|
|
|
|
|
|
|
1964
|
0
|
|
|
|
|
0
|
return $source->pipe( |
1965
|
|
|
|
|
|
|
op_map($observable_factory), |
1966
|
|
|
|
|
|
|
op_switch_all(), |
1967
|
|
|
|
|
|
|
); |
1968
|
|
|
|
|
|
|
} |
1969
|
0
|
|
|
|
|
0
|
} |
1970
|
|
|
|
|
|
|
|
1971
|
|
|
|
|
|
|
sub op_take { |
1972
|
87
|
|
|
87
|
|
157
|
my ($count) = @_; |
1973
|
|
|
|
|
|
|
|
1974
|
87
|
50
|
|
|
|
216
|
croak 'negative argument passed to op_take' unless $count >= 0; |
1975
|
|
|
|
|
|
|
|
1976
|
|
|
|
|
|
|
return sub { |
1977
|
87
|
|
|
87
|
|
150
|
my ($source) = @_; |
1978
|
|
|
|
|
|
|
|
1979
|
|
|
|
|
|
|
return rx_observable->new(sub { |
1980
|
71
|
|
|
|
|
134
|
my ($subscriber) = @_; |
1981
|
|
|
|
|
|
|
|
1982
|
71
|
|
|
|
|
133
|
my $remaining = int $count; |
1983
|
|
|
|
|
|
|
|
1984
|
71
|
100
|
|
|
|
166
|
if ($remaining == 0) { |
1985
|
1
|
50
|
|
|
|
7
|
$subscriber->{complete}->() if defined $subscriber->{complete}; |
1986
|
1
|
|
|
|
|
3
|
return; |
1987
|
|
|
|
|
|
|
} |
1988
|
|
|
|
|
|
|
|
1989
|
|
|
|
|
|
|
my $own_subscriber = { |
1990
|
|
|
|
|
|
|
%$subscriber, |
1991
|
|
|
|
|
|
|
next => sub { |
1992
|
192
|
100
|
|
|
|
660
|
$subscriber->{next}->(@_) if defined $subscriber->{next}; |
1993
|
192
|
100
|
100
|
|
|
886
|
$subscriber->{complete}->() if --$remaining == 0 and defined $subscriber->{complete}; |
1994
|
|
|
|
|
|
|
}, |
1995
|
70
|
|
|
|
|
375
|
}; |
1996
|
|
|
|
|
|
|
|
1997
|
70
|
|
|
|
|
244
|
$source->subscribe($own_subscriber); |
1998
|
|
|
|
|
|
|
|
1999
|
70
|
|
|
|
|
203
|
return; |
2000
|
87
|
|
|
|
|
250
|
}); |
2001
|
87
|
|
|
|
|
400
|
}; |
2002
|
|
|
|
|
|
|
} |
2003
|
|
|
|
|
|
|
|
2004
|
|
|
|
|
|
|
sub op_take_last { |
2005
|
2
|
|
|
2
|
|
6
|
my ($count) = @_; |
2006
|
|
|
|
|
|
|
|
2007
|
|
|
|
|
|
|
return sub { |
2008
|
2
|
|
|
2
|
|
5
|
my ($source) = @_; |
2009
|
|
|
|
|
|
|
|
2010
|
|
|
|
|
|
|
return rx_observable->new(sub { |
2011
|
2
|
|
|
|
|
3
|
my ($subscriber) = @_; |
2012
|
|
|
|
|
|
|
|
2013
|
2
|
|
|
|
|
6
|
my @last_values; |
2014
|
|
|
|
|
|
|
|
2015
|
|
|
|
|
|
|
my $own_subscriber = { |
2016
|
|
|
|
|
|
|
%$subscriber, |
2017
|
|
|
|
|
|
|
next => sub { |
2018
|
7
|
|
|
|
|
15
|
my ($v) = @_; |
2019
|
|
|
|
|
|
|
|
2020
|
7
|
|
|
|
|
10
|
push @last_values, $v; |
2021
|
7
|
100
|
|
|
|
19
|
if (@last_values > $count) { |
2022
|
2
|
|
|
|
|
8
|
shift @last_values; |
2023
|
|
|
|
|
|
|
} |
2024
|
|
|
|
|
|
|
}, |
2025
|
|
|
|
|
|
|
complete => sub { |
2026
|
2
|
|
|
|
|
6
|
foreach my $last_val (@last_values) { |
2027
|
5
|
50
|
|
|
|
16
|
$subscriber->{next}->($last_val) if defined $subscriber->{next}; |
2028
|
|
|
|
|
|
|
} |
2029
|
2
|
50
|
|
|
|
8
|
$subscriber->{complete}->() if defined $subscriber->{complete}; |
2030
|
|
|
|
|
|
|
}, |
2031
|
2
|
|
|
|
|
20
|
}; |
2032
|
|
|
|
|
|
|
|
2033
|
2
|
|
|
|
|
9
|
$source->subscribe($own_subscriber); |
2034
|
|
|
|
|
|
|
|
2035
|
2
|
|
|
|
|
13
|
return; |
2036
|
2
|
|
|
|
|
6
|
}); |
2037
|
|
|
|
|
|
|
} |
2038
|
2
|
|
|
|
|
12
|
} |
2039
|
|
|
|
|
|
|
|
2040
|
|
|
|
|
|
|
sub op_take_until { |
2041
|
1
|
|
|
1
|
|
7
|
my ($notifier_observable) = @_; |
2042
|
|
|
|
|
|
|
|
2043
|
|
|
|
|
|
|
return sub { |
2044
|
1
|
|
|
1
|
|
4
|
my ($source) = @_; |
2045
|
|
|
|
|
|
|
|
2046
|
|
|
|
|
|
|
return rx_observable->new(sub { |
2047
|
1
|
|
|
|
|
2
|
my ($subscriber) = @_; |
2048
|
|
|
|
|
|
|
|
2049
|
|
|
|
|
|
|
my $n_s = $notifier_observable->subscribe( |
2050
|
|
|
|
|
|
|
sub { |
2051
|
1
|
50
|
|
|
|
7
|
$subscriber->{complete}->() if defined $subscriber->{complete}; |
2052
|
|
|
|
|
|
|
}, |
2053
|
|
|
|
|
|
|
sub { |
2054
|
0
|
0
|
|
|
|
0
|
$subscriber->{error}->(@_) if defined $subscriber->{error}; |
2055
|
|
|
|
|
|
|
}, |
2056
|
1
|
|
|
|
|
8
|
); |
2057
|
|
|
|
|
|
|
|
2058
|
1
|
|
|
|
|
5
|
$source->subscribe($subscriber); |
2059
|
|
|
|
|
|
|
|
2060
|
1
|
|
|
|
|
2
|
return $n_s; |
2061
|
1
|
|
|
|
|
4
|
}); |
2062
|
1
|
|
|
|
|
7
|
}; |
2063
|
|
|
|
|
|
|
} |
2064
|
|
|
|
|
|
|
|
2065
|
|
|
|
|
|
|
sub op_take_while { |
2066
|
2
|
|
|
2
|
|
5
|
my ($cond, $include) = @_; |
2067
|
|
|
|
|
|
|
|
2068
|
|
|
|
|
|
|
return sub { |
2069
|
2
|
|
|
2
|
|
6
|
my ($source) = @_; |
2070
|
|
|
|
|
|
|
|
2071
|
|
|
|
|
|
|
return rx_observable->new(sub { |
2072
|
2
|
|
|
|
|
7
|
my ($subscriber) = @_; |
2073
|
|
|
|
|
|
|
|
2074
|
2
|
|
|
|
|
4
|
my $i = 0; |
2075
|
|
|
|
|
|
|
my $own_subscriber = { |
2076
|
|
|
|
|
|
|
%$subscriber, |
2077
|
|
|
|
|
|
|
next => sub { |
2078
|
6
|
|
|
|
|
13
|
my ($value) = @_; |
2079
|
|
|
|
|
|
|
|
2080
|
6
|
100
|
|
|
|
9
|
if (! do { local $_ = $value; $cond->($value, $i++) }) { |
|
6
|
|
|
|
|
11
|
|
|
6
|
|
|
|
|
23
|
|
2081
|
2
|
100
|
66
|
|
|
19
|
$subscriber->{next}->($value) if $include and defined $subscriber->{next}; |
2082
|
2
|
50
|
|
|
|
9
|
$subscriber->{complete}->() if defined $subscriber->{complete}; |
2083
|
2
|
|
|
|
|
14
|
return; |
2084
|
|
|
|
|
|
|
} |
2085
|
|
|
|
|
|
|
|
2086
|
4
|
50
|
|
|
|
34
|
$subscriber->{next}->(@_) if defined $subscriber->{next}; |
2087
|
|
|
|
|
|
|
}, |
2088
|
2
|
|
|
|
|
17
|
}; |
2089
|
|
|
|
|
|
|
|
2090
|
2
|
|
|
|
|
8
|
$source->subscribe($own_subscriber); |
2091
|
|
|
|
|
|
|
|
2092
|
2
|
|
|
|
|
5
|
return; |
2093
|
2
|
|
|
|
|
6
|
}); |
2094
|
2
|
|
|
|
|
10
|
}; |
2095
|
|
|
|
|
|
|
} |
2096
|
|
|
|
|
|
|
|
2097
|
|
|
|
|
|
|
sub op_tap { |
2098
|
0
|
|
|
0
|
|
0
|
my @args = @_; |
2099
|
|
|
|
|
|
|
|
2100
|
|
|
|
|
|
|
return sub { |
2101
|
0
|
|
|
0
|
|
0
|
my ($source) = @_; |
2102
|
|
|
|
|
|
|
|
2103
|
|
|
|
|
|
|
return rx_observable->new(sub { |
2104
|
0
|
|
|
|
|
0
|
my ($subscriber) = @_; |
2105
|
|
|
|
|
|
|
|
2106
|
0
|
|
|
|
|
0
|
my @args = @args; |
2107
|
0
|
0
|
0
|
|
|
0
|
my $tap_subscriber; $tap_subscriber = $args[0] if (reftype($args[0]) // '') eq 'HASH'; |
|
0
|
|
|
|
|
0
|
|
2108
|
|
|
|
|
|
|
$tap_subscriber //= { |
2109
|
0
|
|
0
|
|
|
0
|
map {($_, shift @args)} qw/ next error complete / |
|
0
|
|
|
|
|
0
|
|
2110
|
|
|
|
|
|
|
}; |
2111
|
|
|
|
|
|
|
|
2112
|
0
|
|
|
|
|
0
|
my %own_keys = map {$_ => 1} grep { /^(next|error|complete)\z/ } (keys(%$tap_subscriber), keys(%$subscriber)); |
|
0
|
|
|
|
|
0
|
|
|
0
|
|
|
|
|
0
|
|
2113
|
|
|
|
|
|
|
|
2114
|
|
|
|
|
|
|
my $own_subscriber = { |
2115
|
|
|
|
|
|
|
%$subscriber, |
2116
|
|
|
|
|
|
|
map { |
2117
|
0
|
|
|
|
|
0
|
my $key = $_; |
|
0
|
|
|
|
|
0
|
|
2118
|
|
|
|
|
|
|
($key => sub { |
2119
|
0
|
0
|
|
|
|
0
|
$tap_subscriber->{$key}->(@_) if defined $tap_subscriber->{$key}; |
2120
|
0
|
0
|
|
|
|
0
|
$subscriber->{$key}->(@_) if defined $subscriber->{$key}; |
2121
|
0
|
|
|
|
|
0
|
}); |
2122
|
|
|
|
|
|
|
} keys %own_keys |
2123
|
|
|
|
|
|
|
}; |
2124
|
|
|
|
|
|
|
|
2125
|
0
|
|
|
|
|
0
|
$source->subscribe($own_subscriber); |
2126
|
|
|
|
|
|
|
|
2127
|
0
|
|
|
|
|
0
|
return; |
2128
|
0
|
|
|
|
|
0
|
}); |
2129
|
0
|
|
|
|
|
0
|
}; |
2130
|
|
|
|
|
|
|
} |
2131
|
|
|
|
|
|
|
|
2132
|
|
|
|
|
|
|
sub op_throttle { |
2133
|
0
|
|
|
0
|
|
0
|
my ($duration_selector) = @_; |
2134
|
|
|
|
|
|
|
|
2135
|
|
|
|
|
|
|
return sub { |
2136
|
0
|
|
|
0
|
|
0
|
my ($source) = @_; |
2137
|
|
|
|
|
|
|
|
2138
|
|
|
|
|
|
|
return rx_observable->new(sub { |
2139
|
0
|
|
|
|
|
0
|
my ($subscriber) = @_; |
2140
|
|
|
|
|
|
|
|
2141
|
0
|
|
|
|
|
0
|
my $mini_subscription; |
2142
|
|
|
|
|
|
|
|
2143
|
|
|
|
|
|
|
my $mini_subscriber = { |
2144
|
|
|
|
|
|
|
error => sub { |
2145
|
0
|
0
|
|
|
|
0
|
$subscriber->{error}->(@_) if defined $subscriber->{error}; |
2146
|
|
|
|
|
|
|
}, |
2147
|
|
|
|
|
|
|
complete => sub { |
2148
|
0
|
|
|
|
|
0
|
undef $mini_subscription; |
2149
|
|
|
|
|
|
|
}, |
2150
|
0
|
|
|
|
|
0
|
}; |
2151
|
|
|
|
|
|
|
|
2152
|
|
|
|
|
|
|
my $own_subscriber = { |
2153
|
|
|
|
|
|
|
%$subscriber, |
2154
|
|
|
|
|
|
|
next => sub { |
2155
|
0
|
|
|
|
|
0
|
my ($v) = @_; |
2156
|
|
|
|
|
|
|
|
2157
|
0
|
0
|
|
|
|
0
|
if (! $mini_subscription) { |
2158
|
0
|
0
|
|
|
|
0
|
$subscriber->{next}->(@_) if defined $subscriber->{next}; |
2159
|
0
|
|
|
|
|
0
|
$mini_subscription = do { local $_ = $v; $duration_selector->($v) }->pipe( |
|
0
|
|
|
|
|
0
|
|
|
0
|
|
|
|
|
0
|
|
2160
|
|
|
|
|
|
|
op_take(1), |
2161
|
|
|
|
|
|
|
)->subscribe($mini_subscriber); |
2162
|
|
|
|
|
|
|
} |
2163
|
|
|
|
|
|
|
}, |
2164
|
0
|
|
|
|
|
0
|
}; |
2165
|
|
|
|
|
|
|
|
2166
|
0
|
|
|
|
|
0
|
$subscriber->subscription->add(\$mini_subscription); |
2167
|
|
|
|
|
|
|
|
2168
|
0
|
|
|
|
|
0
|
$source->subscribe($own_subscriber); |
2169
|
|
|
|
|
|
|
|
2170
|
0
|
|
|
|
|
0
|
return; |
2171
|
0
|
|
|
|
|
0
|
}); |
2172
|
0
|
|
|
|
|
0
|
}; |
2173
|
|
|
|
|
|
|
} |
2174
|
|
|
|
|
|
|
|
2175
|
|
|
|
|
|
|
sub op_throttle_time { |
2176
|
0
|
|
|
0
|
|
0
|
my ($duration) = @_; |
2177
|
|
|
|
|
|
|
|
2178
|
0
|
|
|
0
|
|
0
|
return op_throttle(sub { rx_timer($duration) }); |
|
0
|
|
|
|
|
0
|
|
2179
|
|
|
|
|
|
|
} |
2180
|
|
|
|
|
|
|
|
2181
|
|
|
|
|
|
|
sub op_throw_if_empty { |
2182
|
0
|
|
|
0
|
|
0
|
my ($error_factory) = @_; |
2183
|
|
|
|
|
|
|
|
2184
|
|
|
|
|
|
|
return sub { |
2185
|
0
|
|
|
0
|
|
0
|
my ($source) = @_; |
2186
|
|
|
|
|
|
|
|
2187
|
|
|
|
|
|
|
return rx_observable->new(sub { |
2188
|
0
|
|
|
|
|
0
|
my ($subscriber) = @_; |
2189
|
|
|
|
|
|
|
|
2190
|
0
|
|
|
|
|
0
|
my $is_empty = 1; |
2191
|
|
|
|
|
|
|
|
2192
|
|
|
|
|
|
|
my $own_subscriber = { |
2193
|
|
|
|
|
|
|
%$subscriber, |
2194
|
|
|
|
|
|
|
next => sub { |
2195
|
0
|
|
|
|
|
0
|
$is_empty = 0; |
2196
|
0
|
0
|
|
|
|
0
|
$subscriber->{next}->(@_) if defined $subscriber->{next}; |
2197
|
|
|
|
|
|
|
}, |
2198
|
|
|
|
|
|
|
complete => sub { |
2199
|
0
|
0
|
|
|
|
0
|
if ($is_empty) { |
2200
|
0
|
0
|
|
|
|
0
|
$subscriber->{error}->($error_factory->()) if defined $subscriber->{error}; |
2201
|
|
|
|
|
|
|
} else { |
2202
|
0
|
0
|
|
|
|
0
|
$subscriber->{complete}->() if defined $subscriber->{complete}; |
2203
|
|
|
|
|
|
|
} |
2204
|
|
|
|
|
|
|
}, |
2205
|
0
|
|
|
|
|
0
|
}; |
2206
|
|
|
|
|
|
|
|
2207
|
0
|
|
|
|
|
0
|
$source->subscribe($own_subscriber); |
2208
|
|
|
|
|
|
|
|
2209
|
0
|
|
|
|
|
0
|
return; |
2210
|
0
|
|
|
|
|
0
|
}); |
2211
|
0
|
|
|
|
|
0
|
}; |
2212
|
|
|
|
|
|
|
} |
2213
|
|
|
|
|
|
|
|
2214
|
|
|
|
|
|
|
sub op_time_interval { |
2215
|
|
|
|
|
|
|
return sub { |
2216
|
0
|
|
|
0
|
|
0
|
my ($source) = @_; |
2217
|
|
|
|
|
|
|
|
2218
|
0
|
|
|
|
|
0
|
my $t0 = Time::HiRes::time(); |
2219
|
|
|
|
|
|
|
|
2220
|
|
|
|
|
|
|
return $source->pipe( |
2221
|
|
|
|
|
|
|
op_map(sub { |
2222
|
0
|
|
|
|
|
0
|
my $now = Time::HiRes::time(); |
2223
|
0
|
|
|
|
|
0
|
my $interval = $now - $t0; |
2224
|
0
|
|
|
|
|
0
|
$t0 = $now; |
2225
|
0
|
|
|
|
|
0
|
return { value => $_, interval => $interval }; |
2226
|
0
|
|
|
|
|
0
|
}), |
2227
|
|
|
|
|
|
|
); |
2228
|
0
|
|
|
0
|
|
0
|
}; |
2229
|
|
|
|
|
|
|
} |
2230
|
|
|
|
|
|
|
|
2231
|
|
|
|
|
|
|
sub op_timeout { |
2232
|
0
|
|
|
0
|
|
0
|
my ($duration) = @_; |
2233
|
|
|
|
|
|
|
|
2234
|
|
|
|
|
|
|
return sub { |
2235
|
0
|
|
|
0
|
|
0
|
my ($source) = @_; |
2236
|
|
|
|
|
|
|
|
2237
|
|
|
|
|
|
|
return rx_observable->new(sub { |
2238
|
0
|
|
|
|
|
0
|
my ($subscriber) = @_; |
2239
|
|
|
|
|
|
|
|
2240
|
0
|
|
|
|
|
0
|
my $subject = rx_behavior_subject->new(1); |
2241
|
|
|
|
|
|
|
my $s_s = $subject->pipe( |
2242
|
0
|
|
|
|
|
0
|
op_switch_map(sub { rx_timer($duration) }), |
2243
|
|
|
|
|
|
|
)->subscribe(sub { |
2244
|
0
|
0
|
|
|
|
0
|
$subscriber->{error}->('Timeout has occurred') if defined $subscriber->{error}; |
2245
|
0
|
|
|
|
|
0
|
}); |
2246
|
|
|
|
|
|
|
|
2247
|
0
|
|
|
|
|
0
|
my $own_subscription = RxPerl::Subscription->new; |
2248
|
0
|
|
|
|
|
0
|
$subscriber->subscription->add($own_subscription, $s_s); |
2249
|
|
|
|
|
|
|
|
2250
|
|
|
|
|
|
|
my $own_subscriber = { |
2251
|
|
|
|
|
|
|
new_subscription => $own_subscription, |
2252
|
|
|
|
|
|
|
next => sub { |
2253
|
0
|
0
|
|
|
|
0
|
$subject->{next}->(1) if defined $subject->{next}; |
2254
|
0
|
0
|
|
|
|
0
|
$subscriber->{next}->(@_) if defined $subscriber->{next}; |
2255
|
|
|
|
|
|
|
}, |
2256
|
|
|
|
|
|
|
error => sub { |
2257
|
0
|
0
|
|
|
|
0
|
$subscriber->{error}->(@_) if defined $subscriber->{error}; |
2258
|
|
|
|
|
|
|
}, |
2259
|
|
|
|
|
|
|
complete => sub { |
2260
|
0
|
0
|
|
|
|
0
|
$subscriber->{complete}->() if defined $subscriber->{complete}; |
2261
|
|
|
|
|
|
|
} |
2262
|
0
|
|
|
|
|
0
|
}; |
2263
|
|
|
|
|
|
|
|
2264
|
0
|
|
|
|
|
0
|
$source->subscribe($own_subscriber); |
2265
|
|
|
|
|
|
|
|
2266
|
0
|
|
|
|
|
0
|
return; |
2267
|
0
|
|
|
|
|
0
|
}); |
2268
|
0
|
|
|
|
|
0
|
}; |
2269
|
|
|
|
|
|
|
} |
2270
|
|
|
|
|
|
|
|
2271
|
|
|
|
|
|
|
sub op_timestamp { |
2272
|
|
|
|
|
|
|
return op_map(sub { |
2273
|
|
|
|
|
|
|
return { |
2274
|
0
|
|
|
0
|
|
0
|
value => $_, |
2275
|
|
|
|
|
|
|
timestamp => Time::HiRes::time(), |
2276
|
|
|
|
|
|
|
}; |
2277
|
0
|
|
|
0
|
|
0
|
}); |
2278
|
|
|
|
|
|
|
} |
2279
|
|
|
|
|
|
|
|
2280
|
|
|
|
|
|
|
sub op_to_array { |
2281
|
|
|
|
|
|
|
return sub { |
2282
|
0
|
|
|
0
|
|
0
|
my ($source) = @_; |
2283
|
|
|
|
|
|
|
|
2284
|
|
|
|
|
|
|
return rx_observable->new(sub { |
2285
|
0
|
|
|
|
|
0
|
my ($subscriber) = @_; |
2286
|
|
|
|
|
|
|
|
2287
|
0
|
|
|
|
|
0
|
my @values; |
2288
|
|
|
|
|
|
|
|
2289
|
|
|
|
|
|
|
my $own_subscriber = { |
2290
|
|
|
|
|
|
|
%$subscriber, |
2291
|
|
|
|
|
|
|
next => sub { |
2292
|
0
|
|
|
|
|
0
|
my ($v) = @_; |
2293
|
0
|
|
|
|
|
0
|
push @values, $v; |
2294
|
|
|
|
|
|
|
}, |
2295
|
|
|
|
|
|
|
complete => sub { |
2296
|
0
|
0
|
|
|
|
0
|
$subscriber->{next}->(\@values) if defined $subscriber->{next}; |
2297
|
0
|
0
|
|
|
|
0
|
$subscriber->{complete}->() if defined $subscriber->{complete}; |
2298
|
|
|
|
|
|
|
}, |
2299
|
0
|
|
|
|
|
0
|
}; |
2300
|
|
|
|
|
|
|
|
2301
|
0
|
|
|
|
|
0
|
$source->subscribe($own_subscriber); |
2302
|
|
|
|
|
|
|
|
2303
|
0
|
|
|
|
|
0
|
return; |
2304
|
0
|
|
|
|
|
0
|
}); |
2305
|
0
|
|
|
0
|
|
0
|
}; |
2306
|
|
|
|
|
|
|
} |
2307
|
|
|
|
|
|
|
|
2308
|
|
|
|
|
|
|
sub op_with_latest_from { |
2309
|
1
|
|
|
1
|
|
17
|
my (@other_observables) = @_; |
2310
|
|
|
|
|
|
|
|
2311
|
|
|
|
|
|
|
return sub { |
2312
|
1
|
|
|
1
|
|
5
|
my ($source) = @_; |
2313
|
|
|
|
|
|
|
|
2314
|
|
|
|
|
|
|
return rx_observable->new(sub { |
2315
|
1
|
|
|
|
|
4
|
my ($subscriber) = @_; |
2316
|
|
|
|
|
|
|
|
2317
|
1
|
|
|
|
|
3
|
my @other_observables = @other_observables; |
2318
|
1
|
|
|
|
|
2
|
my $i = 0; |
2319
|
1
|
|
|
|
|
2
|
my %didnt_emit = map {($i++, 1)} @other_observables; |
|
1
|
|
|
|
|
4
|
|
2320
|
1
|
|
|
|
|
3
|
my @latest_values; |
2321
|
|
|
|
|
|
|
my %other_subscriptions; |
2322
|
|
|
|
|
|
|
|
2323
|
|
|
|
|
|
|
$subscriber->subscription->add( |
2324
|
1
|
|
|
|
|
3
|
\%other_subscriptions, sub { undef @other_observables }, |
2325
|
1
|
|
|
|
|
4
|
); |
2326
|
|
|
|
|
|
|
|
2327
|
1
|
|
|
|
|
5
|
for (my $i = 0; $i < @other_observables; $i++) { |
2328
|
1
|
|
|
|
|
3
|
my $j = $i; |
2329
|
1
|
|
|
|
|
2
|
my $other_observable = $other_observables[$j]; |
2330
|
|
|
|
|
|
|
|
2331
|
1
|
|
|
|
|
3
|
my $other_subscription = RxPerl::Subscription->new; |
2332
|
1
|
|
|
|
|
3
|
$other_subscriptions{$other_subscription} = $other_subscription; |
2333
|
|
|
|
|
|
|
$other_observable->subscribe({ |
2334
|
|
|
|
|
|
|
new_subscription => $other_subscription, |
2335
|
|
|
|
|
|
|
next => sub { |
2336
|
4
|
|
|
|
|
10
|
my ($value) = @_; |
2337
|
|
|
|
|
|
|
|
2338
|
4
|
|
|
|
|
7
|
$latest_values[$j] = $value; |
2339
|
4
|
|
|
|
|
10
|
delete $didnt_emit{$j}; |
2340
|
|
|
|
|
|
|
}, |
2341
|
|
|
|
|
|
|
error => $subscriber->{error}, |
2342
|
1
|
|
|
|
|
7
|
}); |
2343
|
|
|
|
|
|
|
} |
2344
|
|
|
|
|
|
|
|
2345
|
|
|
|
|
|
|
$source->subscribe({ |
2346
|
|
|
|
|
|
|
%$subscriber, |
2347
|
|
|
|
|
|
|
next => sub { |
2348
|
5
|
|
|
|
|
12
|
my ($value) = @_; |
2349
|
|
|
|
|
|
|
|
2350
|
5
|
100
|
|
|
|
14
|
if (! %didnt_emit) { |
2351
|
4
|
50
|
|
|
|
27
|
$subscriber->{next}->([$value, @latest_values]) if defined $subscriber->{next}; |
2352
|
|
|
|
|
|
|
} |
2353
|
|
|
|
|
|
|
}, |
2354
|
1
|
|
|
|
|
13
|
}); |
2355
|
|
|
|
|
|
|
|
2356
|
1
|
|
|
|
|
4
|
return; |
2357
|
1
|
|
|
|
|
3
|
}); |
2358
|
1
|
|
|
|
|
8
|
}; |
2359
|
|
|
|
|
|
|
} |
2360
|
|
|
|
|
|
|
|
2361
|
|
|
|
|
|
|
sub op_zip_with { |
2362
|
1
|
|
|
1
|
|
4
|
my @other_sources = @_; |
2363
|
|
|
|
|
|
|
|
2364
|
|
|
|
|
|
|
return sub { |
2365
|
1
|
|
|
1
|
|
2
|
my ($source) = @_; |
2366
|
|
|
|
|
|
|
|
2367
|
1
|
|
|
|
|
5
|
return rx_zip( |
2368
|
|
|
|
|
|
|
$source, |
2369
|
|
|
|
|
|
|
@other_sources, |
2370
|
|
|
|
|
|
|
); |
2371
|
1
|
|
|
|
|
11
|
}; |
2372
|
|
|
|
|
|
|
} |
2373
|
|
|
|
|
|
|
|
2374
|
|
|
|
|
|
|
sub _eqq { |
2375
|
20
|
|
|
20
|
|
33
|
my ($x, $y) = @_; |
2376
|
|
|
|
|
|
|
|
2377
|
20
|
100
|
|
|
|
48
|
defined $x or return !defined $y; |
2378
|
17
|
100
|
|
|
|
36
|
defined $y or return !!0; |
2379
|
15
|
100
|
|
|
|
31
|
ref $x eq ref $y or return !!0; |
2380
|
14
|
100
|
|
|
|
72
|
return length(ref $x) ? refaddr $x == refaddr $y : $x eq $y; |
2381
|
|
|
|
|
|
|
} |
2382
|
|
|
|
|
|
|
|
2383
|
|
|
|
|
|
|
1; |