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

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: Updated cf. comment 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: () {
(...skipping 44 matching lines...) Expand 10 before | Expand all | Expand 10 after
368 /** 364 /**
369 * Checks whether [needle] occurs in the elements provided by this stream. 365 * Checks whether [needle] occurs in the elements provided by this stream.
370 * 366 *
371 * Completes the [Future] when the answer is known. 367 * Completes the [Future] when the answer is known.
372 * If this stream reports an error, the [Future] will report that error. 368 * If this stream reports an error, the [Future] will report that error.
373 */ 369 */
374 Future<bool> contains(Object needle) { 370 Future<bool> contains(Object needle) {
375 _FutureImpl<bool> future = new _FutureImpl<bool>(); 371 _FutureImpl<bool> future = new _FutureImpl<bool>();
376 StreamSubscription subscription; 372 StreamSubscription subscription;
377 subscription = this.listen( 373 subscription = this.listen(
378 // TODO(ahe): Restore type when feature is implemented in dart2js 374 (T element) {
379 // checked mode. http://dartbug.com/7733
380 (/*T*/ element) {
381 _runUserCode( 375 _runUserCode(
382 () => (element == needle), 376 () => (element == needle),
383 (bool isMatch) { 377 (bool isMatch) {
384 if (isMatch) { 378 if (isMatch) {
385 subscription.cancel(); 379 subscription.cancel();
386 future._setValue(true); 380 future._setValue(true);
387 } 381 }
388 }, 382 },
389 _cancelAndError(subscription, future) 383 _cancelAndError(subscription, future)
390 ); 384 );
(...skipping 10 matching lines...) Expand all
401 * Executes [action] on each data event of the stream. 395 * Executes [action] on each data event of the stream.
402 * 396 *
403 * Completes the returned [Future] when all events of the stream 397 * Completes the returned [Future] when all events of the stream
404 * have been processed. Completes the future with an error if the 398 * have been processed. Completes the future with an error if the
405 * stream has an error event, or if [action] throws. 399 * stream has an error event, or if [action] throws.
406 */ 400 */
407 Future forEach(void action(T element)) { 401 Future forEach(void action(T element)) {
408 _FutureImpl future = new _FutureImpl(); 402 _FutureImpl future = new _FutureImpl();
409 StreamSubscription subscription; 403 StreamSubscription subscription;
410 subscription = this.listen( 404 subscription = this.listen(
411 // TODO(ahe): Restore type when feature is implemented in dart2js 405 (T element) {
412 // checked mode. http://dartbug.com/7733
413 (/*T*/ element) {
414 _runUserCode( 406 _runUserCode(
415 () => action(element), 407 () => action(element),
416 (_) {}, 408 (_) {},
417 _cancelAndError(subscription, future) 409 _cancelAndError(subscription, future)
418 ); 410 );
419 }, 411 },
420 onError: future._setError, 412 onError: future._setError,
421 onDone: () { 413 onDone: () {
422 future._setValue(null); 414 future._setValue(null);
423 }, 415 },
424 cancelOnError: true); 416 cancelOnError: true);
425 return future; 417 return future;
426 } 418 }
427 419
428 /** 420 /**
429 * Checks whether [test] accepts all elements provided by this stream. 421 * Checks whether [test] accepts all elements provided by this stream.
430 * 422 *
431 * Completes the [Future] when the answer is known. 423 * Completes the [Future] when the answer is known.
432 * If this stream reports an error, the [Future] will report that error. 424 * If this stream reports an error, the [Future] will report that error.
433 */ 425 */
434 Future<bool> every(bool test(T element)) { 426 Future<bool> every(bool test(T element)) {
435 _FutureImpl<bool> future = new _FutureImpl<bool>(); 427 _FutureImpl<bool> future = new _FutureImpl<bool>();
436 StreamSubscription subscription; 428 StreamSubscription subscription;
437 subscription = this.listen( 429 subscription = this.listen(
438 // TODO(ahe): Restore type when feature is implemented in dart2js 430 (T element) {
439 // checked mode. http://dartbug.com/7733
440 (/*T*/ element) {
441 _runUserCode( 431 _runUserCode(
442 () => test(element), 432 () => test(element),
443 (bool isMatch) { 433 (bool isMatch) {
444 if (!isMatch) { 434 if (!isMatch) {
445 subscription.cancel(); 435 subscription.cancel();
446 future._setValue(false); 436 future._setValue(false);
447 } 437 }
448 }, 438 },
449 _cancelAndError(subscription, future) 439 _cancelAndError(subscription, future)
450 ); 440 );
451 }, 441 },
452 onError: future._setError, 442 onError: future._setError,
453 onDone: () { 443 onDone: () {
454 future._setValue(true); 444 future._setValue(true);
455 }, 445 },
456 cancelOnError: true); 446 cancelOnError: true);
457 return future; 447 return future;
458 } 448 }
459 449
460 /** 450 /**
461 * Checks whether [test] accepts any element provided by this stream. 451 * Checks whether [test] accepts any element provided by this stream.
462 * 452 *
463 * Completes the [Future] when the answer is known. 453 * Completes the [Future] when the answer is known.
464 * If this stream reports an error, the [Future] will report that error. 454 * If this stream reports an error, the [Future] will report that error.
465 */ 455 */
466 Future<bool> any(bool test(T element)) { 456 Future<bool> any(bool test(T element)) {
467 _FutureImpl<bool> future = new _FutureImpl<bool>(); 457 _FutureImpl<bool> future = new _FutureImpl<bool>();
468 StreamSubscription subscription; 458 StreamSubscription subscription;
469 subscription = this.listen( 459 subscription = this.listen(
470 // TODO(ahe): Restore type when feature is implemented in dart2js 460 (T element) {
471 // checked mode. http://dartbug.com/7733
472 (/*T*/ element) {
473 _runUserCode( 461 _runUserCode(
474 () => test(element), 462 () => test(element),
475 (bool isMatch) { 463 (bool isMatch) {
476 if (isMatch) { 464 if (isMatch) {
477 subscription.cancel(); 465 subscription.cancel();
478 future._setValue(true); 466 future._setValue(true);
479 } 467 }
480 }, 468 },
481 _cancelAndError(subscription, future) 469 _cancelAndError(subscription, future)
482 ); 470 );
(...skipping 36 matching lines...) Expand 10 before | Expand all | Expand 10 after
519 }, 507 },
520 cancelOnError: true); 508 cancelOnError: true);
521 return future; 509 return future;
522 } 510 }
523 511
524 /** Collects the data of this stream in a [List]. */ 512 /** Collects the data of this stream in a [List]. */
525 Future<List<T>> toList() { 513 Future<List<T>> toList() {
526 List<T> result = <T>[]; 514 List<T> result = <T>[];
527 _FutureImpl<List<T>> future = new _FutureImpl<List<T>>(); 515 _FutureImpl<List<T>> future = new _FutureImpl<List<T>>();
528 this.listen( 516 this.listen(
529 // TODO(ahe): Restore type when feature is implemented in dart2js 517 (T data) {
530 // checked mode. http://dartbug.com/7733
531 (/*T*/ data) {
532 result.add(data); 518 result.add(data);
533 }, 519 },
534 onError: future._setError, 520 onError: future._setError,
535 onDone: () { 521 onDone: () {
536 future._setValue(result); 522 future._setValue(result);
537 }, 523 },
538 cancelOnError: true); 524 cancelOnError: true);
539 return future; 525 return future;
540 } 526 }
541 527
542 /** Collects the data of this stream in a [Set]. */ 528 /** Collects the data of this stream in a [Set]. */
543 Future<Set<T>> toSet() { 529 Future<Set<T>> toSet() {
544 Set<T> result = new Set<T>(); 530 Set<T> result = new Set<T>();
545 _FutureImpl<Set<T>> future = new _FutureImpl<Set<T>>(); 531 _FutureImpl<Set<T>> future = new _FutureImpl<Set<T>>();
546 this.listen( 532 this.listen(
547 // TODO(ahe): Restore type when feature is implemented in dart2js 533 (T data) {
548 // checked mode. http://dartbug.com/7733
549 (/*T*/ data) {
550 result.add(data); 534 result.add(data);
551 }, 535 },
552 onError: future._setError, 536 onError: future._setError,
553 onDone: () { 537 onDone: () {
554 future._setValue(result); 538 future._setValue(result);
555 }, 539 },
556 cancelOnError: true); 540 cancelOnError: true);
557 return future; 541 return future;
558 } 542 }
559 543
(...skipping 71 matching lines...) Expand 10 before | Expand all | Expand 10 after
631 /** 615 /**
632 * Returns the first element. 616 * Returns the first element.
633 * 617 *
634 * If [this] is empty throws a [StateError]. Otherwise this method is 618 * If [this] is empty throws a [StateError]. Otherwise this method is
635 * equivalent to [:this.elementAt(0):] 619 * equivalent to [:this.elementAt(0):]
636 */ 620 */
637 Future<T> get first { 621 Future<T> get first {
638 _FutureImpl<T> future = new _FutureImpl<T>(); 622 _FutureImpl<T> future = new _FutureImpl<T>();
639 StreamSubscription subscription; 623 StreamSubscription subscription;
640 subscription = this.listen( 624 subscription = this.listen(
641 // TODO(ahe): Restore type when feature is implemented in dart2js 625 (T value) {
642 // checked mode. http://dartbug.com/7733
643 (/*T*/ value) {
644 subscription.cancel(); 626 subscription.cancel();
645 future._setValue(value); 627 future._setValue(value);
646 return; 628 return;
647 }, 629 },
648 onError: future._setError, 630 onError: future._setError,
649 onDone: () { 631 onDone: () {
650 future._setError(new StateError("No elements")); 632 future._setError(new StateError("No elements"));
651 }, 633 },
652 cancelOnError: true); 634 cancelOnError: true);
653 return future; 635 return future;
654 } 636 }
655 637
656 /** 638 /**
657 * Returns the last element. 639 * Returns the last element.
658 * 640 *
659 * If [this] is empty throws a [StateError]. 641 * If [this] is empty throws a [StateError].
660 */ 642 */
661 Future<T> get last { 643 Future<T> get last {
662 _FutureImpl<T> future = new _FutureImpl<T>(); 644 _FutureImpl<T> future = new _FutureImpl<T>();
663 T result = null; 645 T result = null;
664 bool foundResult = false; 646 bool foundResult = false;
665 StreamSubscription subscription; 647 StreamSubscription subscription;
666 subscription = this.listen( 648 subscription = this.listen(
667 // TODO(ahe): Restore type when feature is implemented in dart2js 649 (T value) {
668 // checked mode. http://dartbug.com/7733
669 (/*T*/ value) {
670 foundResult = true; 650 foundResult = true;
671 result = value; 651 result = value;
672 }, 652 },
673 onError: future._setError, 653 onError: future._setError,
674 onDone: () { 654 onDone: () {
675 if (foundResult) { 655 if (foundResult) {
676 future._setValue(result); 656 future._setValue(result);
677 return; 657 return;
678 } 658 }
679 future._setError(new StateError("No elements")); 659 future._setError(new StateError("No elements"));
680 }, 660 },
681 cancelOnError: true); 661 cancelOnError: true);
682 return future; 662 return future;
683 } 663 }
684 664
685 /** 665 /**
686 * Returns the single element. 666 * Returns the single element.
687 * 667 *
688 * If [this] is empty or has more than one element throws a [StateError]. 668 * If [this] is empty or has more than one element throws a [StateError].
689 */ 669 */
690 Future<T> get single { 670 Future<T> get single {
691 _FutureImpl<T> future = new _FutureImpl<T>(); 671 _FutureImpl<T> future = new _FutureImpl<T>();
692 T result = null; 672 T result = null;
693 bool foundResult = false; 673 bool foundResult = false;
694 StreamSubscription subscription; 674 StreamSubscription subscription;
695 subscription = this.listen( 675 subscription = this.listen(
696 // TODO(ahe): Restore type when feature is implemented in dart2js 676 (T value) {
697 // checked mode. http://dartbug.com/7733
698 (/*T*/ value) {
699 if (foundResult) { 677 if (foundResult) {
700 subscription.cancel(); 678 subscription.cancel();
701 // This is the second element we get. 679 // This is the second element we get.
702 Error error = new StateError("More than one element"); 680 Error error = new StateError("More than one element");
703 future._setError(error); 681 future._setError(error);
704 return; 682 return;
705 } 683 }
706 foundResult = true; 684 foundResult = true;
707 result = value; 685 result = value;
708 }, 686 },
(...skipping 20 matching lines...) Expand all
729 * becomes the value of the future. 707 * becomes the value of the future.
730 * 708 *
731 * If an error occurs, or if this stream ends without finding a match and 709 * If an error occurs, or if this stream ends without finding a match and
732 * with no [defaultValue] function provided, the future will receive an 710 * with no [defaultValue] function provided, the future will receive an
733 * error. 711 * error.
734 */ 712 */
735 Future<dynamic> firstWhere(bool test(T element), {Object defaultValue()}) { 713 Future<dynamic> firstWhere(bool test(T element), {Object defaultValue()}) {
736 _FutureImpl<dynamic> future = new _FutureImpl(); 714 _FutureImpl<dynamic> future = new _FutureImpl();
737 StreamSubscription subscription; 715 StreamSubscription subscription;
738 subscription = this.listen( 716 subscription = this.listen(
739 // TODO(ahe): Restore type when feature is implemented in dart2js 717 (T value) {
740 // checked mode. http://dartbug.com/7733
741 (/*T*/ value) {
742 _runUserCode( 718 _runUserCode(
743 () => test(value), 719 () => test(value),
744 (bool isMatch) { 720 (bool isMatch) {
745 if (isMatch) { 721 if (isMatch) {
746 subscription.cancel(); 722 subscription.cancel();
747 future._setValue(value); 723 future._setValue(value);
748 } 724 }
749 }, 725 },
750 _cancelAndError(subscription, future) 726 _cancelAndError(subscription, future)
751 ); 727 );
(...skipping 16 matching lines...) Expand all
768 * As [firstWhere], except that the last matching element is found. 744 * As [firstWhere], except that the last matching element is found.
769 * That means that the result cannot be provided before this stream 745 * That means that the result cannot be provided before this stream
770 * is done. 746 * is done.
771 */ 747 */
772 Future<dynamic> lastWhere(bool test(T element), {Object defaultValue()}) { 748 Future<dynamic> lastWhere(bool test(T element), {Object defaultValue()}) {
773 _FutureImpl<dynamic> future = new _FutureImpl(); 749 _FutureImpl<dynamic> future = new _FutureImpl();
774 T result = null; 750 T result = null;
775 bool foundResult = false; 751 bool foundResult = false;
776 StreamSubscription subscription; 752 StreamSubscription subscription;
777 subscription = this.listen( 753 subscription = this.listen(
778 // TODO(ahe): Restore type when feature is implemented in dart2js 754 (T value) {
779 // checked mode. http://dartbug.com/7733
780 (/*T*/ value) {
781 _runUserCode( 755 _runUserCode(
782 () => true == test(value), 756 () => true == test(value),
783 (bool isMatch) { 757 (bool isMatch) {
784 if (isMatch) { 758 if (isMatch) {
785 foundResult = true; 759 foundResult = true;
786 result = value; 760 result = value;
787 } 761 }
788 }, 762 },
789 _cancelAndError(subscription, future) 763 _cancelAndError(subscription, future)
790 ); 764 );
(...skipping 19 matching lines...) Expand all
810 * 784 *
811 * Like [lastMatch], except that it is an error if more than one 785 * Like [lastMatch], except that it is an error if more than one
812 * matching element occurs in the stream. 786 * matching element occurs in the stream.
813 */ 787 */
814 Future<T> singleWhere(bool test(T element)) { 788 Future<T> singleWhere(bool test(T element)) {
815 _FutureImpl<T> future = new _FutureImpl<T>(); 789 _FutureImpl<T> future = new _FutureImpl<T>();
816 T result = null; 790 T result = null;
817 bool foundResult = false; 791 bool foundResult = false;
818 StreamSubscription subscription; 792 StreamSubscription subscription;
819 subscription = this.listen( 793 subscription = this.listen(
820 // TODO(ahe): Restore type when feature is implemented in dart2js 794 (T value) {
821 // checked mode. http://dartbug.com/7733
822 (/*T*/ value) {
823 _runUserCode( 795 _runUserCode(
824 () => true == test(value), 796 () => true == test(value),
825 (bool isMatch) { 797 (bool isMatch) {
826 if (isMatch) { 798 if (isMatch) {
827 if (foundResult) { 799 if (foundResult) {
828 subscription.cancel(); 800 subscription.cancel();
829 future._setError( 801 future._setError(
830 new StateError('Multiple matches for "single"')); 802 new StateError('Multiple matches for "single"'));
831 return; 803 return;
832 } 804 }
(...skipping 22 matching lines...) Expand all
855 * If an error event occurs, the future will end with this error. 827 * If an error event occurs, the future will end with this error.
856 * 828 *
857 * If this stream provides fewer than [index] elements before closing, 829 * If this stream provides fewer than [index] elements before closing,
858 * an error is reported. 830 * an error is reported.
859 */ 831 */
860 Future<T> elementAt(int index) { 832 Future<T> elementAt(int index) {
861 if (index is! int || index < 0) throw new ArgumentError(index); 833 if (index is! int || index < 0) throw new ArgumentError(index);
862 _FutureImpl<T> future = new _FutureImpl<T>(); 834 _FutureImpl<T> future = new _FutureImpl<T>();
863 StreamSubscription subscription; 835 StreamSubscription subscription;
864 subscription = this.listen( 836 subscription = this.listen(
865 // TODO(ahe): Restore type when feature is implemented in dart2js 837 (T value) {
866 // checked mode. http://dartbug.com/7733
867 (/*T*/ value) {
868 if (index == 0) { 838 if (index == 0) {
869 subscription.cancel(); 839 subscription.cancel();
870 future._setValue(value); 840 future._setValue(value);
871 return; 841 return;
872 } 842 }
873 index -= 1; 843 index -= 1;
874 }, 844 },
875 onError: future._setError, 845 onError: future._setError,
876 onDone: () { 846 onDone: () {
877 future._setError(new StateError("Not enough elements for elementAt")); 847 future._setError(new StateError("Not enough elements for elementAt"));
(...skipping 411 matching lines...) Expand 10 before | Expand all | Expand 10 after
1289 * 1259 *
1290 * If a [moveNext] call has been made, it will complete with `false` as value, 1260 * If a [moveNext] call has been made, it will complete with `false` as value,
1291 * as will all further calls to [moveNext]. 1261 * as will all further calls to [moveNext].
1292 * 1262 *
1293 * If you need to stop listening for values before the stream iterator is 1263 * If you need to stop listening for values before the stream iterator is
1294 * automatically closed, you must call [cancel] to ensure that the stream 1264 * automatically closed, you must call [cancel] to ensure that the stream
1295 * is properly closed. 1265 * is properly closed.
1296 */ 1266 */
1297 void cancel(); 1267 void cancel();
1298 } 1268 }
OLDNEW
« no previous file with comments | « sdk/lib/_internal/compiler/implementation/js_backend/emitter.dart ('k') | sdk/lib/async/stream_impl.dart » ('j') | no next file with comments »

Powered by Google App Engine
This is Rietveld 408576698