Chromium Code Reviews
chromiumcodereview-hr@appspot.gserviceaccount.com (chromiumcodereview-hr) | Please choose your nickname with Settings | Help | Chromium Project | Gerrit Changes | Sign out
(4)

Side by Side Diff: test/generated_sdk/lib/async/stream_pipe.dart

Issue 1162723007: remove generated_sdk from checked in code (Closed) Base URL: git@github.com:dart-lang/dev_compiler.git@master
Patch Set: Created 5 years, 6 months ago
Use n/p to move between diff chunks; N/P to move between comments. Draft comments are only viewable by you.
Jump to:
View unified diff | Download patch
OLDNEW
(Empty)
1 // Copyright (c) 2012, the Dart project authors. Please see the AUTHORS file
2 // for details. All rights reserved. Use of this source code is governed by a
3 // BSD-style license that can be found in the LICENSE file.
4
5 part of dart.async;
6
7 /** Runs user code and takes actions depending on success or failure. */
8 _runUserCode(userCode(),
9 onSuccess(value),
10 onError(error, StackTrace stackTrace)) {
11 try {
12 onSuccess(userCode());
13 } catch (e, s) {
14 AsyncError replacement = Zone.current.errorCallback(e, s);
15 if (replacement == null) {
16 onError(e, s);
17 } else {
18 var error = _nonNullError(replacement.error);
19 var stackTrace = replacement.stackTrace;
20 onError(error, stackTrace);
21 }
22 }
23 }
24
25 /** Helper function to cancel a subscription and wait for the potential future,
26 before completing with an error. */
27 void _cancelAndError(StreamSubscription subscription,
28 _Future future,
29 error,
30 StackTrace stackTrace) {
31 var cancelFuture = subscription.cancel();
32 if (cancelFuture is Future) {
33 cancelFuture.whenComplete(() => future._completeError(error, stackTrace));
34 } else {
35 future._completeError(error, stackTrace);
36 }
37 }
38
39 void _cancelAndErrorWithReplacement(StreamSubscription subscription,
40 _Future future,
41 error, StackTrace stackTrace) {
42 AsyncError replacement = Zone.current.errorCallback(error, stackTrace);
43 if (replacement != null) {
44 error = _nonNullError(replacement.error);
45 stackTrace = replacement.stackTrace;
46 }
47 _cancelAndError(subscription, future, error, stackTrace);
48 }
49
50 /** Helper function to make an onError argument to [_runUserCode]. */
51 _cancelAndErrorClosure(StreamSubscription subscription, _Future future) =>
52 ((error, StackTrace stackTrace) => _cancelAndError(
53 subscription, future, error, stackTrace));
54
55 /** Helper function to cancel a subscription and wait for the potential future,
56 before completing with a value. */
57 void _cancelAndValue(StreamSubscription subscription, _Future future, value) {
58 var cancelFuture = subscription.cancel();
59 if (cancelFuture is Future) {
60 cancelFuture.whenComplete(() => future._complete(value));
61 } else {
62 future._complete(value);
63 }
64 }
65
66
67 /**
68 * A [Stream] that forwards subscriptions to another stream.
69 *
70 * This stream implements [Stream], but forwards all subscriptions
71 * to an underlying stream, and wraps the returned subscription to
72 * modify the events on the way.
73 *
74 * This class is intended for internal use only.
75 */
76 abstract class _ForwardingStream<S, T> extends Stream<T> {
77 final Stream<S> _source;
78
79 _ForwardingStream(this._source);
80
81 bool get isBroadcast => _source.isBroadcast;
82
83 StreamSubscription<T> listen(void onData(T value),
84 { Function onError,
85 void onDone(),
86 bool cancelOnError }) {
87 cancelOnError = identical(true, cancelOnError);
88 return _createSubscription(onData, onError, onDone, cancelOnError);
89 }
90
91 StreamSubscription<T> _createSubscription(
92 void onData(T data),
93 Function onError,
94 void onDone(),
95 bool cancelOnError) {
96 return new _ForwardingStreamSubscription<S, T>(
97 this, onData, onError, onDone, cancelOnError);
98 }
99
100 // Override the following methods in subclasses to change the behavior.
101
102 void _handleData(S data, _EventSink<T> sink) {
103 dynamic outputData = data;
104 sink._add(outputData);
105 }
106
107 void _handleError(error, StackTrace stackTrace, _EventSink<T> sink) {
108 sink._addError(error, stackTrace);
109 }
110
111 void _handleDone(_EventSink<T> sink) {
112 sink._close();
113 }
114 }
115
116 /**
117 * Abstract superclass for subscriptions that forward to other subscriptions.
118 */
119 class _ForwardingStreamSubscription<S, T>
120 extends _BufferingStreamSubscription<T> {
121 final _ForwardingStream<S, T> _stream;
122
123 StreamSubscription<S> _subscription;
124
125 _ForwardingStreamSubscription(this._stream, void onData(T data),
126 Function onError, void onDone(),
127 bool cancelOnError)
128 : super(onData, onError, onDone, cancelOnError) {
129 _subscription = _stream._source.listen(_handleData,
130 onError: _handleError,
131 onDone: _handleDone);
132 }
133
134 // _StreamSink interface.
135 // Transformers sending more than one event have no way to know if the stream
136 // is canceled or closed after the first, so we just ignore remaining events.
137
138 void _add(T data) {
139 if (_isClosed) return;
140 super._add(data);
141 }
142
143 void _addError(Object error, StackTrace stackTrace) {
144 if (_isClosed) return;
145 super._addError(error, stackTrace);
146 }
147
148 // StreamSubscription callbacks.
149
150 void _onPause() {
151 if (_subscription == null) return;
152 _subscription.pause();
153 }
154
155 void _onResume() {
156 if (_subscription == null) return;
157 _subscription.resume();
158 }
159
160 Future _onCancel() {
161 if (_subscription != null) {
162 StreamSubscription subscription = _subscription;
163 _subscription = null;
164 subscription.cancel();
165 }
166 return null;
167 }
168
169 // Methods used as listener on source subscription.
170
171 void _handleData(S data) {
172 _stream._handleData(data, this);
173 }
174
175 void _handleError(error, StackTrace stackTrace) {
176 _stream._handleError(error, stackTrace, this);
177 }
178
179 void _handleDone() {
180 _stream._handleDone(this);
181 }
182 }
183
184 // -------------------------------------------------------------------
185 // Stream transformers used by the default Stream implementation.
186 // -------------------------------------------------------------------
187
188 typedef bool _Predicate<T>(T value);
189
190 void _addErrorWithReplacement(_EventSink sink, error, stackTrace) {
191 AsyncError replacement = Zone.current.errorCallback(error, stackTrace);
192 if (replacement != null) {
193 error = _nonNullError(replacement.error);
194 stackTrace = replacement.stackTrace;
195 }
196 sink._addError(error, stackTrace);
197 }
198
199
200 class _WhereStream<T> extends _ForwardingStream<T, T> {
201 final _Predicate<T> _test;
202
203 _WhereStream(Stream<T> source, bool test(T value))
204 : _test = test, super(source);
205
206 void _handleData(T inputEvent, _EventSink<T> sink) {
207 bool satisfies;
208 try {
209 satisfies = _test(inputEvent);
210 } catch (e, s) {
211 _addErrorWithReplacement(sink, e, s);
212 return;
213 }
214 if (satisfies) {
215 sink._add(inputEvent);
216 }
217 }
218 }
219
220
221 typedef T _Transformation<S, T>(S value);
222
223 /**
224 * A stream pipe that converts data events before passing them on.
225 */
226 class _MapStream<S, T> extends _ForwardingStream<S, T> {
227 final _Transformation _transform;
228
229 _MapStream(Stream<S> source, T transform(S event))
230 : this._transform = transform, super(source);
231
232 void _handleData(S inputEvent, _EventSink<T> sink) {
233 T outputEvent;
234 try {
235 outputEvent = _transform(inputEvent);
236 } catch (e, s) {
237 _addErrorWithReplacement(sink, e, s);
238 return;
239 }
240 sink._add(outputEvent);
241 }
242 }
243
244 /**
245 * A stream pipe that converts data events before passing them on.
246 */
247 class _ExpandStream<S, T> extends _ForwardingStream<S, T> {
248 final _Transformation<S, Iterable<T>> _expand;
249
250 _ExpandStream(Stream<S> source, Iterable<T> expand(S event))
251 : this._expand = expand, super(source);
252
253 void _handleData(S inputEvent, _EventSink<T> sink) {
254 try {
255 for (T value in _expand(inputEvent)) {
256 sink._add(value);
257 }
258 } catch (e, s) {
259 // If either _expand or iterating the generated iterator throws,
260 // we abort the iteration.
261 _addErrorWithReplacement(sink, e, s);
262 }
263 }
264 }
265
266
267 typedef bool _ErrorTest(error);
268
269 /**
270 * A stream pipe that converts or disposes error events
271 * before passing them on.
272 */
273 class _HandleErrorStream<T> extends _ForwardingStream<T, T> {
274 final Function _transform;
275 final _ErrorTest _test;
276
277 _HandleErrorStream(Stream<T> source,
278 Function onError,
279 bool test(error))
280 : this._transform = onError, this._test = test, super(source);
281
282 void _handleError(Object error, StackTrace stackTrace, _EventSink<T> sink) {
283 bool matches = true;
284 if (_test != null) {
285 try {
286 matches = _test(error);
287 } catch (e, s) {
288 _addErrorWithReplacement(sink, e, s);
289 return;
290 }
291 }
292 if (matches) {
293 try {
294 _invokeErrorHandler(_transform, error, stackTrace);
295 } catch (e, s) {
296 if (identical(e, error)) {
297 sink._addError(error, stackTrace);
298 } else {
299 _addErrorWithReplacement(sink, e, s);
300 }
301 return;
302 }
303 } else {
304 sink._addError(error, stackTrace);
305 }
306 }
307 }
308
309
310 class _TakeStream<T> extends _ForwardingStream<T, T> {
311 int _remaining;
312
313 _TakeStream(Stream<T> source, int count)
314 : this._remaining = count, super(source) {
315 // This test is done early to avoid handling an async error
316 // in the _handleData method.
317 if (count is! int) throw new ArgumentError(count);
318 }
319
320 void _handleData(T inputEvent, _EventSink<T> sink) {
321 if (_remaining > 0) {
322 sink._add(inputEvent);
323 _remaining -= 1;
324 if (_remaining == 0) {
325 // Closing also unsubscribes all subscribers, which unsubscribes
326 // this from source.
327 sink._close();
328 }
329 }
330 }
331 }
332
333
334 class _TakeWhileStream<T> extends _ForwardingStream<T, T> {
335 final _Predicate<T> _test;
336
337 _TakeWhileStream(Stream<T> source, bool test(T value))
338 : this._test = test, super(source);
339
340 void _handleData(T inputEvent, _EventSink<T> sink) {
341 bool satisfies;
342 try {
343 satisfies = _test(inputEvent);
344 } catch (e, s) {
345 _addErrorWithReplacement(sink, e, s);
346 // The test didn't say true. Didn't say false either, but we stop anyway.
347 sink._close();
348 return;
349 }
350 if (satisfies) {
351 sink._add(inputEvent);
352 } else {
353 sink._close();
354 }
355 }
356 }
357
358 class _SkipStream<T> extends _ForwardingStream<T, T> {
359 int _remaining;
360
361 _SkipStream(Stream<T> source, int count)
362 : this._remaining = count, super(source) {
363 // This test is done early to avoid handling an async error
364 // in the _handleData method.
365 if (count is! int || count < 0) throw new ArgumentError(count);
366 }
367
368 void _handleData(T inputEvent, _EventSink<T> sink) {
369 if (_remaining > 0) {
370 _remaining--;
371 return;
372 }
373 sink._add(inputEvent);
374 }
375 }
376
377 class _SkipWhileStream<T> extends _ForwardingStream<T, T> {
378 final _Predicate<T> _test;
379 bool _hasFailed = false;
380
381 _SkipWhileStream(Stream<T> source, bool test(T value))
382 : this._test = test, super(source);
383
384 void _handleData(T inputEvent, _EventSink<T> sink) {
385 if (_hasFailed) {
386 sink._add(inputEvent);
387 return;
388 }
389 bool satisfies;
390 try {
391 satisfies = _test(inputEvent);
392 } catch (e, s) {
393 _addErrorWithReplacement(sink, e, s);
394 // A failure to return a boolean is considered "not matching".
395 _hasFailed = true;
396 return;
397 }
398 if (!satisfies) {
399 _hasFailed = true;
400 sink._add(inputEvent);
401 }
402 }
403 }
404
405 typedef bool _Equality<T>(T a, T b);
406
407 class _DistinctStream<T> extends _ForwardingStream<T, T> {
408 static var _SENTINEL = new Object();
409
410 _Equality<T> _equals;
411 var _previous = _SENTINEL;
412
413 _DistinctStream(Stream<T> source, bool equals(T a, T b))
414 : _equals = equals, super(source);
415
416 void _handleData(T inputEvent, _EventSink<T> sink) {
417 if (identical(_previous, _SENTINEL)) {
418 _previous = inputEvent;
419 return sink._add(inputEvent);
420 } else {
421 bool isEqual;
422 try {
423 if (_equals == null) {
424 isEqual = (_previous == inputEvent);
425 } else {
426 isEqual = _equals(_previous, inputEvent);
427 }
428 } catch (e, s) {
429 _addErrorWithReplacement(sink, e, s);
430 return null;
431 }
432 if (!isEqual) {
433 sink._add(inputEvent);
434 _previous = inputEvent;
435 }
436 }
437 }
438 }
OLDNEW
« no previous file with comments | « test/generated_sdk/lib/async/stream_impl.dart ('k') | test/generated_sdk/lib/async/stream_transformers.dart » ('j') | no next file with comments »

Powered by Google App Engine
This is Rietveld 408576698