| 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 232 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... |
| 243 int _state = _STATE_INITIAL; | 243 int _state = _STATE_INITIAL; |
| 244 | 244 |
| 245 /** | 245 /** |
| 246 * Future completed when the stream sends its last event. | 246 * Future completed when the stream sends its last event. |
| 247 * | 247 * |
| 248 * This is also the future returned by [close]. | 248 * This is also the future returned by [close]. |
| 249 */ | 249 */ |
| 250 // TODO(lrn): Could this be stored in the varData field too, if it's not | 250 // TODO(lrn): Could this be stored in the varData field too, if it's not |
| 251 // accessed until the call to "close"? Then we need to special case if it's | 251 // accessed until the call to "close"? Then we need to special case if it's |
| 252 // accessed earlier, or if close is called before subscribing. | 252 // accessed earlier, or if close is called before subscribing. |
| 253 _FutureImpl _doneFuture; | 253 _Future _doneFuture; |
| 254 | 254 |
| 255 _StreamController(); | 255 _StreamController(); |
| 256 | 256 |
| 257 _NotificationHandler get _onListen; | 257 _NotificationHandler get _onListen; |
| 258 _NotificationHandler get _onPause; | 258 _NotificationHandler get _onPause; |
| 259 _NotificationHandler get _onResume; | 259 _NotificationHandler get _onResume; |
| 260 _NotificationHandler get _onCancel; | 260 _NotificationHandler get _onCancel; |
| 261 | 261 |
| 262 // Return a new stream every time. The streams are equal, but not identical. | 262 // Return a new stream every time. The streams are equal, but not identical. |
| 263 Stream<T> get stream => new _ControllerStream(this); | 263 Stream<T> get stream => new _ControllerStream(this); |
| (...skipping 76 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... |
| 340 if (isClosed) { | 340 if (isClosed) { |
| 341 return new StateError("Cannot add event after closing"); | 341 return new StateError("Cannot add event after closing"); |
| 342 } | 342 } |
| 343 assert(_isAddingStream); | 343 assert(_isAddingStream); |
| 344 return new StateError("Cannot add event while adding a stream"); | 344 return new StateError("Cannot add event while adding a stream"); |
| 345 } | 345 } |
| 346 | 346 |
| 347 // StreamSink interface. | 347 // StreamSink interface. |
| 348 Future addStream(Stream<T> source) { | 348 Future addStream(Stream<T> source) { |
| 349 if (!_mayAddEvent) throw _badEventState(); | 349 if (!_mayAddEvent) throw _badEventState(); |
| 350 if (_isCanceled) return new _FutureImpl.immediate(null); | 350 if (_isCanceled) return new _Future.immediate(null); |
| 351 _StreamControllerAddStreamState addState = | 351 _StreamControllerAddStreamState addState = |
| 352 new _StreamControllerAddStreamState(this, _varData, source); | 352 new _StreamControllerAddStreamState(this, _varData, source); |
| 353 _varData = addState; | 353 _varData = addState; |
| 354 _state |= _STATE_ADDSTREAM; | 354 _state |= _STATE_ADDSTREAM; |
| 355 return addState.addStreamFuture; | 355 return addState.addStreamFuture; |
| 356 } | 356 } |
| 357 | 357 |
| 358 Future get done => _ensureDoneFuture(); | 358 Future get done => _ensureDoneFuture(); |
| 359 | 359 |
| 360 Future _ensureDoneFuture() { | 360 Future _ensureDoneFuture() { |
| 361 if (_doneFuture == null) { | 361 if (_doneFuture == null) { |
| 362 _doneFuture = new _FutureImpl(); | 362 _doneFuture = new _Future(); |
| 363 if (_isCanceled) _doneFuture._setValue(null); | 363 if (_isCanceled) _doneFuture._complete(null); |
| 364 } | 364 } |
| 365 return _doneFuture; | 365 return _doneFuture; |
| 366 } | 366 } |
| 367 | 367 |
| 368 /** | 368 /** |
| 369 * Send or enqueue a data event. | 369 * Send or enqueue a data event. |
| 370 */ | 370 */ |
| 371 void add(T value) { | 371 void add(T value) { |
| 372 if (!_mayAddEvent) throw _badEventState(); | 372 if (!_mayAddEvent) throw _badEventState(); |
| 373 _add(value); | 373 _add(value); |
| (...skipping 98 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... |
| 472 void _recordCancel(StreamSubscription<T> subscription) { | 472 void _recordCancel(StreamSubscription<T> subscription) { |
| 473 if (_isAddingStream) { | 473 if (_isAddingStream) { |
| 474 _StreamControllerAddStreamState addState = _varData; | 474 _StreamControllerAddStreamState addState = _varData; |
| 475 addState.cancel(); | 475 addState.cancel(); |
| 476 } | 476 } |
| 477 _varData = null; | 477 _varData = null; |
| 478 _state = | 478 _state = |
| 479 (_state & ~(_STATE_SUBSCRIBED | _STATE_ADDSTREAM)) | _STATE_CANCELED; | 479 (_state & ~(_STATE_SUBSCRIBED | _STATE_ADDSTREAM)) | _STATE_CANCELED; |
| 480 _runGuarded(_onCancel); | 480 _runGuarded(_onCancel); |
| 481 if (_doneFuture != null && _doneFuture._mayComplete) { | 481 if (_doneFuture != null && _doneFuture._mayComplete) { |
| 482 _doneFuture._asyncSetValue(null); | 482 _doneFuture._asyncComplete(null); |
| 483 } | 483 } |
| 484 } | 484 } |
| 485 | 485 |
| 486 void _recordPause(StreamSubscription<T> subscription) { | 486 void _recordPause(StreamSubscription<T> subscription) { |
| 487 if (_isAddingStream) { | 487 if (_isAddingStream) { |
| 488 _StreamControllerAddStreamState addState = _varData; | 488 _StreamControllerAddStreamState addState = _varData; |
| 489 addState.pause(); | 489 addState.pause(); |
| 490 } | 490 } |
| 491 _runGuarded(_onPause); | 491 _runGuarded(_onPause); |
| 492 } | 492 } |
| (...skipping 149 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... |
| 642 Future close() => _target.close(); | 642 Future close() => _target.close(); |
| 643 Future addStream(Stream<T> source) => _target.addStream(source); | 643 Future addStream(Stream<T> source) => _target.addStream(source); |
| 644 Future get done => _target.done; | 644 Future get done => _target.done; |
| 645 } | 645 } |
| 646 | 646 |
| 647 /** | 647 /** |
| 648 * Object containing the state used to handle [StreamController.addStream]. | 648 * Object containing the state used to handle [StreamController.addStream]. |
| 649 */ | 649 */ |
| 650 class _AddStreamState<T> { | 650 class _AddStreamState<T> { |
| 651 // [_FutureImpl] returned by call to addStream. | 651 // [_FutureImpl] returned by call to addStream. |
| 652 _FutureImpl addStreamFuture; | 652 _Future addStreamFuture; |
| 653 | 653 |
| 654 // Subscription on stream argument to addStream. | 654 // Subscription on stream argument to addStream. |
| 655 StreamSubscription addSubscription; | 655 StreamSubscription addSubscription; |
| 656 | 656 |
| 657 _AddStreamState(_EventSink<T> controller, Stream source) | 657 _AddStreamState(_EventSink<T> controller, Stream source) |
| 658 : addStreamFuture = new _FutureImpl(), | 658 : addStreamFuture = new _Future(), |
| 659 addSubscription = source.listen(controller._add, | 659 addSubscription = source.listen(controller._add, |
| 660 onError: controller._addError, | 660 onError: controller._addError, |
| 661 onDone: controller._close, | 661 onDone: controller._close, |
| 662 cancelOnError: true); | 662 cancelOnError: true); |
| 663 | 663 |
| 664 void pause() { | 664 void pause() { |
| 665 addSubscription.pause(); | 665 addSubscription.pause(); |
| 666 } | 666 } |
| 667 | 667 |
| 668 void resume() { | 668 void resume() { |
| 669 addSubscription.resume(); | 669 addSubscription.resume(); |
| 670 } | 670 } |
| 671 | 671 |
| 672 void cancel() { | 672 void cancel() { |
| 673 addSubscription.cancel(); | 673 addSubscription.cancel(); |
| 674 complete(); | 674 complete(); |
| 675 } | 675 } |
| 676 | 676 |
| 677 void complete() { | 677 void complete() { |
| 678 addStreamFuture._asyncSetValue(null); | 678 addStreamFuture._asyncComplete(null); |
| 679 } | 679 } |
| 680 } | 680 } |
| 681 | 681 |
| 682 class _StreamControllerAddStreamState<T> extends _AddStreamState<T> { | 682 class _StreamControllerAddStreamState<T> extends _AddStreamState<T> { |
| 683 // The subscription or pending data of a _StreamController. | 683 // The subscription or pending data of a _StreamController. |
| 684 // Stored here because we reuse the `_varData` field in the _StreamController | 684 // Stored here because we reuse the `_varData` field in the _StreamController |
| 685 // to store this state object. | 685 // to store this state object. |
| 686 var varData; | 686 var varData; |
| 687 | 687 |
| 688 _StreamControllerAddStreamState(_StreamController controller, | 688 _StreamControllerAddStreamState(_StreamController controller, |
| 689 this.varData, | 689 this.varData, |
| 690 Stream source) : super(controller, source) { | 690 Stream source) : super(controller, source) { |
| 691 if (controller.isPaused) { | 691 if (controller.isPaused) { |
| 692 addSubscription.pause(); | 692 addSubscription.pause(); |
| 693 } | 693 } |
| 694 } | 694 } |
| 695 } | 695 } |
| OLD | NEW |