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

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

Issue 23967011: Revert "Rewrite Futures." (Closed) Base URL: https://dart.googlecode.com/svn/branches/bleeding_edge/dart
Patch Set: Created 7 years, 3 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
« no previous file with comments | « sdk/lib/async/future_impl.dart ('k') | sdk/lib/async/stream_controller.dart » ('j') | no next file with comments »
Toggle Intra-line Diffs ('i') | Expand Comments ('e') | Collapse Comments ('c') | Show Comments Hide Comments ('s')
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 254 matching lines...) Expand 10 before | Expand all | Expand 10 after
265 * Returns the result of [:streamTransformer.bind:] itself. 265 * Returns the result of [:streamTransformer.bind:] itself.
266 */ 266 */
267 Stream transform(StreamTransformer<T, dynamic> streamTransformer) { 267 Stream transform(StreamTransformer<T, dynamic> streamTransformer) {
268 return streamTransformer.bind(this); 268 return streamTransformer.bind(this);
269 } 269 }
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 _Future<T> result = new _Future<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 (T element) { 280 (T element) {
281 if (seenFirst) { 281 if (seenFirst) {
282 _runUserCode(() => combine(value, element), 282 _runUserCode(() => combine(value, element),
283 (T newValue) { value = newValue; }, 283 (T newValue) { value = newValue; },
284 _cancelAndError(subscription, result)); 284 _cancelAndError(subscription, result));
285 } else { 285 } else {
286 value = element; 286 value = element;
287 seenFirst = true; 287 seenFirst = true;
288 } 288 }
289 }, 289 },
290 onError: result._completeError, 290 onError: result._setError,
291 onDone: () { 291 onDone: () {
292 if (!seenFirst) { 292 if (!seenFirst) {
293 result._completeError(new StateError("No elements")); 293 result._setError(new StateError("No elements"));
294 } else { 294 } else {
295 result._complete(value); 295 result._setValue(value);
296 } 296 }
297 }, 297 },
298 cancelOnError: true 298 cancelOnError: true
299 ); 299 );
300 return result; 300 return result;
301 } 301 }
302 302
303 /** Reduces a sequence of values by repeatedly applying [combine]. */ 303 /** Reduces a sequence of values by repeatedly applying [combine]. */
304 Future fold(var initialValue, combine(var previous, T element)) { 304 Future fold(var initialValue, combine(var previous, T element)) {
305 _Future result = new _Future(); 305 _FutureImpl result = new _FutureImpl();
306 var value = initialValue; 306 var value = initialValue;
307 StreamSubscription subscription; 307 StreamSubscription subscription;
308 subscription = this.listen( 308 subscription = this.listen(
309 (T element) { 309 (T element) {
310 _runUserCode( 310 _runUserCode(
311 () => combine(value, element), 311 () => combine(value, element),
312 (newValue) { value = newValue; }, 312 (newValue) { value = newValue; },
313 _cancelAndError(subscription, result) 313 _cancelAndError(subscription, result)
314 ); 314 );
315 }, 315 },
316 onError: (e) { 316 onError: (e) {
317 result._completeError(e); 317 result._setError(e);
318 }, 318 },
319 onDone: () { 319 onDone: () {
320 result._complete(value); 320 result._setValue(value);
321 }, 321 },
322 cancelOnError: true); 322 cancelOnError: true);
323 return result; 323 return result;
324 } 324 }
325 325
326 /** 326 /**
327 * Collects string of data events' string representations. 327 * Collects string of data events' string representations.
328 * 328 *
329 * If [separator] is provided, it is inserted between any two 329 * If [separator] is provided, it is inserted between any two
330 * elements. 330 * elements.
331 * 331 *
332 * Any error in the stream causes the future to complete with that 332 * Any error in the stream causes the future to complete with that
333 * error. Otherwise it completes with the collected string when 333 * error. Otherwise it completes with the collected string when
334 * the "done" event arrives. 334 * the "done" event arrives.
335 */ 335 */
336 Future<String> join([String separator = ""]) { 336 Future<String> join([String separator = ""]) {
337 _Future<String> result = new _Future<String>(); 337 _FutureImpl<String> result = new _FutureImpl<String>();
338 StringBuffer buffer = new StringBuffer(); 338 StringBuffer buffer = new StringBuffer();
339 StreamSubscription subscription; 339 StreamSubscription subscription;
340 bool first = true; 340 bool first = true;
341 subscription = this.listen( 341 subscription = this.listen(
342 (T element) { 342 (T element) {
343 if (!first) { 343 if (!first) {
344 buffer.write(separator); 344 buffer.write(separator);
345 } 345 }
346 first = false; 346 first = false;
347 try { 347 try {
348 buffer.write(element); 348 buffer.write(element);
349 } catch (e, s) { 349 } catch (e, s) {
350 subscription.cancel(); 350 subscription.cancel();
351 result._completeError(_asyncError(e, s)); 351 result._setError(_asyncError(e, s));
352 } 352 }
353 }, 353 },
354 onError: (e) { 354 onError: (e) {
355 result._completeError(e); 355 result._setError(e);
356 }, 356 },
357 onDone: () { 357 onDone: () {
358 result._complete(buffer.toString()); 358 result._setValue(buffer.toString());
359 }, 359 },
360 cancelOnError: true); 360 cancelOnError: true);
361 return result; 361 return result;
362 } 362 }
363 363
364 /** 364 /**
365 * Checks whether [needle] occurs in the elements provided by this stream. 365 * Checks whether [needle] occurs in the elements provided by this stream.
366 * 366 *
367 * Completes the [Future] when the answer is known. 367 * Completes the [Future] when the answer is known.
368 * 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.
369 */ 369 */
370 Future<bool> contains(Object needle) { 370 Future<bool> contains(Object needle) {
371 _Future<bool> future = new _Future<bool>(); 371 _FutureImpl<bool> future = new _FutureImpl<bool>();
372 StreamSubscription subscription; 372 StreamSubscription subscription;
373 subscription = this.listen( 373 subscription = this.listen(
374 (T element) { 374 (T element) {
375 _runUserCode( 375 _runUserCode(
376 () => (element == needle), 376 () => (element == needle),
377 (bool isMatch) { 377 (bool isMatch) {
378 if (isMatch) { 378 if (isMatch) {
379 subscription.cancel(); 379 subscription.cancel();
380 future._complete(true); 380 future._setValue(true);
381 } 381 }
382 }, 382 },
383 _cancelAndError(subscription, future) 383 _cancelAndError(subscription, future)
384 ); 384 );
385 }, 385 },
386 onError: future._completeError, 386 onError: future._setError,
387 onDone: () { 387 onDone: () {
388 future._complete(false); 388 future._setValue(false);
389 }, 389 },
390 cancelOnError: true); 390 cancelOnError: true);
391 return future; 391 return future;
392 } 392 }
393 393
394 /** 394 /**
395 * Executes [action] on each data event of the stream. 395 * Executes [action] on each data event of the stream.
396 * 396 *
397 * Completes the returned [Future] when all events of the stream 397 * Completes the returned [Future] when all events of the stream
398 * have been processed. Completes the future with an error if the 398 * have been processed. Completes the future with an error if the
399 * stream has an error event, or if [action] throws. 399 * stream has an error event, or if [action] throws.
400 */ 400 */
401 Future forEach(void action(T element)) { 401 Future forEach(void action(T element)) {
402 _Future future = new _Future(); 402 _FutureImpl future = new _FutureImpl();
403 StreamSubscription subscription; 403 StreamSubscription subscription;
404 subscription = this.listen( 404 subscription = this.listen(
405 (T element) { 405 (T element) {
406 _runUserCode( 406 _runUserCode(
407 () => action(element), 407 () => action(element),
408 (_) {}, 408 (_) {},
409 _cancelAndError(subscription, future) 409 _cancelAndError(subscription, future)
410 ); 410 );
411 }, 411 },
412 onError: future._completeError, 412 onError: future._setError,
413 onDone: () { 413 onDone: () {
414 future._complete(null); 414 future._setValue(null);
415 }, 415 },
416 cancelOnError: true); 416 cancelOnError: true);
417 return future; 417 return future;
418 } 418 }
419 419
420 /** 420 /**
421 * Checks whether [test] accepts all elements provided by this stream. 421 * Checks whether [test] accepts all elements provided by this stream.
422 * 422 *
423 * Completes the [Future] when the answer is known. 423 * Completes the [Future] when the answer is known.
424 * 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.
425 */ 425 */
426 Future<bool> every(bool test(T element)) { 426 Future<bool> every(bool test(T element)) {
427 _Future<bool> future = new _Future<bool>(); 427 _FutureImpl<bool> future = new _FutureImpl<bool>();
428 StreamSubscription subscription; 428 StreamSubscription subscription;
429 subscription = this.listen( 429 subscription = this.listen(
430 (T element) { 430 (T element) {
431 _runUserCode( 431 _runUserCode(
432 () => test(element), 432 () => test(element),
433 (bool isMatch) { 433 (bool isMatch) {
434 if (!isMatch) { 434 if (!isMatch) {
435 subscription.cancel(); 435 subscription.cancel();
436 future._complete(false); 436 future._setValue(false);
437 } 437 }
438 }, 438 },
439 _cancelAndError(subscription, future) 439 _cancelAndError(subscription, future)
440 ); 440 );
441 }, 441 },
442 onError: future._completeError, 442 onError: future._setError,
443 onDone: () { 443 onDone: () {
444 future._complete(true); 444 future._setValue(true);
445 }, 445 },
446 cancelOnError: true); 446 cancelOnError: true);
447 return future; 447 return future;
448 } 448 }
449 449
450 /** 450 /**
451 * Checks whether [test] accepts any element provided by this stream. 451 * Checks whether [test] accepts any element provided by this stream.
452 * 452 *
453 * Completes the [Future] when the answer is known. 453 * Completes the [Future] when the answer is known.
454 * 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.
455 */ 455 */
456 Future<bool> any(bool test(T element)) { 456 Future<bool> any(bool test(T element)) {
457 _Future<bool> future = new _Future<bool>(); 457 _FutureImpl<bool> future = new _FutureImpl<bool>();
458 StreamSubscription subscription; 458 StreamSubscription subscription;
459 subscription = this.listen( 459 subscription = this.listen(
460 (T element) { 460 (T element) {
461 _runUserCode( 461 _runUserCode(
462 () => test(element), 462 () => test(element),
463 (bool isMatch) { 463 (bool isMatch) {
464 if (isMatch) { 464 if (isMatch) {
465 subscription.cancel(); 465 subscription.cancel();
466 future._complete(true); 466 future._setValue(true);
467 } 467 }
468 }, 468 },
469 _cancelAndError(subscription, future) 469 _cancelAndError(subscription, future)
470 ); 470 );
471 }, 471 },
472 onError: future._completeError, 472 onError: future._setError,
473 onDone: () { 473 onDone: () {
474 future._complete(false); 474 future._setValue(false);
475 }, 475 },
476 cancelOnError: true); 476 cancelOnError: true);
477 return future; 477 return future;
478 } 478 }
479 479
480 480
481 /** Counts the elements in the stream. */ 481 /** Counts the elements in the stream. */
482 Future<int> get length { 482 Future<int> get length {
483 _Future<int> future = new _Future<int>(); 483 _FutureImpl<int> future = new _FutureImpl<int>();
484 int count = 0; 484 int count = 0;
485 this.listen( 485 this.listen(
486 (_) { count++; }, 486 (_) { count++; },
487 onError: future._completeError, 487 onError: future._setError,
488 onDone: () { 488 onDone: () {
489 future._complete(count); 489 future._setValue(count);
490 }, 490 },
491 cancelOnError: true); 491 cancelOnError: true);
492 return future; 492 return future;
493 } 493 }
494 494
495 /** Reports whether this stream contains any elements. */ 495 /** Reports whether this stream contains any elements. */
496 Future<bool> get isEmpty { 496 Future<bool> get isEmpty {
497 _Future<bool> future = new _Future<bool>(); 497 _FutureImpl<bool> future = new _FutureImpl<bool>();
498 StreamSubscription subscription; 498 StreamSubscription subscription;
499 subscription = this.listen( 499 subscription = this.listen(
500 (_) { 500 (_) {
501 subscription.cancel(); 501 subscription.cancel();
502 future._complete(false); 502 future._setValue(false);
503 }, 503 },
504 onError: future._completeError, 504 onError: future._setError,
505 onDone: () { 505 onDone: () {
506 future._complete(true); 506 future._setValue(true);
507 }, 507 },
508 cancelOnError: true); 508 cancelOnError: true);
509 return future; 509 return future;
510 } 510 }
511 511
512 /** Collects the data of this stream in a [List]. */ 512 /** Collects the data of this stream in a [List]. */
513 Future<List<T>> toList() { 513 Future<List<T>> toList() {
514 List<T> result = <T>[]; 514 List<T> result = <T>[];
515 _Future<List<T>> future = new _Future<List<T>>(); 515 _FutureImpl<List<T>> future = new _FutureImpl<List<T>>();
516 this.listen( 516 this.listen(
517 (T data) { 517 (T data) {
518 result.add(data); 518 result.add(data);
519 }, 519 },
520 onError: future._completeError, 520 onError: future._setError,
521 onDone: () { 521 onDone: () {
522 future._complete(result); 522 future._setValue(result);
523 }, 523 },
524 cancelOnError: true); 524 cancelOnError: true);
525 return future; 525 return future;
526 } 526 }
527 527
528 /** Collects the data of this stream in a [Set]. */ 528 /** Collects the data of this stream in a [Set]. */
529 Future<Set<T>> toSet() { 529 Future<Set<T>> toSet() {
530 Set<T> result = new Set<T>(); 530 Set<T> result = new Set<T>();
531 _Future<Set<T>> future = new _Future<Set<T>>(); 531 _FutureImpl<Set<T>> future = new _FutureImpl<Set<T>>();
532 this.listen( 532 this.listen(
533 (T data) { 533 (T data) {
534 result.add(data); 534 result.add(data);
535 }, 535 },
536 onError: future._completeError, 536 onError: future._setError,
537 onDone: () { 537 onDone: () {
538 future._complete(result); 538 future._setValue(result);
539 }, 539 },
540 cancelOnError: true); 540 cancelOnError: true);
541 return future; 541 return future;
542 } 542 }
543 543
544 /** 544 /**
545 * Discards all data on the stream, but signals when it's done or an error 545 * Discards all data on the stream, but signals when it's done or an error
546 * occured. 546 * occured.
547 * 547 *
548 * When subscribing using [drain], cancelOnError will be true. This means 548 * When subscribing using [drain], cancelOnError will be true. This means
(...skipping 71 matching lines...) Expand 10 before | Expand all | Expand 10 after
620 * If an error event occurs before the first data event, the resulting future 620 * If an error event occurs before the first data event, the resulting future
621 * is completed with that error. 621 * is completed with that error.
622 * 622 *
623 * If this stream is empty (a done event occurs before the first data event), 623 * If this stream is empty (a done event occurs before the first data event),
624 * the resulting future completes with a [StateError]. 624 * the resulting future completes with a [StateError].
625 * 625 *
626 * Except for the type of the error, this method is equivalent to 626 * Except for the type of the error, this method is equivalent to
627 * [:this.elementAt(0):]. 627 * [:this.elementAt(0):].
628 */ 628 */
629 Future<T> get first { 629 Future<T> get first {
630 _Future<T> future = new _Future<T>(); 630 _FutureImpl<T> future = new _FutureImpl<T>();
631 StreamSubscription subscription; 631 StreamSubscription subscription;
632 subscription = this.listen( 632 subscription = this.listen(
633 (T value) { 633 (T value) {
634 subscription.cancel(); 634 subscription.cancel();
635 future._complete(value); 635 future._setValue(value);
636 return; 636 return;
637 }, 637 },
638 onError: future._completeError, 638 onError: future._setError,
639 onDone: () { 639 onDone: () {
640 future._completeError(new StateError("No elements")); 640 future._setError(new StateError("No elements"));
641 }, 641 },
642 cancelOnError: true); 642 cancelOnError: true);
643 return future; 643 return future;
644 } 644 }
645 645
646 /** 646 /**
647 * Returns the last element of the stream. 647 * Returns the last element of the stream.
648 * 648 *
649 * If an error event occurs before the first data event, the resulting future 649 * If an error event occurs before the first data event, the resulting future
650 * is completed with that error. 650 * is completed with that error.
651 * 651 *
652 * If this stream is empty (a done event occurs before the first data event), 652 * If this stream is empty (a done event occurs before the first data event),
653 * the resulting future completes with a [StateError]. 653 * the resulting future completes with a [StateError].
654 */ 654 */
655 Future<T> get last { 655 Future<T> get last {
656 _Future<T> future = new _Future<T>(); 656 _FutureImpl<T> future = new _FutureImpl<T>();
657 T result = null; 657 T result = null;
658 bool foundResult = false; 658 bool foundResult = false;
659 StreamSubscription subscription; 659 StreamSubscription subscription;
660 subscription = this.listen( 660 subscription = this.listen(
661 (T value) { 661 (T value) {
662 foundResult = true; 662 foundResult = true;
663 result = value; 663 result = value;
664 }, 664 },
665 onError: future._completeError, 665 onError: future._setError,
666 onDone: () { 666 onDone: () {
667 if (foundResult) { 667 if (foundResult) {
668 future._complete(result); 668 future._setValue(result);
669 return; 669 return;
670 } 670 }
671 future._completeError(new StateError("No elements")); 671 future._setError(new StateError("No elements"));
672 }, 672 },
673 cancelOnError: true); 673 cancelOnError: true);
674 return future; 674 return future;
675 } 675 }
676 676
677 /** 677 /**
678 * Returns the single element. 678 * Returns the single element.
679 * 679 *
680 * If [this] is empty or has more than one element throws a [StateError]. 680 * If [this] is empty or has more than one element throws a [StateError].
681 */ 681 */
682 Future<T> get single { 682 Future<T> get single {
683 _Future<T> future = new _Future<T>(); 683 _FutureImpl<T> future = new _FutureImpl<T>();
684 T result = null; 684 T result = null;
685 bool foundResult = false; 685 bool foundResult = false;
686 StreamSubscription subscription; 686 StreamSubscription subscription;
687 subscription = this.listen( 687 subscription = this.listen(
688 (T value) { 688 (T value) {
689 if (foundResult) { 689 if (foundResult) {
690 subscription.cancel(); 690 subscription.cancel();
691 // This is the second element we get. 691 // This is the second element we get.
692 Error error = new StateError("More than one element"); 692 Error error = new StateError("More than one element");
693 future._completeError(error); 693 future._setError(error);
694 return; 694 return;
695 } 695 }
696 foundResult = true; 696 foundResult = true;
697 result = value; 697 result = value;
698 }, 698 },
699 onError: future._completeError, 699 onError: future._setError,
700 onDone: () { 700 onDone: () {
701 if (foundResult) { 701 if (foundResult) {
702 future._complete(result); 702 future._setValue(result);
703 return; 703 return;
704 } 704 }
705 future._completeError(new StateError("No elements")); 705 future._setError(new StateError("No elements"));
706 }, 706 },
707 cancelOnError: true); 707 cancelOnError: true);
708 return future; 708 return future;
709 } 709 }
710 710
711 /** 711 /**
712 * Finds the first element of this stream matching [test]. 712 * Finds the first element of this stream matching [test].
713 * 713 *
714 * Returns a future that is filled with the first element of this stream 714 * Returns a future that is filled with the first element of this stream
715 * that [test] returns true for. 715 * that [test] returns true for.
716 * 716 *
717 * If no such element is found before this stream is done, and a 717 * If no such element is found before this stream is done, and a
718 * [defaultValue] function is provided, the result of calling [defaultValue] 718 * [defaultValue] function is provided, the result of calling [defaultValue]
719 * becomes the value of the future. 719 * becomes the value of the future.
720 * 720 *
721 * If an error occurs, or if this stream ends without finding a match and 721 * If an error occurs, or if this stream ends without finding a match and
722 * with no [defaultValue] function provided, the future will receive an 722 * with no [defaultValue] function provided, the future will receive an
723 * error. 723 * error.
724 */ 724 */
725 Future<dynamic> firstWhere(bool test(T element), {Object defaultValue()}) { 725 Future<dynamic> firstWhere(bool test(T element), {Object defaultValue()}) {
726 _Future<dynamic> future = new _Future(); 726 _FutureImpl<dynamic> future = new _FutureImpl();
727 StreamSubscription subscription; 727 StreamSubscription subscription;
728 subscription = this.listen( 728 subscription = this.listen(
729 (T value) { 729 (T value) {
730 _runUserCode( 730 _runUserCode(
731 () => test(value), 731 () => test(value),
732 (bool isMatch) { 732 (bool isMatch) {
733 if (isMatch) { 733 if (isMatch) {
734 subscription.cancel(); 734 subscription.cancel();
735 future._complete(value); 735 future._setValue(value);
736 } 736 }
737 }, 737 },
738 _cancelAndError(subscription, future) 738 _cancelAndError(subscription, future)
739 ); 739 );
740 }, 740 },
741 onError: future._completeError, 741 onError: future._setError,
742 onDone: () { 742 onDone: () {
743 if (defaultValue != null) { 743 if (defaultValue != null) {
744 _runUserCode(defaultValue, future._complete, future._completeError); 744 _runUserCode(defaultValue, future._setValue, future._setError);
745 return; 745 return;
746 } 746 }
747 future._completeError(new StateError("firstMatch ended without match")); 747 future._setError(new StateError("firstMatch ended without match"));
748 }, 748 },
749 cancelOnError: true); 749 cancelOnError: true);
750 return future; 750 return future;
751 } 751 }
752 752
753 /** 753 /**
754 * Finds the last element in this stream matching [test]. 754 * Finds the last element in this stream matching [test].
755 * 755 *
756 * As [firstWhere], except that the last matching element is found. 756 * As [firstWhere], except that the last matching element is found.
757 * That means that the result cannot be provided before this stream 757 * That means that the result cannot be provided before this stream
758 * is done. 758 * is done.
759 */ 759 */
760 Future<dynamic> lastWhere(bool test(T element), {Object defaultValue()}) { 760 Future<dynamic> lastWhere(bool test(T element), {Object defaultValue()}) {
761 _Future<dynamic> future = new _Future(); 761 _FutureImpl<dynamic> future = new _FutureImpl();
762 T result = null; 762 T result = null;
763 bool foundResult = false; 763 bool foundResult = false;
764 StreamSubscription subscription; 764 StreamSubscription subscription;
765 subscription = this.listen( 765 subscription = this.listen(
766 (T value) { 766 (T value) {
767 _runUserCode( 767 _runUserCode(
768 () => true == test(value), 768 () => true == test(value),
769 (bool isMatch) { 769 (bool isMatch) {
770 if (isMatch) { 770 if (isMatch) {
771 foundResult = true; 771 foundResult = true;
772 result = value; 772 result = value;
773 } 773 }
774 }, 774 },
775 _cancelAndError(subscription, future) 775 _cancelAndError(subscription, future)
776 ); 776 );
777 }, 777 },
778 onError: future._completeError, 778 onError: future._setError,
779 onDone: () { 779 onDone: () {
780 if (foundResult) { 780 if (foundResult) {
781 future._complete(result); 781 future._setValue(result);
782 return; 782 return;
783 } 783 }
784 if (defaultValue != null) { 784 if (defaultValue != null) {
785 _runUserCode(defaultValue, future._complete, future._completeError); 785 _runUserCode(defaultValue, future._setValue, future._setError);
786 return; 786 return;
787 } 787 }
788 future._completeError(new StateError("lastMatch ended without match")); 788 future._setError(new StateError("lastMatch ended without match"));
789 }, 789 },
790 cancelOnError: true); 790 cancelOnError: true);
791 return future; 791 return future;
792 } 792 }
793 793
794 /** 794 /**
795 * Finds the single element in this stream matching [test]. 795 * Finds the single element in this stream matching [test].
796 * 796 *
797 * Like [lastMatch], except that it is an error if more than one 797 * Like [lastMatch], except that it is an error if more than one
798 * matching element occurs in the stream. 798 * matching element occurs in the stream.
799 */ 799 */
800 Future<T> singleWhere(bool test(T element)) { 800 Future<T> singleWhere(bool test(T element)) {
801 _Future<T> future = new _Future<T>(); 801 _FutureImpl<T> future = new _FutureImpl<T>();
802 T result = null; 802 T result = null;
803 bool foundResult = false; 803 bool foundResult = false;
804 StreamSubscription subscription; 804 StreamSubscription subscription;
805 subscription = this.listen( 805 subscription = this.listen(
806 (T value) { 806 (T value) {
807 _runUserCode( 807 _runUserCode(
808 () => true == test(value), 808 () => true == test(value),
809 (bool isMatch) { 809 (bool isMatch) {
810 if (isMatch) { 810 if (isMatch) {
811 if (foundResult) { 811 if (foundResult) {
812 subscription.cancel(); 812 subscription.cancel();
813 future._completeError( 813 future._setError(
814 new StateError('Multiple matches for "single"')); 814 new StateError('Multiple matches for "single"'));
815 return; 815 return;
816 } 816 }
817 foundResult = true; 817 foundResult = true;
818 result = value; 818 result = value;
819 } 819 }
820 }, 820 },
821 _cancelAndError(subscription, future) 821 _cancelAndError(subscription, future)
822 ); 822 );
823 }, 823 },
824 onError: future._completeError, 824 onError: future._setError,
825 onDone: () { 825 onDone: () {
826 if (foundResult) { 826 if (foundResult) {
827 future._complete(result); 827 future._setValue(result);
828 return; 828 return;
829 } 829 }
830 future._completeError(new StateError("single ended without match")); 830 future._setError(new StateError("single ended without match"));
831 }, 831 },
832 cancelOnError: true); 832 cancelOnError: true);
833 return future; 833 return future;
834 } 834 }
835 835
836 /** 836 /**
837 * Returns the value of the [index]th data event of this stream. 837 * Returns the value of the [index]th data event of this stream.
838 * 838 *
839 * Stops listening to the stream after a value has been found. 839 * Stops listening to the stream after a value has been found.
840 * 840 *
841 * If an error event occurs before the value is found, the future completes 841 * If an error event occurs before the value is found, the future completes
842 * with this error. 842 * with this error.
843 * 843 *
844 * If a done event occurs before the value is found, the future completes 844 * If a done event occurs before the value is found, the future completes
845 * with a [RangeError]. 845 * with a [RangeError].
846 */ 846 */
847 Future<T> elementAt(int index) { 847 Future<T> elementAt(int index) {
848 if (index is! int || index < 0) throw new ArgumentError(index); 848 if (index is! int || index < 0) throw new ArgumentError(index);
849 _Future<T> future = new _Future<T>(); 849 _FutureImpl<T> future = new _FutureImpl<T>();
850 StreamSubscription subscription; 850 StreamSubscription subscription;
851 subscription = this.listen( 851 subscription = this.listen(
852 (T value) { 852 (T value) {
853 if (index == 0) { 853 if (index == 0) {
854 subscription.cancel(); 854 subscription.cancel();
855 future._complete(value); 855 future._setValue(value);
856 return; 856 return;
857 } 857 }
858 index -= 1; 858 index -= 1;
859 }, 859 },
860 onError: future._completeError, 860 onError: future._setError,
861 onDone: () { 861 onDone: () {
862 future._completeError(new RangeError.value(index)); 862 future._setError(new RangeError.value(index));
863 }, 863 },
864 cancelOnError: true); 864 cancelOnError: true);
865 return future; 865 return future;
866 } 866 }
867 } 867 }
868 868
869 /** 869 /**
870 * A control object for the subscription on a [Stream]. 870 * A control object for the subscription on a [Stream].
871 * 871 *
872 * When you subscribe on a [Stream] using [Stream.listen], 872 * When you subscribe on a [Stream] using [Stream.listen],
(...skipping 401 matching lines...) Expand 10 before | Expand all | Expand 10 after
1274 * 1274 *
1275 * If a [moveNext] call has been made, it will complete with `false` as value, 1275 * If a [moveNext] call has been made, it will complete with `false` as value,
1276 * as will all further calls to [moveNext]. 1276 * as will all further calls to [moveNext].
1277 * 1277 *
1278 * If you need to stop listening for values before the stream iterator is 1278 * If you need to stop listening for values before the stream iterator is
1279 * automatically closed, you must call [cancel] to ensure that the stream 1279 * automatically closed, you must call [cancel] to ensure that the stream
1280 * is properly closed. 1280 * is properly closed.
1281 */ 1281 */
1282 void cancel(); 1282 void cancel();
1283 } 1283 }
OLDNEW
« no previous file with comments | « sdk/lib/async/future_impl.dart ('k') | sdk/lib/async/stream_controller.dart » ('j') | no next file with comments »

Powered by Google App Engine
This is Rietveld 408576698