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

Side by Side Diff: sdk/lib/async/stream.dart

Issue 18749002: Use typedefs in lib. (Closed) Base URL: https://dart.googlecode.com/svn/branches/bleeding_edge/dart
Patch Set: Created 7 years, 5 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 | Annotate | Revision Log
OLDNEW
1 // Copyright (c) 2013, the Dart project authors. Please see the AUTHORS file 1 // Copyright (c) 2013, the Dart project authors. Please see the AUTHORS file
2 // for details. All rights reserved. Use of this source code is governed by a 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. 3 // BSD-style license that can be found in the LICENSE file.
4 4
5 part of dart.async; 5 part of dart.async;
6 6
7 // ------------------------------------------------------------------- 7 // -------------------------------------------------------------------
8 // Core Stream types 8 // Core Stream types
9 // ------------------------------------------------------------------- 9 // -------------------------------------------------------------------
10 10
(...skipping 259 matching lines...) Expand 10 before | Expand all | Expand 10 after
270 270
271 /** 271 /**
272 * Reduces a sequence of values by repeatedly applying [combine]. 272 * Reduces a sequence of values by repeatedly applying [combine].
273 */ 273 */
274 Future<T> reduce(T combine(T previous, T element)) { 274 Future<T> reduce(T combine(T previous, T element)) {
275 _FutureImpl<T> result = new _FutureImpl<T>(); 275 _FutureImpl<T> result = new _FutureImpl<T>();
276 bool seenFirst = false; 276 bool seenFirst = false;
277 T value; 277 T value;
278 StreamSubscription subscription; 278 StreamSubscription subscription;
279 subscription = this.listen( 279 subscription = this.listen(
280 // TODO(ahe): Restore type when feature is implemented in dart2js 280 (T element) {
281 // checked mode. http://dartbug.com/7733
282 (/* T */ element) {
283 if (seenFirst) { 281 if (seenFirst) {
284 _runUserCode(() => combine(value, element), 282 _runUserCode(() => combine(value, element),
285 (T newValue) { value = newValue; }, 283 (T newValue) { value = newValue; },
286 _cancelAndError(subscription, result)); 284 _cancelAndError(subscription, result));
287 } else { 285 } else {
288 value = element; 286 value = element;
289 seenFirst = true; 287 seenFirst = true;
290 } 288 }
291 }, 289 },
292 onError: result._setError, 290 onError: result._setError,
293 onDone: () { 291 onDone: () {
294 if (!seenFirst) { 292 if (!seenFirst) {
295 result._setError(new StateError("No elements")); 293 result._setError(new StateError("No elements"));
296 } else { 294 } else {
297 result._setValue(value); 295 result._setValue(value);
298 } 296 }
299 }, 297 },
300 cancelOnError: true 298 cancelOnError: true
301 ); 299 );
302 return result; 300 return result;
303 } 301 }
304 302
305 /** Reduces a sequence of values by repeatedly applying [combine]. */ 303 /** Reduces a sequence of values by repeatedly applying [combine]. */
306 Future fold(var initialValue, combine(var previous, T element)) { 304 Future fold(var initialValue, combine(var previous, T element)) {
307 _FutureImpl result = new _FutureImpl(); 305 _FutureImpl result = new _FutureImpl();
308 var value = initialValue; 306 var value = initialValue;
309 StreamSubscription subscription; 307 StreamSubscription subscription;
310 subscription = this.listen( 308 subscription = this.listen(
311 // TODO(ahe): Restore type when feature is implemented in dart2js 309 (T element) {
312 // checked mode. http://dartbug.com/7733
313 (/*T*/ element) {
314 _runUserCode( 310 _runUserCode(
315 () => combine(value, element), 311 () => combine(value, element),
316 (newValue) { value = newValue; }, 312 (newValue) { value = newValue; },
317 _cancelAndError(subscription, result) 313 _cancelAndError(subscription, result)
318 ); 314 );
319 }, 315 },
320 onError: (e) { 316 onError: (e) {
321 result._setError(e); 317 result._setError(e);
322 }, 318 },
323 onDone: () { 319 onDone: () {
324 result._setValue(value); 320 result._setValue(value);
325 }, 321 },
326 cancelOnError: true); 322 cancelOnError: true);
327 return result; 323 return result;
328 } 324 }
329 325
330 /** 326 /**
331 * Checks whether [needle] occurs in the elements provided by this stream. 327 * Checks whether [needle] occurs in the elements provided by this stream.
332 * 328 *
333 * Completes the [Future] when the answer is known. 329 * Completes the [Future] when the answer is known.
334 * If this stream reports an error, the [Future] will report that error. 330 * If this stream reports an error, the [Future] will report that error.
335 */ 331 */
336 Future<bool> contains(Object needle) { 332 Future<bool> contains(Object needle) {
337 _FutureImpl<bool> future = new _FutureImpl<bool>(); 333 _FutureImpl<bool> future = new _FutureImpl<bool>();
338 StreamSubscription subscription; 334 StreamSubscription subscription;
339 subscription = this.listen( 335 subscription = this.listen(
340 // TODO(ahe): Restore type when feature is implemented in dart2js 336 (T element) {
341 // checked mode. http://dartbug.com/7733
342 (/*T*/ element) {
343 _runUserCode( 337 _runUserCode(
344 () => (element == needle), 338 () => (element == needle),
345 (bool isMatch) { 339 (bool isMatch) {
346 if (isMatch) { 340 if (isMatch) {
347 subscription.cancel(); 341 subscription.cancel();
348 future._setValue(true); 342 future._setValue(true);
349 } 343 }
350 }, 344 },
351 _cancelAndError(subscription, future) 345 _cancelAndError(subscription, future)
352 ); 346 );
(...skipping 10 matching lines...) Expand all
363 * Executes [action] on each data event of the stream. 357 * Executes [action] on each data event of the stream.
364 * 358 *
365 * Completes the returned [Future] when all events of the stream 359 * Completes the returned [Future] when all events of the stream
366 * have been processed. Completes the future with an error if the 360 * have been processed. Completes the future with an error if the
367 * stream has an error event, or if [action] throws. 361 * stream has an error event, or if [action] throws.
368 */ 362 */
369 Future forEach(void action(T element)) { 363 Future forEach(void action(T element)) {
370 _FutureImpl future = new _FutureImpl(); 364 _FutureImpl future = new _FutureImpl();
371 StreamSubscription subscription; 365 StreamSubscription subscription;
372 subscription = this.listen( 366 subscription = this.listen(
373 // TODO(ahe): Restore type when feature is implemented in dart2js 367 (T element) {
374 // checked mode. http://dartbug.com/7733
375 (/*T*/ element) {
376 _runUserCode( 368 _runUserCode(
377 () => action(element), 369 () => action(element),
378 (_) {}, 370 (_) {},
379 _cancelAndError(subscription, future) 371 _cancelAndError(subscription, future)
380 ); 372 );
381 }, 373 },
382 onError: future._setError, 374 onError: future._setError,
383 onDone: () { 375 onDone: () {
384 future._setValue(null); 376 future._setValue(null);
385 }, 377 },
386 cancelOnError: true); 378 cancelOnError: true);
387 return future; 379 return future;
388 } 380 }
389 381
390 /** 382 /**
391 * Checks whether [test] accepts all elements provided by this stream. 383 * Checks whether [test] accepts all elements provided by this stream.
392 * 384 *
393 * Completes the [Future] when the answer is known. 385 * Completes the [Future] when the answer is known.
394 * If this stream reports an error, the [Future] will report that error. 386 * If this stream reports an error, the [Future] will report that error.
395 */ 387 */
396 Future<bool> every(bool test(T element)) { 388 Future<bool> every(bool test(T element)) {
397 _FutureImpl<bool> future = new _FutureImpl<bool>(); 389 _FutureImpl<bool> future = new _FutureImpl<bool>();
398 StreamSubscription subscription; 390 StreamSubscription subscription;
399 subscription = this.listen( 391 subscription = this.listen(
400 // TODO(ahe): Restore type when feature is implemented in dart2js 392 (T element) {
401 // checked mode. http://dartbug.com/7733
402 (/*T*/ element) {
403 _runUserCode( 393 _runUserCode(
404 () => test(element), 394 () => test(element),
405 (bool isMatch) { 395 (bool isMatch) {
406 if (!isMatch) { 396 if (!isMatch) {
407 subscription.cancel(); 397 subscription.cancel();
408 future._setValue(false); 398 future._setValue(false);
409 } 399 }
410 }, 400 },
411 _cancelAndError(subscription, future) 401 _cancelAndError(subscription, future)
412 ); 402 );
413 }, 403 },
414 onError: future._setError, 404 onError: future._setError,
415 onDone: () { 405 onDone: () {
416 future._setValue(true); 406 future._setValue(true);
417 }, 407 },
418 cancelOnError: true); 408 cancelOnError: true);
419 return future; 409 return future;
420 } 410 }
421 411
422 /** 412 /**
423 * Checks whether [test] accepts any element provided by this stream. 413 * Checks whether [test] accepts any element provided by this stream.
424 * 414 *
425 * Completes the [Future] when the answer is known. 415 * Completes the [Future] when the answer is known.
426 * If this stream reports an error, the [Future] will report that error. 416 * If this stream reports an error, the [Future] will report that error.
427 */ 417 */
428 Future<bool> any(bool test(T element)) { 418 Future<bool> any(bool test(T element)) {
429 _FutureImpl<bool> future = new _FutureImpl<bool>(); 419 _FutureImpl<bool> future = new _FutureImpl<bool>();
430 StreamSubscription subscription; 420 StreamSubscription subscription;
431 subscription = this.listen( 421 subscription = this.listen(
432 // TODO(ahe): Restore type when feature is implemented in dart2js 422 (T element) {
433 // checked mode. http://dartbug.com/7733
434 (/*T*/ element) {
435 _runUserCode( 423 _runUserCode(
436 () => test(element), 424 () => test(element),
437 (bool isMatch) { 425 (bool isMatch) {
438 if (isMatch) { 426 if (isMatch) {
439 subscription.cancel(); 427 subscription.cancel();
440 future._setValue(true); 428 future._setValue(true);
441 } 429 }
442 }, 430 },
443 _cancelAndError(subscription, future) 431 _cancelAndError(subscription, future)
444 ); 432 );
(...skipping 36 matching lines...) Expand 10 before | Expand all | Expand 10 after
481 }, 469 },
482 cancelOnError: true); 470 cancelOnError: true);
483 return future; 471 return future;
484 } 472 }
485 473
486 /** Collects the data of this stream in a [List]. */ 474 /** Collects the data of this stream in a [List]. */
487 Future<List<T>> toList() { 475 Future<List<T>> toList() {
488 List<T> result = <T>[]; 476 List<T> result = <T>[];
489 _FutureImpl<List<T>> future = new _FutureImpl<List<T>>(); 477 _FutureImpl<List<T>> future = new _FutureImpl<List<T>>();
490 this.listen( 478 this.listen(
491 // TODO(ahe): Restore type when feature is implemented in dart2js 479 (T data) {
492 // checked mode. http://dartbug.com/7733
493 (/*T*/ data) {
494 result.add(data); 480 result.add(data);
495 }, 481 },
496 onError: future._setError, 482 onError: future._setError,
497 onDone: () { 483 onDone: () {
498 future._setValue(result); 484 future._setValue(result);
499 }, 485 },
500 cancelOnError: true); 486 cancelOnError: true);
501 return future; 487 return future;
502 } 488 }
503 489
504 /** Collects the data of this stream in a [Set]. */ 490 /** Collects the data of this stream in a [Set]. */
505 Future<Set<T>> toSet() { 491 Future<Set<T>> toSet() {
506 Set<T> result = new Set<T>(); 492 Set<T> result = new Set<T>();
507 _FutureImpl<Set<T>> future = new _FutureImpl<Set<T>>(); 493 _FutureImpl<Set<T>> future = new _FutureImpl<Set<T>>();
508 this.listen( 494 this.listen(
509 // TODO(ahe): Restore type when feature is implemented in dart2js 495 (T data) {
510 // checked mode. http://dartbug.com/7733
511 (/*T*/ data) {
512 result.add(data); 496 result.add(data);
513 }, 497 },
514 onError: future._setError, 498 onError: future._setError,
515 onDone: () { 499 onDone: () {
516 future._setValue(result); 500 future._setValue(result);
517 }, 501 },
518 cancelOnError: true); 502 cancelOnError: true);
519 return future; 503 return future;
520 } 504 }
521 505
(...skipping 71 matching lines...) Expand 10 before | Expand all | Expand 10 after
593 /** 577 /**
594 * Returns the first element. 578 * Returns the first element.
595 * 579 *
596 * If [this] is empty throws a [StateError]. Otherwise this method is 580 * If [this] is empty throws a [StateError]. Otherwise this method is
597 * equivalent to [:this.elementAt(0):] 581 * equivalent to [:this.elementAt(0):]
598 */ 582 */
599 Future<T> get first { 583 Future<T> get first {
600 _FutureImpl<T> future = new _FutureImpl<T>(); 584 _FutureImpl<T> future = new _FutureImpl<T>();
601 StreamSubscription subscription; 585 StreamSubscription subscription;
602 subscription = this.listen( 586 subscription = this.listen(
603 // TODO(ahe): Restore type when feature is implemented in dart2js 587 (T value) {
604 // checked mode. http://dartbug.com/7733
605 (/*T*/ value) {
606 subscription.cancel(); 588 subscription.cancel();
607 future._setValue(value); 589 future._setValue(value);
608 return; 590 return;
609 }, 591 },
610 onError: future._setError, 592 onError: future._setError,
611 onDone: () { 593 onDone: () {
612 future._setError(new StateError("No elements")); 594 future._setError(new StateError("No elements"));
613 }, 595 },
614 cancelOnError: true); 596 cancelOnError: true);
615 return future; 597 return future;
616 } 598 }
617 599
618 /** 600 /**
619 * Returns the last element. 601 * Returns the last element.
620 * 602 *
621 * If [this] is empty throws a [StateError]. 603 * If [this] is empty throws a [StateError].
622 */ 604 */
623 Future<T> get last { 605 Future<T> get last {
624 _FutureImpl<T> future = new _FutureImpl<T>(); 606 _FutureImpl<T> future = new _FutureImpl<T>();
625 T result = null; 607 T result = null;
626 bool foundResult = false; 608 bool foundResult = false;
627 StreamSubscription subscription; 609 StreamSubscription subscription;
628 subscription = this.listen( 610 subscription = this.listen(
629 // TODO(ahe): Restore type when feature is implemented in dart2js 611 (T value) {
630 // checked mode. http://dartbug.com/7733
631 (/*T*/ value) {
632 foundResult = true; 612 foundResult = true;
633 result = value; 613 result = value;
634 }, 614 },
635 onError: future._setError, 615 onError: future._setError,
636 onDone: () { 616 onDone: () {
637 if (foundResult) { 617 if (foundResult) {
638 future._setValue(result); 618 future._setValue(result);
639 return; 619 return;
640 } 620 }
641 future._setError(new StateError("No elements")); 621 future._setError(new StateError("No elements"));
642 }, 622 },
643 cancelOnError: true); 623 cancelOnError: true);
644 return future; 624 return future;
645 } 625 }
646 626
647 /** 627 /**
648 * Returns the single element. 628 * Returns the single element.
649 * 629 *
650 * If [this] is empty or has more than one element throws a [StateError]. 630 * If [this] is empty or has more than one element throws a [StateError].
651 */ 631 */
652 Future<T> get single { 632 Future<T> get single {
653 _FutureImpl<T> future = new _FutureImpl<T>(); 633 _FutureImpl<T> future = new _FutureImpl<T>();
654 T result = null; 634 T result = null;
655 bool foundResult = false; 635 bool foundResult = false;
656 StreamSubscription subscription; 636 StreamSubscription subscription;
657 subscription = this.listen( 637 subscription = this.listen(
658 // TODO(ahe): Restore type when feature is implemented in dart2js 638 (T value) {
659 // checked mode. http://dartbug.com/7733
660 (/*T*/ value) {
661 if (foundResult) { 639 if (foundResult) {
662 subscription.cancel(); 640 subscription.cancel();
663 // This is the second element we get. 641 // This is the second element we get.
664 Error error = new StateError("More than one element"); 642 Error error = new StateError("More than one element");
665 future._setError(error); 643 future._setError(error);
666 return; 644 return;
667 } 645 }
668 foundResult = true; 646 foundResult = true;
669 result = value; 647 result = value;
670 }, 648 },
(...skipping 20 matching lines...) Expand all
691 * becomes the value of the future. 669 * becomes the value of the future.
692 * 670 *
693 * If an error occurs, or if this stream ends without finding a match and 671 * If an error occurs, or if this stream ends without finding a match and
694 * with no [defaultValue] function provided, the future will receive an 672 * with no [defaultValue] function provided, the future will receive an
695 * error. 673 * error.
696 */ 674 */
697 Future<dynamic> firstWhere(bool test(T element), {Object defaultValue()}) { 675 Future<dynamic> firstWhere(bool test(T element), {Object defaultValue()}) {
698 _FutureImpl<dynamic> future = new _FutureImpl(); 676 _FutureImpl<dynamic> future = new _FutureImpl();
699 StreamSubscription subscription; 677 StreamSubscription subscription;
700 subscription = this.listen( 678 subscription = this.listen(
701 // TODO(ahe): Restore type when feature is implemented in dart2js 679 (T value) {
702 // checked mode. http://dartbug.com/7733
703 (/*T*/ value) {
704 _runUserCode( 680 _runUserCode(
705 () => test(value), 681 () => test(value),
706 (bool isMatch) { 682 (bool isMatch) {
707 if (isMatch) { 683 if (isMatch) {
708 subscription.cancel(); 684 subscription.cancel();
709 future._setValue(value); 685 future._setValue(value);
710 } 686 }
711 }, 687 },
712 _cancelAndError(subscription, future) 688 _cancelAndError(subscription, future)
713 ); 689 );
(...skipping 16 matching lines...) Expand all
730 * As [firstWhere], except that the last matching element is found. 706 * As [firstWhere], except that the last matching element is found.
731 * That means that the result cannot be provided before this stream 707 * That means that the result cannot be provided before this stream
732 * is done. 708 * is done.
733 */ 709 */
734 Future<dynamic> lastWhere(bool test(T element), {Object defaultValue()}) { 710 Future<dynamic> lastWhere(bool test(T element), {Object defaultValue()}) {
735 _FutureImpl<dynamic> future = new _FutureImpl(); 711 _FutureImpl<dynamic> future = new _FutureImpl();
736 T result = null; 712 T result = null;
737 bool foundResult = false; 713 bool foundResult = false;
738 StreamSubscription subscription; 714 StreamSubscription subscription;
739 subscription = this.listen( 715 subscription = this.listen(
740 // TODO(ahe): Restore type when feature is implemented in dart2js 716 (T value) {
741 // checked mode. http://dartbug.com/7733
742 (/*T*/ value) {
743 _runUserCode( 717 _runUserCode(
744 () => true == test(value), 718 () => true == test(value),
745 (bool isMatch) { 719 (bool isMatch) {
746 if (isMatch) { 720 if (isMatch) {
747 foundResult = true; 721 foundResult = true;
748 result = value; 722 result = value;
749 } 723 }
750 }, 724 },
751 _cancelAndError(subscription, future) 725 _cancelAndError(subscription, future)
752 ); 726 );
(...skipping 19 matching lines...) Expand all
772 * 746 *
773 * Like [lastMatch], except that it is an error if more than one 747 * Like [lastMatch], except that it is an error if more than one
774 * matching element occurs in the stream. 748 * matching element occurs in the stream.
775 */ 749 */
776 Future<T> singleWhere(bool test(T element)) { 750 Future<T> singleWhere(bool test(T element)) {
777 _FutureImpl<T> future = new _FutureImpl<T>(); 751 _FutureImpl<T> future = new _FutureImpl<T>();
778 T result = null; 752 T result = null;
779 bool foundResult = false; 753 bool foundResult = false;
780 StreamSubscription subscription; 754 StreamSubscription subscription;
781 subscription = this.listen( 755 subscription = this.listen(
782 // TODO(ahe): Restore type when feature is implemented in dart2js 756 (T value) {
783 // checked mode. http://dartbug.com/7733
784 (/*T*/ value) {
785 _runUserCode( 757 _runUserCode(
786 () => true == test(value), 758 () => true == test(value),
787 (bool isMatch) { 759 (bool isMatch) {
788 if (isMatch) { 760 if (isMatch) {
789 if (foundResult) { 761 if (foundResult) {
790 subscription.cancel(); 762 subscription.cancel();
791 future._setError( 763 future._setError(
792 new StateError('Multiple matches for "single"')); 764 new StateError('Multiple matches for "single"'));
793 return; 765 return;
794 } 766 }
(...skipping 22 matching lines...) Expand all
817 * If an error event occurs, the future will end with this error. 789 * If an error event occurs, the future will end with this error.
818 * 790 *
819 * If this stream provides fewer than [index] elements before closing, 791 * If this stream provides fewer than [index] elements before closing,
820 * an error is reported. 792 * an error is reported.
821 */ 793 */
822 Future<T> elementAt(int index) { 794 Future<T> elementAt(int index) {
823 if (index is! int || index < 0) throw new ArgumentError(index); 795 if (index is! int || index < 0) throw new ArgumentError(index);
824 _FutureImpl<T> future = new _FutureImpl<T>(); 796 _FutureImpl<T> future = new _FutureImpl<T>();
825 StreamSubscription subscription; 797 StreamSubscription subscription;
826 subscription = this.listen( 798 subscription = this.listen(
827 // TODO(ahe): Restore type when feature is implemented in dart2js 799 (T value) {
828 // checked mode. http://dartbug.com/7733
829 (/*T*/ value) {
830 if (index == 0) { 800 if (index == 0) {
831 subscription.cancel(); 801 subscription.cancel();
832 future._setValue(value); 802 future._setValue(value);
833 return; 803 return;
834 } 804 }
835 index -= 1; 805 index -= 1;
836 }, 806 },
837 onError: future._setError, 807 onError: future._setError,
838 onDone: () { 808 onDone: () {
839 future._setError(new StateError("Not enough elements for elementAt")); 809 future._setError(new StateError("Not enough elements for elementAt"));
(...skipping 424 matching lines...) Expand 10 before | Expand all | Expand 10 after
1264 * 1234 *
1265 * If a [moveNext] call has been made, it will complete with `false` as value, 1235 * If a [moveNext] call has been made, it will complete with `false` as value,
1266 * as will all further calls to [moveNext]. 1236 * as will all further calls to [moveNext].
1267 * 1237 *
1268 * If you need to stop listening for values before the stream iterator is 1238 * If you need to stop listening for values before the stream iterator is
1269 * automatically closed, you must call [cancel] to ensure that the stream 1239 * automatically closed, you must call [cancel] to ensure that the stream
1270 * is properly closed. 1240 * is properly closed.
1271 */ 1241 */
1272 void cancel(); 1242 void cancel();
1273 } 1243 }
OLDNEW

Powered by Google App Engine
This is Rietveld 408576698