| OLD | NEW |
|---|
| (Empty) |
| 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 | |
| 3 // BSD-style license that can be found in the LICENSE file. | |
| 4 | |
| 5 part of dart.async; | |
| 6 | |
| 7 /** Abstract and private interface for a place to put events. */ | |
| 8 abstract class _EventSink<T> { | |
| 9 void _add(T data); | |
| 10 void _addError(Object error, StackTrace stackTrace); | |
| 11 void _close(); | |
| 12 } | |
| 13 | |
| 14 /** | |
| 15 * Abstract and private interface for a place to send events. | |
| 16 * | |
| 17 * Used by event buffering to finally dispatch the pending event, where | |
| 18 * [_EventSink] is where the event first enters the stream subscription, | |
| 19 * and may yet be buffered. | |
| 20 */ | |
| 21 abstract class _EventDispatch<T> { | |
| 22 void _sendData(T data); | |
| 23 void _sendError(Object error, StackTrace stackTrace); | |
| 24 void _sendDone(); | |
| 25 } | |
| 26 | |
| 27 /** | |
| 28 * Default implementation of stream subscription of buffering events. | |
| 29 * | |
| 30 * The only public methods are those of [StreamSubscription], so instances of | |
| 31 * [_BufferingStreamSubscription] can be returned directly as a | |
| 32 * [StreamSubscription] without exposing internal functionality. | |
| 33 * | |
| 34 * The [StreamController] is a public facing version of [Stream] and this class, | |
| 35 * with some methods made public. | |
| 36 * | |
| 37 * The user interface of [_BufferingStreamSubscription] are the following | |
| 38 * methods: | |
| 39 * | |
| 40 * * [_add]: Add a data event to the stream. | |
| 41 * * [_addError]: Add an error event to the stream. | |
| 42 * * [_close]: Request to close the stream. | |
| 43 * * [_onCancel]: Called when the subscription will provide no more events, | |
| 44 * either due to being actively canceled, or after sending a done event. | |
| 45 * * [_onPause]: Called when the subscription wants the event source to pause. | |
| 46 * * [_onResume]: Called when allowing new events after a pause. | |
| 47 * | |
| 48 * The user should not add new events when the subscription requests a paused, | |
| 49 * but if it happens anyway, the subscription will enqueue the events just as | |
| 50 * when new events arrive while still firing an old event. | |
| 51 */ | |
| 52 class _BufferingStreamSubscription<T> implements StreamSubscription<T>, | |
| 53 _EventSink<T>, | |
| 54 _EventDispatch<T> { | |
| 55 /** The `cancelOnError` flag from the `listen` call. */ | |
| 56 static const int _STATE_CANCEL_ON_ERROR = 1; | |
| 57 /** | |
| 58 * Whether the "done" event has been received. | |
| 59 * No further events are accepted after this. | |
| 60 */ | |
| 61 static const int _STATE_CLOSED = 2; | |
| 62 /** | |
| 63 * Set if the input has been asked not to send events. | |
| 64 * | |
| 65 * This is not the same as being paused, since the input will remain paused | |
| 66 * after a call to [resume] if there are pending events. | |
| 67 */ | |
| 68 static const int _STATE_INPUT_PAUSED = 4; | |
| 69 /** | |
| 70 * Whether the subscription has been canceled. | |
| 71 * | |
| 72 * Set by calling [cancel], or by handling a "done" event, or an "error" event | |
| 73 * when `cancelOnError` is true. | |
| 74 */ | |
| 75 static const int _STATE_CANCELED = 8; | |
| 76 /** | |
| 77 * Set when either: | |
| 78 * | |
| 79 * * an error is sent, and [cancelOnError] is true, or | |
| 80 * * a done event is sent. | |
| 81 * | |
| 82 * If the subscription is canceled while _STATE_WAIT_FOR_CANCEL is set, the | |
| 83 * state is unset, and no furher events must be delivered. | |
| 84 */ | |
| 85 static const int _STATE_WAIT_FOR_CANCEL = 16; | |
| 86 static const int _STATE_IN_CALLBACK = 32; | |
| 87 static const int _STATE_HAS_PENDING = 64; | |
| 88 static const int _STATE_PAUSE_COUNT = 128; | |
| 89 static const int _STATE_PAUSE_COUNT_SHIFT = 7; | |
| 90 | |
| 91 /* Event handlers provided in constructor. */ | |
| 92 _DataHandler<T> _onData; | |
| 93 Function _onError; | |
| 94 _DoneHandler _onDone; | |
| 95 final Zone _zone = Zone.current; | |
| 96 | |
| 97 /** Bit vector based on state-constants above. */ | |
| 98 int _state; | |
| 99 | |
| 100 // TODO(floitsch): reuse another field | |
| 101 /** The future [_onCancel] may return. */ | |
| 102 Future _cancelFuture; | |
| 103 | |
| 104 /** | |
| 105 * Queue of pending events. | |
| 106 * | |
| 107 * Is created when necessary, or set in constructor for preconfigured events. | |
| 108 */ | |
| 109 _PendingEvents _pending; | |
| 110 | |
| 111 _BufferingStreamSubscription(void onData(T data), | |
| 112 Function onError, | |
| 113 void onDone(), | |
| 114 bool cancelOnError) | |
| 115 : _state = (cancelOnError ? _STATE_CANCEL_ON_ERROR : 0) { | |
| 116 this.onData(onData); | |
| 117 this.onError(onError); | |
| 118 this.onDone(onDone); | |
| 119 } | |
| 120 | |
| 121 /** | |
| 122 * Sets the subscription's pending events object. | |
| 123 * | |
| 124 * This can only be done once. The pending events object is used for the | |
| 125 * rest of the subscription's life cycle. | |
| 126 */ | |
| 127 void _setPendingEvents(_PendingEvents pendingEvents) { | |
| 128 assert(_pending == null); | |
| 129 if (pendingEvents == null) return; | |
| 130 _pending = pendingEvents; | |
| 131 if (!pendingEvents.isEmpty) { | |
| 132 _state |= _STATE_HAS_PENDING; | |
| 133 _pending.schedule(this); | |
| 134 } | |
| 135 } | |
| 136 | |
| 137 /** | |
| 138 * Extracts the pending events from a canceled stream. | |
| 139 * | |
| 140 * This can only be done during the [_onCancel] method call. After that, | |
| 141 * any remaining pending events will be cleared. | |
| 142 */ | |
| 143 _PendingEvents _extractPending() { | |
| 144 assert(_isCanceled); | |
| 145 _PendingEvents events = _pending; | |
| 146 _pending = null; | |
| 147 return events; | |
| 148 } | |
| 149 | |
| 150 // StreamSubscription interface. | |
| 151 | |
| 152 void onData(void handleData(T event)) { | |
| 153 if (handleData == null) handleData = _nullDataHandler; | |
| 154 _onData = _zone.registerUnaryCallback(handleData); | |
| 155 } | |
| 156 | |
| 157 void onError(Function handleError) { | |
| 158 if (handleError == null) handleError = _nullErrorHandler; | |
| 159 _onError = _registerErrorHandler(handleError, _zone); | |
| 160 } | |
| 161 | |
| 162 void onDone(void handleDone()) { | |
| 163 if (handleDone == null) handleDone = _nullDoneHandler; | |
| 164 _onDone = _zone.registerCallback(handleDone); | |
| 165 } | |
| 166 | |
| 167 void pause([Future resumeSignal]) { | |
| 168 if (_isCanceled) return; | |
| 169 bool wasPaused = _isPaused; | |
| 170 bool wasInputPaused = _isInputPaused; | |
| 171 // Increment pause count and mark input paused (if it isn't already). | |
| 172 _state = (_state + _STATE_PAUSE_COUNT) | _STATE_INPUT_PAUSED; | |
| 173 if (resumeSignal != null) resumeSignal.whenComplete(resume); | |
| 174 if (!wasPaused && _pending != null) _pending.cancelSchedule(); | |
| 175 if (!wasInputPaused && !_inCallback) _guardCallback(_onPause); | |
| 176 } | |
| 177 | |
| 178 void resume() { | |
| 179 if (_isCanceled) return; | |
| 180 if (_isPaused) { | |
| 181 _decrementPauseCount(); | |
| 182 if (!_isPaused) { | |
| 183 if (_hasPending && !_pending.isEmpty) { | |
| 184 // Input is still paused. | |
| 185 _pending.schedule(this); | |
| 186 } else { | |
| 187 assert(_mayResumeInput); | |
| 188 _state &= ~_STATE_INPUT_PAUSED; | |
| 189 if (!_inCallback) _guardCallback(_onResume); | |
| 190 } | |
| 191 } | |
| 192 } | |
| 193 } | |
| 194 | |
| 195 Future cancel() { | |
| 196 // The user doesn't want to receive any further events. If there is an | |
| 197 // error or done event pending (waiting for the cancel to be done) discard | |
| 198 // that event. | |
| 199 _state &= ~_STATE_WAIT_FOR_CANCEL; | |
| 200 if (_isCanceled) return _cancelFuture; | |
| 201 _cancel(); | |
| 202 return _cancelFuture; | |
| 203 } | |
| 204 | |
| 205 Future asFuture([var futureValue]) { | |
| 206 _Future<T> result = new _Future<T>(); | |
| 207 | |
| 208 // Overwrite the onDone and onError handlers. | |
| 209 _onDone = () { result._complete(futureValue); }; | |
| 210 _onError = (error, stackTrace) { | |
| 211 cancel(); | |
| 212 result._completeError(error, stackTrace); | |
| 213 }; | |
| 214 | |
| 215 return result; | |
| 216 } | |
| 217 | |
| 218 // State management. | |
| 219 | |
| 220 bool get _isInputPaused => (_state & _STATE_INPUT_PAUSED) != 0; | |
| 221 bool get _isClosed => (_state & _STATE_CLOSED) != 0; | |
| 222 bool get _isCanceled => (_state & _STATE_CANCELED) != 0; | |
| 223 bool get _waitsForCancel => (_state & _STATE_WAIT_FOR_CANCEL) != 0; | |
| 224 bool get _inCallback => (_state & _STATE_IN_CALLBACK) != 0; | |
| 225 bool get _hasPending => (_state & _STATE_HAS_PENDING) != 0; | |
| 226 bool get _isPaused => _state >= _STATE_PAUSE_COUNT; | |
| 227 bool get _canFire => _state < _STATE_IN_CALLBACK; | |
| 228 bool get _mayResumeInput => | |
| 229 !_isPaused && (_pending == null || _pending.isEmpty); | |
| 230 bool get _cancelOnError => (_state & _STATE_CANCEL_ON_ERROR) != 0; | |
| 231 | |
| 232 bool get isPaused => _isPaused; | |
| 233 | |
| 234 void _cancel() { | |
| 235 _state |= _STATE_CANCELED; | |
| 236 if (_hasPending) { | |
| 237 _pending.cancelSchedule(); | |
| 238 } | |
| 239 if (!_inCallback) _pending = null; | |
| 240 _cancelFuture = _onCancel(); | |
| 241 } | |
| 242 | |
| 243 /** | |
| 244 * Increment the pause count. | |
| 245 * | |
| 246 * Also marks input as paused. | |
| 247 */ | |
| 248 void _incrementPauseCount() { | |
| 249 _state = (_state + _STATE_PAUSE_COUNT) | _STATE_INPUT_PAUSED; | |
| 250 } | |
| 251 | |
| 252 /** | |
| 253 * Decrements the pause count. | |
| 254 * | |
| 255 * Does not automatically unpause the input (call [_onResume]) when | |
| 256 * the pause count reaches zero. This is handled elsewhere, and only | |
| 257 * if there are no pending events buffered. | |
| 258 */ | |
| 259 void _decrementPauseCount() { | |
| 260 assert(_isPaused); | |
| 261 _state -= _STATE_PAUSE_COUNT; | |
| 262 } | |
| 263 | |
| 264 // _EventSink interface. | |
| 265 | |
| 266 void _add(T data) { | |
| 267 assert(!_isClosed); | |
| 268 if (_isCanceled) return; | |
| 269 if (_canFire) { | |
| 270 _sendData(data); | |
| 271 } else { | |
| 272 _addPending(new _DelayedData(data)); | |
| 273 } | |
| 274 } | |
| 275 | |
| 276 void _addError(Object error, StackTrace stackTrace) { | |
| 277 if (_isCanceled) return; | |
| 278 if (_canFire) { | |
| 279 _sendError(error, stackTrace); // Reports cancel after sending. | |
| 280 } else { | |
| 281 _addPending(new _DelayedError(error, stackTrace)); | |
| 282 } | |
| 283 } | |
| 284 | |
| 285 void _close() { | |
| 286 assert(!_isClosed); | |
| 287 if (_isCanceled) return; | |
| 288 _state |= _STATE_CLOSED; | |
| 289 if (_canFire) { | |
| 290 _sendDone(); | |
| 291 } else { | |
| 292 _addPending(const _DelayedDone()); | |
| 293 } | |
| 294 } | |
| 295 | |
| 296 // Hooks called when the input is paused, unpaused or canceled. | |
| 297 // These must not throw. If overwritten to call user code, include suitable | |
| 298 // try/catch wrapping and send any errors to | |
| 299 // [_Zone.current.handleUncaughtError]. | |
| 300 void _onPause() { | |
| 301 assert(_isInputPaused); | |
| 302 } | |
| 303 | |
| 304 void _onResume() { | |
| 305 assert(!_isInputPaused); | |
| 306 } | |
| 307 | |
| 308 Future _onCancel() { | |
| 309 assert(_isCanceled); | |
| 310 return null; | |
| 311 } | |
| 312 | |
| 313 // Handle pending events. | |
| 314 | |
| 315 /** | |
| 316 * Add a pending event. | |
| 317 * | |
| 318 * If the subscription is not paused, this also schedules a firing | |
| 319 * of pending events later (if necessary). | |
| 320 */ | |
| 321 void _addPending(_DelayedEvent event) { | |
| 322 _StreamImplEvents pending = _pending; | |
| 323 if (_pending == null) pending = _pending = new _StreamImplEvents(); | |
| 324 pending.add(event); | |
| 325 if (!_hasPending) { | |
| 326 _state |= _STATE_HAS_PENDING; | |
| 327 if (!_isPaused) { | |
| 328 _pending.schedule(this); | |
| 329 } | |
| 330 } | |
| 331 } | |
| 332 | |
| 333 /* _EventDispatch interface. */ | |
| 334 | |
| 335 void _sendData(T data) { | |
| 336 assert(!_isCanceled); | |
| 337 assert(!_isPaused); | |
| 338 assert(!_inCallback); | |
| 339 bool wasInputPaused = _isInputPaused; | |
| 340 _state |= _STATE_IN_CALLBACK; | |
| 341 _zone.runUnaryGuarded(_onData, data); | |
| 342 _state &= ~_STATE_IN_CALLBACK; | |
| 343 _checkState(wasInputPaused); | |
| 344 } | |
| 345 | |
| 346 void _sendError(Object error, StackTrace stackTrace) { | |
| 347 assert(!_isCanceled); | |
| 348 assert(!_isPaused); | |
| 349 assert(!_inCallback); | |
| 350 bool wasInputPaused = _isInputPaused; | |
| 351 | |
| 352 void sendError() { | |
| 353 // If the subscription has been canceled while waiting for the cancel | |
| 354 // future to finish we must not report the error. | |
| 355 if (_isCanceled && !_waitsForCancel) return; | |
| 356 _state |= _STATE_IN_CALLBACK; | |
| 357 if (_onError is ZoneBinaryCallback) { | |
| 358 _zone.runBinaryGuarded(_onError, error, stackTrace); | |
| 359 } else { | |
| 360 _zone.runUnaryGuarded(_onError, error); | |
| 361 } | |
| 362 _state &= ~_STATE_IN_CALLBACK; | |
| 363 } | |
| 364 | |
| 365 if (_cancelOnError) { | |
| 366 _state |= _STATE_WAIT_FOR_CANCEL; | |
| 367 _cancel(); | |
| 368 if (_cancelFuture is Future) { | |
| 369 _cancelFuture.whenComplete(sendError); | |
| 370 } else { | |
| 371 sendError(); | |
| 372 } | |
| 373 } else { | |
| 374 sendError(); | |
| 375 // Only check state if not cancelOnError. | |
| 376 _checkState(wasInputPaused); | |
| 377 } | |
| 378 } | |
| 379 | |
| 380 void _sendDone() { | |
| 381 assert(!_isCanceled); | |
| 382 assert(!_isPaused); | |
| 383 assert(!_inCallback); | |
| 384 | |
| 385 void sendDone() { | |
| 386 // If the subscription has been canceled while waiting for the cancel | |
| 387 // future to finish we must not report the done event. | |
| 388 if (!_waitsForCancel) return; | |
| 389 _state |= (_STATE_CANCELED | _STATE_CLOSED | _STATE_IN_CALLBACK); | |
| 390 _zone.runGuarded(_onDone); | |
| 391 _state &= ~_STATE_IN_CALLBACK; | |
| 392 } | |
| 393 | |
| 394 _cancel(); | |
| 395 _state |= _STATE_WAIT_FOR_CANCEL; | |
| 396 if (_cancelFuture is Future) { | |
| 397 _cancelFuture.whenComplete(sendDone); | |
| 398 } else { | |
| 399 sendDone(); | |
| 400 } | |
| 401 } | |
| 402 | |
| 403 /** | |
| 404 * Call a hook function. | |
| 405 * | |
| 406 * The call is properly wrapped in code to avoid other callbacks | |
| 407 * during the call, and it checks for state changes after the call | |
| 408 * that should cause further callbacks. | |
| 409 */ | |
| 410 void _guardCallback(callback) { | |
| 411 assert(!_inCallback); | |
| 412 bool wasInputPaused = _isInputPaused; | |
| 413 _state |= _STATE_IN_CALLBACK; | |
| 414 callback(); | |
| 415 _state &= ~_STATE_IN_CALLBACK; | |
| 416 _checkState(wasInputPaused); | |
| 417 } | |
| 418 | |
| 419 /** | |
| 420 * Check if the input needs to be informed of state changes. | |
| 421 * | |
| 422 * State changes are pausing, resuming and canceling. | |
| 423 * | |
| 424 * After canceling, no further callbacks will happen. | |
| 425 * | |
| 426 * The cancel callback is called after a user cancel, or after | |
| 427 * the final done event is sent. | |
| 428 */ | |
| 429 void _checkState(bool wasInputPaused) { | |
| 430 assert(!_inCallback); | |
| 431 if (_hasPending && _pending.isEmpty) { | |
| 432 _state &= ~_STATE_HAS_PENDING; | |
| 433 if (_isInputPaused && _mayResumeInput) { | |
| 434 _state &= ~_STATE_INPUT_PAUSED; | |
| 435 } | |
| 436 } | |
| 437 // If the state changes during a callback, we immediately | |
| 438 // make a new state-change callback. Loop until the state didn't change. | |
| 439 while (true) { | |
| 440 if (_isCanceled) { | |
| 441 _pending = null; | |
| 442 return; | |
| 443 } | |
| 444 bool isInputPaused = _isInputPaused; | |
| 445 if (wasInputPaused == isInputPaused) break; | |
| 446 _state ^= _STATE_IN_CALLBACK; | |
| 447 if (isInputPaused) { | |
| 448 _onPause(); | |
| 449 } else { | |
| 450 _onResume(); | |
| 451 } | |
| 452 _state &= ~_STATE_IN_CALLBACK; | |
| 453 wasInputPaused = isInputPaused; | |
| 454 } | |
| 455 if (_hasPending && !_isPaused) { | |
| 456 _pending.schedule(this); | |
| 457 } | |
| 458 } | |
| 459 } | |
| 460 | |
| 461 // ------------------------------------------------------------------- | |
| 462 // Common base class for single and multi-subscription streams. | |
| 463 // ------------------------------------------------------------------- | |
| 464 abstract class _StreamImpl<T> extends Stream<T> { | |
| 465 // ------------------------------------------------------------------ | |
| 466 // Stream interface. | |
| 467 | |
| 468 StreamSubscription<T> listen(void onData(T data), | |
| 469 { Function onError, | |
| 470 void onDone(), | |
| 471 bool cancelOnError }) { | |
| 472 cancelOnError = identical(true, cancelOnError); | |
| 473 StreamSubscription subscription = | |
| 474 _createSubscription(onData, onError, onDone, cancelOnError); | |
| 475 _onListen(subscription); | |
| 476 return subscription; | |
| 477 } | |
| 478 | |
| 479 // ------------------------------------------------------------------- | |
| 480 /** Create a subscription object. Called by [subcribe]. */ | |
| 481 StreamSubscription<T> _createSubscription( | |
| 482 void onData(T data), | |
| 483 Function onError, | |
| 484 void onDone(), | |
| 485 bool cancelOnError) { | |
| 486 return new _BufferingStreamSubscription<T>(onData, onError, onDone, | |
| 487 cancelOnError); | |
| 488 } | |
| 489 | |
| 490 /** Hook called when the subscription has been created. */ | |
| 491 void _onListen(StreamSubscription subscription) {} | |
| 492 } | |
| 493 | |
| 494 typedef _PendingEvents _EventGenerator(); | |
| 495 | |
| 496 /** Stream that generates its own events. */ | |
| 497 class _GeneratedStreamImpl<T> extends _StreamImpl<T> { | |
| 498 final _EventGenerator _pending; | |
| 499 bool _isUsed = false; | |
| 500 /** | |
| 501 * Initializes the stream to have only the events provided by a | |
| 502 * [_PendingEvents]. | |
| 503 * | |
| 504 * A new [_PendingEvents] must be generated for each listen. | |
| 505 */ | |
| 506 _GeneratedStreamImpl(this._pending); | |
| 507 | |
| 508 StreamSubscription<T> _createSubscription( | |
| 509 void onData(T data), | |
| 510 Function onError, | |
| 511 void onDone(), | |
| 512 bool cancelOnError) { | |
| 513 if (_isUsed) throw new StateError("Stream has already been listened to."); | |
| 514 _isUsed = true; | |
| 515 return new _BufferingStreamSubscription( | |
| 516 onData, onError, onDone, cancelOnError).._setPendingEvents(_pending()); | |
| 517 } | |
| 518 } | |
| 519 | |
| 520 | |
| 521 /** Pending events object that gets its events from an [Iterable]. */ | |
| 522 class _IterablePendingEvents<T> extends _PendingEvents { | |
| 523 // The iterator providing data for data events. | |
| 524 // Set to null when iteration has completed. | |
| 525 Iterator<T> _iterator; | |
| 526 | |
| 527 _IterablePendingEvents(Iterable<T> data) : _iterator = data.iterator; | |
| 528 | |
| 529 bool get isEmpty => _iterator == null; | |
| 530 | |
| 531 void handleNext(_EventDispatch dispatch) { | |
| 532 if (_iterator == null) { | |
| 533 throw new StateError("No events pending."); | |
| 534 } | |
| 535 // Send one event per call to moveNext. | |
| 536 // If moveNext returns true, send the current element as data. | |
| 537 // If moveNext returns false, send a done event and clear the _iterator. | |
| 538 // If moveNext throws an error, send an error and clear the _iterator. | |
| 539 // After an error, no further events will be sent. | |
| 540 bool isDone; | |
| 541 try { | |
| 542 isDone = !_iterator.moveNext(); | |
| 543 } catch (e, s) { | |
| 544 _iterator = null; | |
| 545 dispatch._sendError(e, s); | |
| 546 return; | |
| 547 } | |
| 548 if (!isDone) { | |
| 549 dispatch._sendData(_iterator.current); | |
| 550 } else { | |
| 551 _iterator = null; | |
| 552 dispatch._sendDone(); | |
| 553 } | |
| 554 } | |
| 555 | |
| 556 void clear() { | |
| 557 if (isScheduled) cancelSchedule(); | |
| 558 _iterator = null; | |
| 559 } | |
| 560 } | |
| 561 | |
| 562 | |
| 563 // Internal helpers. | |
| 564 | |
| 565 // Types of the different handlers on a stream. Types used to type fields. | |
| 566 typedef void _DataHandler<T>(T value); | |
| 567 typedef void _DoneHandler(); | |
| 568 | |
| 569 | |
| 570 /** Default data handler, does nothing. */ | |
| 571 void _nullDataHandler(var value) {} | |
| 572 | |
| 573 /** Default error handler, reports the error to the current zone's handler. */ | |
| 574 void _nullErrorHandler(error, [StackTrace stackTrace]) { | |
| 575 Zone.current.handleUncaughtError(error, stackTrace); | |
| 576 } | |
| 577 | |
| 578 /** Default done handler, does nothing. */ | |
| 579 void _nullDoneHandler() {} | |
| 580 | |
| 581 | |
| 582 /** A delayed event on a buffering stream subscription. */ | |
| 583 abstract class _DelayedEvent<T> { | |
| 584 /** Added as a linked list on the [StreamController]. */ | |
| 585 _DelayedEvent next; | |
| 586 /** Execute the delayed event on the [StreamController]. */ | |
| 587 void perform(_EventDispatch<T> dispatch); | |
| 588 } | |
| 589 | |
| 590 /** A delayed data event. */ | |
| 591 class _DelayedData<T> extends _DelayedEvent<T> { | |
| 592 final T value; | |
| 593 _DelayedData(this.value); | |
| 594 void perform(_EventDispatch<T> dispatch) { | |
| 595 dispatch._sendData(value); | |
| 596 } | |
| 597 } | |
| 598 | |
| 599 /** A delayed error event. */ | |
| 600 class _DelayedError extends _DelayedEvent { | |
| 601 final error; | |
| 602 final StackTrace stackTrace; | |
| 603 | |
| 604 _DelayedError(this.error, this.stackTrace); | |
| 605 void perform(_EventDispatch dispatch) { | |
| 606 dispatch._sendError(error, stackTrace); | |
| 607 } | |
| 608 } | |
| 609 | |
| 610 /** A delayed done event. */ | |
| 611 class _DelayedDone implements _DelayedEvent { | |
| 612 const _DelayedDone(); | |
| 613 void perform(_EventDispatch dispatch) { | |
| 614 dispatch._sendDone(); | |
| 615 } | |
| 616 | |
| 617 _DelayedEvent get next => null; | |
| 618 | |
| 619 void set next(_DelayedEvent _) { | |
| 620 throw new StateError("No events after a done."); | |
| 621 } | |
| 622 } | |
| 623 | |
| 624 /** Superclass for provider of pending events. */ | |
| 625 abstract class _PendingEvents { | |
| 626 // No async event has been scheduled. | |
| 627 static const int _STATE_UNSCHEDULED = 0; | |
| 628 // An async event has been scheduled to run a function. | |
| 629 static const int _STATE_SCHEDULED = 1; | |
| 630 // An async event has been scheduled, but it will do nothing when it runs. | |
| 631 // Async events can't be preempted. | |
| 632 static const int _STATE_CANCELED = 3; | |
| 633 | |
| 634 /** | |
| 635 * State of being scheduled. | |
| 636 * | |
| 637 * Set to [_STATE_SCHEDULED] when pending events are scheduled for | |
| 638 * async dispatch. Since we can't cancel a [scheduleMicrotask] call, if | |
| 639 * scheduling is "canceled", the _state is simply set to [_STATE_CANCELED] | |
| 640 * which will make the async code do nothing except resetting [_state]. | |
| 641 * | |
| 642 * If events are scheduled while the state is [_STATE_CANCELED], it is | |
| 643 * merely switched back to [_STATE_SCHEDULED], but no new call to | |
| 644 * [scheduleMicrotask] is performed. | |
| 645 */ | |
| 646 int _state = _STATE_UNSCHEDULED; | |
| 647 | |
| 648 bool get isEmpty; | |
| 649 | |
| 650 bool get isScheduled => _state == _STATE_SCHEDULED; | |
| 651 bool get _eventScheduled => _state >= _STATE_SCHEDULED; | |
| 652 | |
| 653 /** | |
| 654 * Schedule an event to run later. | |
| 655 * | |
| 656 * If called more than once, it should be called with the same dispatch as | |
| 657 * argument each time. It may reuse an earlier argument in some cases. | |
| 658 */ | |
| 659 void schedule(_EventDispatch dispatch) { | |
| 660 if (isScheduled) return; | |
| 661 assert(!isEmpty); | |
| 662 if (_eventScheduled) { | |
| 663 assert(_state == _STATE_CANCELED); | |
| 664 _state = _STATE_SCHEDULED; | |
| 665 return; | |
| 666 } | |
| 667 scheduleMicrotask(() { | |
| 668 int oldState = _state; | |
| 669 _state = _STATE_UNSCHEDULED; | |
| 670 if (oldState == _STATE_CANCELED) return; | |
| 671 handleNext(dispatch); | |
| 672 }); | |
| 673 _state = _STATE_SCHEDULED; | |
| 674 } | |
| 675 | |
| 676 void cancelSchedule() { | |
| 677 if (isScheduled) _state = _STATE_CANCELED; | |
| 678 } | |
| 679 | |
| 680 void handleNext(_EventDispatch dispatch); | |
| 681 | |
| 682 /** Throw away any pending events and cancel scheduled events. */ | |
| 683 void clear(); | |
| 684 } | |
| 685 | |
| 686 | |
| 687 /** Class holding pending events for a [_StreamImpl]. */ | |
| 688 class _StreamImplEvents extends _PendingEvents { | |
| 689 /// Single linked list of [_DelayedEvent] objects. | |
| 690 _DelayedEvent firstPendingEvent = null; | |
| 691 /// Last element in the list of pending events. New events are added after it. | |
| 692 _DelayedEvent lastPendingEvent = null; | |
| 693 | |
| 694 bool get isEmpty => lastPendingEvent == null; | |
| 695 | |
| 696 void add(_DelayedEvent event) { | |
| 697 if (lastPendingEvent == null) { | |
| 698 firstPendingEvent = lastPendingEvent = event; | |
| 699 } else { | |
| 700 lastPendingEvent = lastPendingEvent.next = event; | |
| 701 } | |
| 702 } | |
| 703 | |
| 704 void handleNext(_EventDispatch dispatch) { | |
| 705 assert(!isScheduled); | |
| 706 _DelayedEvent event = firstPendingEvent; | |
| 707 firstPendingEvent = event.next; | |
| 708 if (firstPendingEvent == null) { | |
| 709 lastPendingEvent = null; | |
| 710 } | |
| 711 event.perform(dispatch); | |
| 712 } | |
| 713 | |
| 714 void clear() { | |
| 715 if (isScheduled) cancelSchedule(); | |
| 716 firstPendingEvent = lastPendingEvent = null; | |
| 717 } | |
| 718 } | |
| 719 | |
| 720 class _BroadcastLinkedList { | |
| 721 _BroadcastLinkedList _next; | |
| 722 _BroadcastLinkedList _previous; | |
| 723 | |
| 724 void _unlink() { | |
| 725 _previous._next = _next; | |
| 726 _next._previous = _previous; | |
| 727 _next = _previous = this; | |
| 728 } | |
| 729 | |
| 730 void _insertBefore(_BroadcastLinkedList newNext) { | |
| 731 _BroadcastLinkedList newPrevious = newNext._previous; | |
| 732 newPrevious._next = this; | |
| 733 newNext._previous = _previous; | |
| 734 _previous._next = newNext; | |
| 735 _previous = newPrevious; | |
| 736 } | |
| 737 } | |
| 738 | |
| 739 typedef void _broadcastCallback(StreamSubscription subscription); | |
| 740 | |
| 741 /** | |
| 742 * Done subscription that will send one done event as soon as possible. | |
| 743 */ | |
| 744 class _DoneStreamSubscription<T> implements StreamSubscription<T> { | |
| 745 static const int _DONE_SENT = 1; | |
| 746 static const int _SCHEDULED = 2; | |
| 747 static const int _PAUSED = 4; | |
| 748 | |
| 749 final Zone _zone; | |
| 750 int _state = 0; | |
| 751 _DoneHandler _onDone; | |
| 752 | |
| 753 _DoneStreamSubscription(this._onDone) : _zone = Zone.current { | |
| 754 _schedule(); | |
| 755 } | |
| 756 | |
| 757 bool get _isSent => (_state & _DONE_SENT) != 0; | |
| 758 bool get _isScheduled => (_state & _SCHEDULED) != 0; | |
| 759 bool get isPaused => _state >= _PAUSED; | |
| 760 | |
| 761 void _schedule() { | |
| 762 if (_isScheduled) return; | |
| 763 _zone.scheduleMicrotask(_sendDone); | |
| 764 _state |= _SCHEDULED; | |
| 765 } | |
| 766 | |
| 767 void onData(void handleData(T data)) {} | |
| 768 void onError(Function handleError) {} | |
| 769 void onDone(void handleDone()) { _onDone = handleDone; } | |
| 770 | |
| 771 void pause([Future resumeSignal]) { | |
| 772 _state += _PAUSED; | |
| 773 if (resumeSignal != null) resumeSignal.whenComplete(resume); | |
| 774 } | |
| 775 | |
| 776 void resume() { | |
| 777 if (isPaused) { | |
| 778 _state -= _PAUSED; | |
| 779 if (!isPaused && !_isSent) { | |
| 780 _schedule(); | |
| 781 } | |
| 782 } | |
| 783 } | |
| 784 | |
| 785 Future cancel() => null; | |
| 786 | |
| 787 Future asFuture([futureValue]) { | |
| 788 _Future result = new _Future(); | |
| 789 _onDone = () { result._completeWithValue(null); }; | |
| 790 return result; | |
| 791 } | |
| 792 | |
| 793 void _sendDone() { | |
| 794 _state &= ~_SCHEDULED; | |
| 795 if (isPaused) return; | |
| 796 _state |= _DONE_SENT; | |
| 797 if (_onDone != null) _zone.runGuarded(_onDone); | |
| 798 } | |
| 799 } | |
| 800 | |
| 801 class _AsBroadcastStream<T> extends Stream<T> { | |
| 802 final Stream<T> _source; | |
| 803 final _broadcastCallback _onListenHandler; | |
| 804 final _broadcastCallback _onCancelHandler; | |
| 805 final Zone _zone; | |
| 806 | |
| 807 _AsBroadcastStreamController<T> _controller; | |
| 808 StreamSubscription<T> _subscription; | |
| 809 | |
| 810 _AsBroadcastStream(this._source, | |
| 811 void onListenHandler(StreamSubscription subscription), | |
| 812 void onCancelHandler(StreamSubscription subscription)) | |
| 813 : _onListenHandler = Zone.current.registerUnaryCallback(onListenHandler), | |
| 814 _onCancelHandler = Zone.current.registerUnaryCallback(onCancelHandler), | |
| 815 _zone = Zone.current { | |
| 816 _controller = new _AsBroadcastStreamController<T>(_onListen, _onCancel); | |
| 817 } | |
| 818 | |
| 819 bool get isBroadcast => true; | |
| 820 | |
| 821 StreamSubscription<T> listen(void onData(T data), | |
| 822 { Function onError, | |
| 823 void onDone(), | |
| 824 bool cancelOnError}) { | |
| 825 if (_controller == null || _controller.isClosed) { | |
| 826 // Return a dummy subscription backed by nothing, since | |
| 827 // it will only ever send one done event. | |
| 828 return new _DoneStreamSubscription<T>(onDone); | |
| 829 } | |
| 830 if (_subscription == null) { | |
| 831 _subscription = _source.listen(_controller.add, | |
| 832 onError: _controller.addError, | |
| 833 onDone: _controller.close); | |
| 834 } | |
| 835 cancelOnError = identical(true, cancelOnError); | |
| 836 return _controller._subscribe(onData, onError, onDone, cancelOnError); | |
| 837 } | |
| 838 | |
| 839 void _onCancel() { | |
| 840 bool shutdown = (_controller == null) || _controller.isClosed; | |
| 841 if (_onCancelHandler != null) { | |
| 842 _zone.runUnary(_onCancelHandler, new _BroadcastSubscriptionWrapper(this)); | |
| 843 } | |
| 844 if (shutdown) { | |
| 845 if (_subscription != null) { | |
| 846 _subscription.cancel(); | |
| 847 _subscription = null; | |
| 848 } | |
| 849 } | |
| 850 } | |
| 851 | |
| 852 void _onListen() { | |
| 853 if (_onListenHandler != null) { | |
| 854 _zone.runUnary(_onListenHandler, new _BroadcastSubscriptionWrapper(this)); | |
| 855 } | |
| 856 } | |
| 857 | |
| 858 // Methods called from _BroadcastSubscriptionWrapper. | |
| 859 void _cancelSubscription() { | |
| 860 if (_subscription == null) return; | |
| 861 // Called by [_controller] when it has no subscribers left. | |
| 862 StreamSubscription subscription = _subscription; | |
| 863 _subscription = null; | |
| 864 _controller = null; // Marks the stream as no longer listenable. | |
| 865 subscription.cancel(); | |
| 866 } | |
| 867 | |
| 868 void _pauseSubscription(Future resumeSignal) { | |
| 869 if (_subscription == null) return; | |
| 870 _subscription.pause(resumeSignal); | |
| 871 } | |
| 872 | |
| 873 void _resumeSubscription() { | |
| 874 if (_subscription == null) return; | |
| 875 _subscription.resume(); | |
| 876 } | |
| 877 | |
| 878 bool get _isSubscriptionPaused { | |
| 879 if (_subscription == null) return false; | |
| 880 return _subscription.isPaused; | |
| 881 } | |
| 882 } | |
| 883 | |
| 884 /** | |
| 885 * Wrapper for subscription that disallows changing handlers. | |
| 886 */ | |
| 887 class _BroadcastSubscriptionWrapper<T> implements StreamSubscription<T> { | |
| 888 final _AsBroadcastStream _stream; | |
| 889 | |
| 890 _BroadcastSubscriptionWrapper(this._stream); | |
| 891 | |
| 892 void onData(void handleData(T data)) { | |
| 893 throw new UnsupportedError( | |
| 894 "Cannot change handlers of asBroadcastStream source subscription."); | |
| 895 } | |
| 896 | |
| 897 void onError(Function handleError) { | |
| 898 throw new UnsupportedError( | |
| 899 "Cannot change handlers of asBroadcastStream source subscription."); | |
| 900 } | |
| 901 | |
| 902 void onDone(void handleDone()) { | |
| 903 throw new UnsupportedError( | |
| 904 "Cannot change handlers of asBroadcastStream source subscription."); | |
| 905 } | |
| 906 | |
| 907 void pause([Future resumeSignal]) { | |
| 908 _stream._pauseSubscription(resumeSignal); | |
| 909 } | |
| 910 | |
| 911 void resume() { | |
| 912 _stream._resumeSubscription(); | |
| 913 } | |
| 914 | |
| 915 Future cancel() { | |
| 916 _stream._cancelSubscription(); | |
| 917 return null; | |
| 918 } | |
| 919 | |
| 920 bool get isPaused { | |
| 921 return _stream._isSubscriptionPaused; | |
| 922 } | |
| 923 | |
| 924 Future asFuture([var futureValue]) { | |
| 925 throw new UnsupportedError( | |
| 926 "Cannot change handlers of asBroadcastStream source subscription."); | |
| 927 } | |
| 928 } | |
| 929 | |
| 930 | |
| 931 /** | |
| 932 * Simple implementation of [StreamIterator]. | |
| 933 */ | |
| 934 class _StreamIteratorImpl<T> implements StreamIterator<T> { | |
| 935 // Internal state of the stream iterator. | |
| 936 // At any time, it is in one of these states. | |
| 937 // The interpretation of the [_futureOrPrefecth] field depends on the state. | |
| 938 // In _STATE_MOVING, the _data field holds the most recently returned | |
| 939 // future. | |
| 940 // When in one of the _STATE_EXTRA_* states, the it may hold the | |
| 941 // next data/error object, and the subscription is paused. | |
| 942 | |
| 943 /// The simple state where [_data] holds the data to return, and [moveNext] | |
| 944 /// is allowed. The subscription is actively listening. | |
| 945 static const int _STATE_FOUND = 0; | |
| 946 /// State set after [moveNext] has returned false or an error, | |
| 947 /// or after calling [cancel]. The subscription is always canceled. | |
| 948 static const int _STATE_DONE = 1; | |
| 949 /// State set after calling [moveNext], but before its returned future has | |
| 950 /// completed. Calling [moveNext] again is not allowed in this state. | |
| 951 /// The subscription is actively listening. | |
| 952 static const int _STATE_MOVING = 2; | |
| 953 /// States set when another event occurs while in _STATE_FOUND. | |
| 954 /// This extra overflow event is cached until the next call to [moveNext], | |
| 955 /// which will complete as if it received the event normally. | |
| 956 /// The subscription is paused in these states, so we only ever get one | |
| 957 /// event too many. | |
| 958 static const int _STATE_EXTRA_DATA = 3; | |
| 959 static const int _STATE_EXTRA_ERROR = 4; | |
| 960 static const int _STATE_EXTRA_DONE = 5; | |
| 961 | |
| 962 /// Subscription being listened to. | |
| 963 StreamSubscription _subscription; | |
| 964 | |
| 965 /// The current element represented by the most recent call to moveNext. | |
| 966 /// | |
| 967 /// Is null between the time moveNext is called and its future completes. | |
| 968 T _current = null; | |
| 969 | |
| 970 /// The future returned by the most recent call to [moveNext]. | |
| 971 /// | |
| 972 /// Also used to store the next value/error in case the stream provides an | |
| 973 /// event before [moveNext] is called again. In that case, the stream will | |
| 974 /// be paused to prevent further events. | |
| 975 var _futureOrPrefetch = null; | |
| 976 | |
| 977 /// The current state. | |
| 978 int _state = _STATE_FOUND; | |
| 979 | |
| 980 _StreamIteratorImpl(final Stream<T> stream) { | |
| 981 _subscription = stream.listen(_onData, | |
| 982 onError: _onError, | |
| 983 onDone: _onDone, | |
| 984 cancelOnError: true); | |
| 985 } | |
| 986 | |
| 987 T get current => _current; | |
| 988 | |
| 989 Future<bool> moveNext() { | |
| 990 if (_state == _STATE_DONE) { | |
| 991 return new _Future<bool>.immediate(false); | |
| 992 } | |
| 993 if (_state == _STATE_MOVING) { | |
| 994 throw new StateError("Already waiting for next."); | |
| 995 } | |
| 996 if (_state == _STATE_FOUND) { | |
| 997 _state = _STATE_MOVING; | |
| 998 _current = null; | |
| 999 _futureOrPrefetch = new _Future<bool>(); | |
| 1000 return _futureOrPrefetch; | |
| 1001 } else { | |
| 1002 assert(_state >= _STATE_EXTRA_DATA); | |
| 1003 switch (_state) { | |
| 1004 case _STATE_EXTRA_DATA: | |
| 1005 _state = _STATE_FOUND; | |
| 1006 _current = _futureOrPrefetch; | |
| 1007 _futureOrPrefetch = null; | |
| 1008 _subscription.resume(); | |
| 1009 return new _Future<bool>.immediate(true); | |
| 1010 case _STATE_EXTRA_ERROR: | |
| 1011 AsyncError prefetch = _futureOrPrefetch; | |
| 1012 _clear(); | |
| 1013 return new _Future<bool>.immediateError(prefetch.error, | |
| 1014 prefetch.stackTrace); | |
| 1015 case _STATE_EXTRA_DONE: | |
| 1016 _clear(); | |
| 1017 return new _Future<bool>.immediate(false); | |
| 1018 } | |
| 1019 } | |
| 1020 } | |
| 1021 | |
| 1022 /** Clears up the internal state when the iterator ends. */ | |
| 1023 void _clear() { | |
| 1024 _subscription = null; | |
| 1025 _futureOrPrefetch = null; | |
| 1026 _current = null; | |
| 1027 _state = _STATE_DONE; | |
| 1028 } | |
| 1029 | |
| 1030 Future cancel() { | |
| 1031 StreamSubscription subscription = _subscription; | |
| 1032 if (_state == _STATE_MOVING) { | |
| 1033 _Future<bool> hasNext = _futureOrPrefetch; | |
| 1034 _clear(); | |
| 1035 hasNext._complete(false); | |
| 1036 } else { | |
| 1037 _clear(); | |
| 1038 } | |
| 1039 return subscription.cancel(); | |
| 1040 } | |
| 1041 | |
| 1042 void _onData(T data) { | |
| 1043 if (_state == _STATE_MOVING) { | |
| 1044 _current = data; | |
| 1045 _Future<bool> hasNext = _futureOrPrefetch; | |
| 1046 _futureOrPrefetch = null; | |
| 1047 _state = _STATE_FOUND; | |
| 1048 hasNext._complete(true); | |
| 1049 return; | |
| 1050 } | |
| 1051 _subscription.pause(); | |
| 1052 assert(_futureOrPrefetch == null); | |
| 1053 _futureOrPrefetch = data; | |
| 1054 _state = _STATE_EXTRA_DATA; | |
| 1055 } | |
| 1056 | |
| 1057 void _onError(Object error, [StackTrace stackTrace]) { | |
| 1058 if (_state == _STATE_MOVING) { | |
| 1059 _Future<bool> hasNext = _futureOrPrefetch; | |
| 1060 // We have cancelOnError: true, so the subscription is canceled. | |
| 1061 _clear(); | |
| 1062 hasNext._completeError(error, stackTrace); | |
| 1063 return; | |
| 1064 } | |
| 1065 _subscription.pause(); | |
| 1066 assert(_futureOrPrefetch == null); | |
| 1067 _futureOrPrefetch = new AsyncError(error, stackTrace); | |
| 1068 _state = _STATE_EXTRA_ERROR; | |
| 1069 } | |
| 1070 | |
| 1071 void _onDone() { | |
| 1072 if (_state == _STATE_MOVING) { | |
| 1073 _Future<bool> hasNext = _futureOrPrefetch; | |
| 1074 _clear(); | |
| 1075 hasNext._complete(false); | |
| 1076 return; | |
| 1077 } | |
| 1078 _subscription.pause(); | |
| 1079 _futureOrPrefetch = null; | |
| 1080 _state = _STATE_EXTRA_DONE; | |
| 1081 } | |
| 1082 } | |
| OLD | NEW |
|---|
Chromium Code Reviews
Issue 1162723007:
remove generated_sdk from checked in code (Closed)
Base URL: git@github.com:dart-lang/dev_compiler.git@master