| 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 class _BroadcastStream<T> extends _ControllerStream<T> { | 7 class _BroadcastStream<T> extends _ControllerStream<T> { |
| 8 _BroadcastStream(_StreamControllerLifecycle controller) : super(controller); | 8 _BroadcastStream(_StreamControllerLifecycle controller) : super(controller); |
| 9 | 9 |
| 10 bool get isBroadcast => true; | 10 bool get isBroadcast => true; |
| (...skipping 89 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... |
| 100 * The future is completed whenever the done event has been sent to all | 100 * The future is completed whenever the done event has been sent to all |
| 101 * relevant listeners. | 101 * relevant listeners. |
| 102 * The relevant listeners are the ones that were listening when [close] was | 102 * The relevant listeners are the ones that were listening when [close] was |
| 103 * called. When all of these have been canceled (sending the done event makes | 103 * called. When all of these have been canceled (sending the done event makes |
| 104 * them cancel, but they can also be canceled before sending the event), | 104 * them cancel, but they can also be canceled before sending the event), |
| 105 * this future completes. | 105 * this future completes. |
| 106 * | 106 * |
| 107 * Any attempt to listen after calling [close] will throw, so there won't | 107 * Any attempt to listen after calling [close] will throw, so there won't |
| 108 * be any further listeners. | 108 * be any further listeners. |
| 109 */ | 109 */ |
| 110 _Future _doneFuture; | 110 _FutureImpl _doneFuture; |
| 111 | 111 |
| 112 _BroadcastStreamController(this._onListen, this._onCancel) | 112 _BroadcastStreamController(this._onListen, this._onCancel) |
| 113 : _state = _STATE_INITIAL { | 113 : _state = _STATE_INITIAL { |
| 114 _next = _previous = this; | 114 _next = _previous = this; |
| 115 } | 115 } |
| 116 | 116 |
| 117 // StreamController interface. | 117 // StreamController interface. |
| 118 | 118 |
| 119 Stream<T> get stream => new _BroadcastStream<T>(this); | 119 Stream<T> get stream => new _BroadcastStream<T>(this); |
| 120 | 120 |
| (...skipping 12 matching lines...) Expand all Loading... |
| 133 /** Whether there are currently one or more subscribers. */ | 133 /** Whether there are currently one or more subscribers. */ |
| 134 bool get hasListener => !_isEmpty; | 134 bool get hasListener => !_isEmpty; |
| 135 | 135 |
| 136 /** Whether an event is being fired (sent to some, but not all, listeners). */ | 136 /** Whether an event is being fired (sent to some, but not all, listeners). */ |
| 137 bool get _isFiring => (_state & _STATE_FIRING) != 0; | 137 bool get _isFiring => (_state & _STATE_FIRING) != 0; |
| 138 | 138 |
| 139 bool get _isAddingStream => (_state & _STATE_ADDSTREAM) != 0; | 139 bool get _isAddingStream => (_state & _STATE_ADDSTREAM) != 0; |
| 140 | 140 |
| 141 bool get _mayAddEvent => (_state < _STATE_CLOSED); | 141 bool get _mayAddEvent => (_state < _STATE_CLOSED); |
| 142 | 142 |
| 143 _Future _ensureDoneFuture() { | 143 _FutureImpl _ensureDoneFuture() { |
| 144 if (_doneFuture != null) return _doneFuture; | 144 if (_doneFuture != null) return _doneFuture; |
| 145 return _doneFuture = new _Future(); | 145 return _doneFuture = new _FutureImpl(); |
| 146 } | 146 } |
| 147 | 147 |
| 148 // Linked list helpers | 148 // Linked list helpers |
| 149 | 149 |
| 150 bool get _isEmpty => identical(_next, this); | 150 bool get _isEmpty => identical(_next, this); |
| 151 | 151 |
| 152 /** Adds subscription to linked list of active listeners. */ | 152 /** Adds subscription to linked list of active listeners. */ |
| 153 void _addListener(_BroadcastSubscription<T> subscription) { | 153 void _addListener(_BroadcastSubscription<T> subscription) { |
| 154 assert(identical(subscription._next, subscription)); | 154 assert(identical(subscription._next, subscription)); |
| 155 // Insert in linked list just before `this`. | 155 // Insert in linked list just before `this`. |
| (...skipping 151 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... |
| 307 | 307 |
| 308 if (_isEmpty) { | 308 if (_isEmpty) { |
| 309 _callOnCancel(); | 309 _callOnCancel(); |
| 310 } | 310 } |
| 311 } | 311 } |
| 312 | 312 |
| 313 void _callOnCancel() { | 313 void _callOnCancel() { |
| 314 assert(_isEmpty); | 314 assert(_isEmpty); |
| 315 if (isClosed && _doneFuture._mayComplete) { | 315 if (isClosed && _doneFuture._mayComplete) { |
| 316 // When closed, _doneFuture is not null. | 316 // When closed, _doneFuture is not null. |
| 317 _doneFuture._asyncComplete(null); | 317 _doneFuture._asyncSetValue(null); |
| 318 } | 318 } |
| 319 _runGuarded(_onCancel); | 319 _runGuarded(_onCancel); |
| 320 } | 320 } |
| 321 } | 321 } |
| 322 | 322 |
| 323 class _SyncBroadcastStreamController<T> extends _BroadcastStreamController<T> { | 323 class _SyncBroadcastStreamController<T> extends _BroadcastStreamController<T> { |
| 324 _SyncBroadcastStreamController(void onListen(), void onCancel()) | 324 _SyncBroadcastStreamController(void onListen(), void onCancel()) |
| 325 : super(onListen, onCancel); | 325 : super(onListen, onCancel); |
| 326 | 326 |
| 327 // EventDispatch interface. | 327 // EventDispatch interface. |
| (...skipping 13 matching lines...) Expand all Loading... |
| 341 } | 341 } |
| 342 | 342 |
| 343 void _sendDone() { | 343 void _sendDone() { |
| 344 if (!_isEmpty) { | 344 if (!_isEmpty) { |
| 345 _forEachListener((_BroadcastSubscription<T> subscription) { | 345 _forEachListener((_BroadcastSubscription<T> subscription) { |
| 346 subscription._close(); | 346 subscription._close(); |
| 347 }); | 347 }); |
| 348 } else { | 348 } else { |
| 349 assert(_doneFuture != null); | 349 assert(_doneFuture != null); |
| 350 assert(_doneFuture._mayComplete); | 350 assert(_doneFuture._mayComplete); |
| 351 _doneFuture._asyncComplete(null); | 351 _doneFuture._asyncSetValue(null); |
| 352 } | 352 } |
| 353 } | 353 } |
| 354 } | 354 } |
| 355 | 355 |
| 356 class _AsyncBroadcastStreamController<T> extends _BroadcastStreamController<T> { | 356 class _AsyncBroadcastStreamController<T> extends _BroadcastStreamController<T> { |
| 357 _AsyncBroadcastStreamController(void onListen(), void onCancel()) | 357 _AsyncBroadcastStreamController(void onListen(), void onCancel()) |
| 358 : super(onListen, onCancel); | 358 : super(onListen, onCancel); |
| 359 | 359 |
| 360 // EventDispatch interface. | 360 // EventDispatch interface. |
| 361 | 361 |
| (...skipping 19 matching lines...) Expand all Loading... |
| 381 if (!_isEmpty) { | 381 if (!_isEmpty) { |
| 382 for (_BroadcastSubscriptionLink link = _next; | 382 for (_BroadcastSubscriptionLink link = _next; |
| 383 !identical(link, this); | 383 !identical(link, this); |
| 384 link = link._next) { | 384 link = link._next) { |
| 385 _BroadcastSubscription<T> subscription = link; | 385 _BroadcastSubscription<T> subscription = link; |
| 386 subscription._addPending(const _DelayedDone()); | 386 subscription._addPending(const _DelayedDone()); |
| 387 } | 387 } |
| 388 } else { | 388 } else { |
| 389 assert(_doneFuture != null); | 389 assert(_doneFuture != null); |
| 390 assert(_doneFuture._mayComplete); | 390 assert(_doneFuture._mayComplete); |
| 391 _doneFuture._asyncComplete(null); | 391 _doneFuture._asyncSetValue(null); |
| 392 } | 392 } |
| 393 } | 393 } |
| 394 } | 394 } |
| 395 | 395 |
| 396 /** | 396 /** |
| 397 * Stream controller that is used by [Stream.asBroadcastStream]. | 397 * Stream controller that is used by [Stream.asBroadcastStream]. |
| 398 * | 398 * |
| 399 * This stream controller allows incoming events while it is firing | 399 * This stream controller allows incoming events while it is firing |
| 400 * other events. This is handled by delaying the events until the | 400 * other events. This is handled by delaying the events until the |
| 401 * current event is done firing, and then fire the pending events. | 401 * current event is done firing, and then fire the pending events. |
| (...skipping 71 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... |
| 473 void pause([Future resumeSignal]) { | 473 void pause([Future resumeSignal]) { |
| 474 if (resumeSignal != null) resumeSignal.then(_resume); | 474 if (resumeSignal != null) resumeSignal.then(_resume); |
| 475 _pauseCount++; | 475 _pauseCount++; |
| 476 } | 476 } |
| 477 void resume() { _resume(null); } | 477 void resume() { _resume(null); } |
| 478 void _resume(_) { | 478 void _resume(_) { |
| 479 if (_pauseCount > 0) _pauseCount--; | 479 if (_pauseCount > 0) _pauseCount--; |
| 480 } | 480 } |
| 481 void cancel() {} | 481 void cancel() {} |
| 482 bool get isPaused => _pauseCount > 0; | 482 bool get isPaused => _pauseCount > 0; |
| 483 Future asFuture([Object value]) => new _Future(); | 483 Future asFuture([Object value]) => new _FutureImpl(); |
| 484 } | 484 } |
| OLD | NEW |