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

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

Issue 269283007: Make errors from StreamController onCancel calls end up in the returned future. (Closed) Base URL: https://dart.googlecode.com/svn/branches/bleeding_edge/dart
Patch Set: Add tests. Propagate errors during cancel only to cancel future. Created 6 years, 7 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) 2012, the Dart project authors. Please see the AUTHORS file 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 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 // Controller for creating and adding events to a stream. 8 // Controller for creating and adding events to a stream.
9 // ------------------------------------------------------------------- 9 // -------------------------------------------------------------------
10 10
(...skipping 402 matching lines...) Expand 10 before | Expand all | Expand 10 after
413 * After closing, no further events may be added using [add] or [addError]. 413 * After closing, no further events may be added using [add] or [addError].
414 * 414 *
415 * You are allowed to close the controller more than once, but only the first 415 * You are allowed to close the controller more than once, but only the first
416 * call has any effect. 416 * call has any effect.
417 * 417 *
418 * The first time a controller is closed, a "done" event is sent to its 418 * The first time a controller is closed, a "done" event is sent to its
419 * stream. 419 * stream.
420 */ 420 */
421 Future close() { 421 Future close() {
422 if (isClosed) { 422 if (isClosed) {
423 _ensureDoneFuture(); 423 return _ensureDoneFuture();
424 return _doneFuture;
425 } 424 }
426 if (!_mayAddEvent) throw _badEventState(); 425 if (!_mayAddEvent) throw _badEventState();
427 _state |= _STATE_CLOSED; 426 _state |= _STATE_CLOSED;
428 if (hasListener) { 427 if (hasListener) {
429 _sendDone(); 428 _sendDone();
430 } else if (_isInitialState) { 429 } else if (_isInitialState) {
431 _ensurePendingEvents().add(const _DelayedDone()); 430 _ensurePendingEvents().add(const _DelayedDone());
432 } 431 }
433 _ensureDoneFuture(); 432 return _ensureDoneFuture();
434 return _doneFuture;
435 } 433 }
436 434
437 // EventSink interface. Used by the [addStream] events. 435 // EventSink interface. Used by the [addStream] events.
438 436
439 // Add data event, used both by the [addStream] events and by [add]. 437 // Add data event, used both by the [addStream] events and by [add].
440 void _add(T value) { 438 void _add(T value) {
441 if (hasListener) { 439 if (hasListener) {
442 _sendData(value); 440 _sendData(value);
443 } else if (_isInitialState) { 441 } else if (_isInitialState) {
444 _ensurePendingEvents().add(new _DelayedData<T>(value)); 442 _ensurePendingEvents().add(new _DelayedData<T>(value));
(...skipping 37 matching lines...) Expand 10 before | Expand all | Expand 10 after
482 } 480 }
483 subscription._setPendingEvents(pendingEvents); 481 subscription._setPendingEvents(pendingEvents);
484 subscription._guardCallback(() { 482 subscription._guardCallback(() {
485 _runGuarded(_onListen); 483 _runGuarded(_onListen);
486 }); 484 });
487 485
488 return subscription; 486 return subscription;
489 } 487 }
490 488
491 Future _recordCancel(StreamSubscription<T> subscription) { 489 Future _recordCancel(StreamSubscription<T> subscription) {
490 Future result;
492 if (_isAddingStream) { 491 if (_isAddingStream) {
493 _StreamControllerAddStreamState addState = _varData; 492 _StreamControllerAddStreamState addState = _varData;
494 addState.cancel(); 493 result = addState.cancel();
495 } 494 }
496 _varData = null; 495 _varData = null;
497 _state = 496 _state =
498 (_state & ~(_STATE_SUBSCRIBED | _STATE_ADDSTREAM)) | _STATE_CANCELED; 497 (_state & ~(_STATE_SUBSCRIBED | _STATE_ADDSTREAM)) | _STATE_CANCELED;
499 498
499 if (_onCancel != null) {
floitsch 2014/05/09 15:39:21 Give general overview of how we want to handle err
500 if (result == null) {
501 try {
502 result = _onCancel();
503 } catch (e, s) {
504 result = new _Future().._asyncCompleteError(e, s);
floitsch 2014/05/09 15:39:21 add comment why this is necessary.
505 }
506 } else {
507 result = result.whenComplete(_onCancel);
508 }
509 }
510
500 void complete() { 511 void complete() {
501 if (_doneFuture != null && _doneFuture._mayComplete) { 512 if (_doneFuture != null && _doneFuture._mayComplete) {
502 _doneFuture._asyncComplete(null); 513 _doneFuture._asyncComplete(null);
503 } 514 }
504 } 515 }
505 516
506 Future future = _runGuarded(_onCancel); 517 if (result != null) {
507 if (future != null) { 518 result = result.whenComplete(complete);
508 future = future.whenComplete(complete);
509 } else { 519 } else {
510 complete(); 520 complete();
511 } 521 }
512 return future; 522
523 return result;
513 } 524 }
514 525
515 void _recordPause(StreamSubscription<T> subscription) { 526 void _recordPause(StreamSubscription<T> subscription) {
516 if (_isAddingStream) { 527 if (_isAddingStream) {
517 _StreamControllerAddStreamState addState = _varData; 528 _StreamControllerAddStreamState addState = _varData;
518 addState.pause(); 529 addState.pause();
519 } 530 }
520 _runGuarded(_onPause); 531 _runGuarded(_onPause);
521 } 532 }
522 533
(...skipping 173 matching lines...) Expand 10 before | Expand all | Expand 10 after
696 }; 707 };
697 708
698 void pause() { 709 void pause() {
699 addSubscription.pause(); 710 addSubscription.pause();
700 } 711 }
701 712
702 void resume() { 713 void resume() {
703 addSubscription.resume(); 714 addSubscription.resume();
704 } 715 }
705 716
706 void cancel() { 717 /**
707 addSubscription.cancel(); 718 * Stop adding the stream.
708 complete(); 719 *
720 * Complete the future returned by `StreamController.addStream` when
721 * the cancel is complete.
722 *
723 * Return a future if the cancel takes time, otherwise return `null`.
724 */
725 Future cancel() {
726 var cancel = addSubscription.cancel();
727 if (cancel == null) {
728 addStreamFuture._asyncComplete(null);
729 return null;
730 }
731 return cancel.whenComplete(() { addStreamFuture._asyncComplete(null); });
709 } 732 }
710 733
711 void complete() { 734 void complete() {
712 addStreamFuture._asyncComplete(null); 735 addStreamFuture._asyncComplete(null);
713 } 736 }
714 } 737 }
715 738
716 class _StreamControllerAddStreamState<T> extends _AddStreamState<T> { 739 class _StreamControllerAddStreamState<T> extends _AddStreamState<T> {
717 // The subscription or pending data of a _StreamController. 740 // The subscription or pending data of a _StreamController.
718 // Stored here because we reuse the `_varData` field in the _StreamController 741 // Stored here because we reuse the `_varData` field in the _StreamController
719 // to store this state object. 742 // to store this state object.
720 var varData; 743 var varData;
721 744
722 _StreamControllerAddStreamState(_StreamController controller, 745 _StreamControllerAddStreamState(_StreamController controller,
723 this.varData, 746 this.varData,
724 Stream source, 747 Stream source,
725 bool cancelOnError) 748 bool cancelOnError)
726 : super(controller, source, cancelOnError) { 749 : super(controller, source, cancelOnError) {
727 if (controller.isPaused) { 750 if (controller.isPaused) {
728 addSubscription.pause(); 751 addSubscription.pause();
729 } 752 }
730 } 753 }
731 } 754 }
OLDNEW

Powered by Google App Engine
This is Rietveld 408576698