Chromium Code Reviews| OLD | NEW |
|---|---|
| 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 Loading... | |
| 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 Loading... | |
| 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 Loading... | |
| 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 } |
| OLD | NEW |