| 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 // ------------------------------------------------------------------- | |
| 8 // Controller for creating and adding events to a stream. | |
| 9 // ------------------------------------------------------------------- | |
| 10 | |
| 11 /** | |
| 12 * A controller with the stream it controls. | |
| 13 * | |
| 14 * This controller allows sending data, error and done events on | |
| 15 * its [stream]. | |
| 16 * This class can be used to create a simple stream that others | |
| 17 * can listen on, and to push events to that stream. | |
| 18 * | |
| 19 * It's possible to check whether the stream is paused or not, and whether | |
| 20 * it has subscribers or not, as well as getting a callback when either of | |
| 21 * these change. | |
| 22 * | |
| 23 * If the stream starts or stops having listeners (first listener subscribing, | |
| 24 * last listener unsubscribing), the `onSubscriptionStateChange` callback | |
| 25 * is notified as soon as possible. If the subscription stat changes during | |
| 26 * an event firing or a callback being executed, the change will not be reported | |
| 27 * until the current event or callback has finished. | |
| 28 * If the pause state has also changed during an event or callback, only the | |
| 29 * subscription state callback is notified. | |
| 30 * | |
| 31 * If the subscriber state has not changed, but the pause state has, the | |
| 32 * `onPauseStateChange` callback is notified as soon as possible, after firing | |
| 33 * a current event or completing another callback. This happens if the stream | |
| 34 * is not paused, and a listener pauses it, or if the stream has been resumed | |
| 35 * from pause and has no pending events. If the listeners resume a paused stream | |
| 36 * while it still has queued events, the controller will still consider the | |
| 37 * stream paused until all queued events have been dispatched. | |
| 38 * | |
| 39 * Whether to invoke a callback depends only on the state before and after | |
| 40 * a stream action, for example firing an event. If the state changes multiple | |
| 41 * times during the action, and then ends up in the same state as before, no | |
| 42 * callback is performed. | |
| 43 * | |
| 44 * If listeners are added after the stream has completed (sent a "done" event), | |
| 45 * the listeners will be sent a "done" event eventually, but they won't affect | |
| 46 * the stream at all, and won't trigger callbacks. From the controller's point | |
| 47 * of view, the stream is completely inert when has completed. | |
| 48 */ | |
| 49 abstract class StreamController<T> implements StreamSink<T> { | |
| 50 /** The stream that this controller is controlling. */ | |
| 51 Stream<T> get stream; | |
| 52 | |
| 53 /** | |
| 54 * A controller with a [stream] that supports only one single subscriber. | |
| 55 * | |
| 56 * If [sync] is true, events may be passed directly to the stream's listener | |
| 57 * during an [add], [addError] or [close] call. If [sync] is false, the event | |
| 58 * will be passed to the listener at a later time, after the code creating | |
| 59 * the event has returned. | |
| 60 * | |
| 61 * The controller will buffer all incoming events until the subscriber is | |
| 62 * registered. | |
| 63 * | |
| 64 * The [onPause] function is called when the stream becomes | |
| 65 * paused. [onResume] is called when the stream resumed. | |
| 66 * | |
| 67 * The [onListen] callback is called when the stream | |
| 68 * receives its listener and [onCancel] when the listener ends | |
| 69 * its subscription. If [onCancel] needs to perform an asynchronous operation, | |
| 70 * [onCancel] should return a future that completes when the cancel operation | |
| 71 * is done. | |
| 72 * | |
| 73 * If the stream is canceled before the controller needs new data the | |
| 74 * [onResume] call might not be executed. | |
| 75 */ | |
| 76 factory StreamController({void onListen(), | |
| 77 void onPause(), | |
| 78 void onResume(), | |
| 79 onCancel(), | |
| 80 bool sync: false}) { | |
| 81 if (onListen == null && onPause == null && | |
| 82 onResume == null && onCancel == null) { | |
| 83 return sync | |
| 84 ? new _NoCallbackSyncStreamController/*<T>*/() | |
| 85 : new _NoCallbackAsyncStreamController/*<T>*/(); | |
| 86 } | |
| 87 return sync | |
| 88 ? new _SyncStreamController<T>(onListen, onPause, onResume, onCancel) | |
| 89 : new _AsyncStreamController<T>(onListen, onPause, onResume, onCancel); | |
| 90 } | |
| 91 | |
| 92 /** | |
| 93 * A controller where [stream] can be listened to more than once. | |
| 94 * | |
| 95 * The [Stream] returned by [stream] is a broadcast stream. | |
| 96 * It can be listened to more than once. | |
| 97 * | |
| 98 * The controller distributes any events to all currently subscribed | |
| 99 * listeners at the time when [add], [addError] or [close] is called. | |
| 100 * It is not allowed to call `add`, `addError`, or `close` before a previous | |
| 101 * call has returned. The controller does not have any internal queue of | |
| 102 * events, and if there are no listeners at the time the event is added, | |
| 103 * it will just be dropped, or, if it is an error, be reported as uncaught. | |
| 104 * | |
| 105 * Each listener subscription is handled independently, | |
| 106 * and if one pauses, only the pausing listener is affected. | |
| 107 * A paused listener will buffer events internally until unpaused or canceled. | |
| 108 * | |
| 109 * If [sync] is true, events may be fired directly by the stream's | |
| 110 * subscriptions during an [add], [addError] or [close] call. | |
| 111 * If [sync] is false, the event will be fired at a later time, | |
| 112 * after the code adding the event has completed. | |
| 113 * | |
| 114 * When [sync] is false, no guarantees are given with regard to when | |
| 115 * multiple listeners get the events, except that each listener will get | |
| 116 * all events in the correct order. Each subscription handles the events | |
| 117 * individually. | |
| 118 * If two events are sent on an async controller with two listeners, | |
| 119 * one of the listeners may get both events | |
| 120 * before the other listener gets any. | |
| 121 * A listener must be subscribed both when the event is initiated | |
| 122 * (that is, when [add] is called) | |
| 123 * and when the event is later delivered, | |
| 124 * in order to receive the event. | |
| 125 * | |
| 126 * The [onListen] callback is called when the first listener is subscribed, | |
| 127 * and the [onCancel] is called when there are no longer any active listeners. | |
| 128 * If a listener is added again later, after the [onCancel] was called, | |
| 129 * the [onListen] will be called again. | |
| 130 */ | |
| 131 factory StreamController.broadcast({void onListen(), | |
| 132 void onCancel(), | |
| 133 bool sync: false}) { | |
| 134 return sync | |
| 135 ? new _SyncBroadcastStreamController<T>(onListen, onCancel) | |
| 136 : new _AsyncBroadcastStreamController<T>(onListen, onCancel); | |
| 137 } | |
| 138 | |
| 139 /** | |
| 140 * Returns a view of this object that only exposes the [StreamSink] interface. | |
| 141 */ | |
| 142 StreamSink<T> get sink; | |
| 143 | |
| 144 /** | |
| 145 * Whether the stream is closed for adding more events. | |
| 146 * | |
| 147 * If true, the "done" event might not have fired yet, but it has been | |
| 148 * scheduled, and it is too late to add more events. | |
| 149 */ | |
| 150 bool get isClosed; | |
| 151 | |
| 152 /** | |
| 153 * Whether the subscription would need to buffer events. | |
| 154 * | |
| 155 * This is the case if the controller's stream has a listener and it is | |
| 156 * paused, or if it has not received a listener yet. In that case, the | |
| 157 * controller is considered paused as well. | |
| 158 * | |
| 159 * A broadcast stream controller is never considered paused. It always | |
| 160 * forwards its events to all uncanceled listeners, if any, and let them | |
| 161 * handle their own pausing. | |
| 162 */ | |
| 163 bool get isPaused; | |
| 164 | |
| 165 /** Whether there is a subscriber on the [Stream]. */ | |
| 166 bool get hasListener; | |
| 167 | |
| 168 /** | |
| 169 * Send or enqueue an error event. | |
| 170 * | |
| 171 * If [error] is `null`, it is replaced by a [NullThrownError]. | |
| 172 * | |
| 173 * Also allows an objection stack trace object, on top of what [EventSink] | |
| 174 * allows. | |
| 175 */ | |
| 176 void addError(Object error, [StackTrace stackTrace]); | |
| 177 | |
| 178 /** | |
| 179 * Receives events from [source] and puts them into this controller's stream. | |
| 180 * | |
| 181 * Returns a future which completes when the source stream is done. | |
| 182 * | |
| 183 * Events must not be added directly to this controller using [add], | |
| 184 * [addError], [close] or [addStream], until the returned future | |
| 185 * is complete. | |
| 186 * | |
| 187 * Data and error events are forwarded to this controller's stream. A done | |
| 188 * event on the source will end the `addStream` operation and complete the | |
| 189 * returned future. | |
| 190 * | |
| 191 * If [cancelOnError] is true, only the first error on [source] is | |
| 192 * forwarded to the controller's stream, and the `addStream` ends | |
| 193 * after this. If [cancelOnError] is false, all errors are forwarded | |
| 194 * and only a done event will end the `addStream`. | |
| 195 */ | |
| 196 Future addStream(Stream<T> source, {bool cancelOnError: true}); | |
| 197 } | |
| 198 | |
| 199 | |
| 200 abstract class _StreamControllerLifecycle<T> { | |
| 201 StreamSubscription<T> _subscribe( | |
| 202 void onData(T data), | |
| 203 Function onError, | |
| 204 void onDone(), | |
| 205 bool cancelOnError); | |
| 206 void _recordPause(StreamSubscription<T> subscription) {} | |
| 207 void _recordResume(StreamSubscription<T> subscription) {} | |
| 208 Future _recordCancel(StreamSubscription<T> subscription) => null; | |
| 209 } | |
| 210 | |
| 211 /** | |
| 212 * Default implementation of [StreamController]. | |
| 213 * | |
| 214 * Controls a stream that only supports a single controller. | |
| 215 */ | |
| 216 abstract class _StreamController<T> implements StreamController<T>, | |
| 217 _StreamControllerLifecycle<T>, | |
| 218 _EventSink<T>, | |
| 219 _EventDispatch<T> { | |
| 220 // The states are bit-flags. More than one can be set at a time. | |
| 221 // | |
| 222 // The "subscription state" goes through the states: | |
| 223 // initial -> subscribed -> canceled. | |
| 224 // These are mutually exclusive. | |
| 225 // The "closed" state records whether the [close] method has been called | |
| 226 // on the controller. This can be done at any time. If done before | |
| 227 // subscription, the done event is queued. If done after cancel, the done | |
| 228 // event is ignored (just as any other event after a cancel). | |
| 229 | |
| 230 /** The controller is in its initial state with no subscription. */ | |
| 231 static const int _STATE_INITIAL = 0; | |
| 232 /** The controller has a subscription, but hasn't been closed or canceled. */ | |
| 233 static const int _STATE_SUBSCRIBED = 1; | |
| 234 /** The subscription is canceled. */ | |
| 235 static const int _STATE_CANCELED = 2; | |
| 236 /** Mask for the subscription state. */ | |
| 237 static const int _STATE_SUBSCRIPTION_MASK = 3; | |
| 238 | |
| 239 // The following state relate to the controller, not the subscription. | |
| 240 // If closed, adding more events is not allowed. | |
| 241 // If executing an [addStream], new events are not allowed either, but will | |
| 242 // be added by the stream. | |
| 243 | |
| 244 /** | |
| 245 * The controller is closed due to calling [close]. | |
| 246 * | |
| 247 * When the stream is closed, you can neither add new events nor add new | |
| 248 * listeners. | |
| 249 */ | |
| 250 static const int _STATE_CLOSED = 4; | |
| 251 /** | |
| 252 * The controller is in the middle of an [addStream] operation. | |
| 253 * | |
| 254 * While adding events from a stream, no new events can be added directly | |
| 255 * on the controller. | |
| 256 */ | |
| 257 static const int _STATE_ADDSTREAM = 8; | |
| 258 | |
| 259 /** | |
| 260 * Field containing different data depending on the current subscription | |
| 261 * state. | |
| 262 * | |
| 263 * If [_state] is [_STATE_INITIAL], the field may contain a [_PendingEvents] | |
| 264 * for events added to the controller before a subscription. | |
| 265 * | |
| 266 * While [_state] is [_STATE_SUBSCRIBED], the field contains the subscription. | |
| 267 * | |
| 268 * When [_state] is [_STATE_CANCELED] the field is currently not used. | |
| 269 */ | |
| 270 var _varData; | |
| 271 | |
| 272 /** Current state of the controller. */ | |
| 273 int _state = _STATE_INITIAL; | |
| 274 | |
| 275 /** | |
| 276 * Future completed when the stream sends its last event. | |
| 277 * | |
| 278 * This is also the future returned by [close]. | |
| 279 */ | |
| 280 // TODO(lrn): Could this be stored in the varData field too, if it's not | |
| 281 // accessed until the call to "close"? Then we need to special case if it's | |
| 282 // accessed earlier, or if close is called before subscribing. | |
| 283 _Future _doneFuture; | |
| 284 | |
| 285 _StreamController(); | |
| 286 | |
| 287 _NotificationHandler get _onListen; | |
| 288 _NotificationHandler get _onPause; | |
| 289 _NotificationHandler get _onResume; | |
| 290 _NotificationHandler get _onCancel; | |
| 291 | |
| 292 // Return a new stream every time. The streams are equal, but not identical. | |
| 293 Stream<T> get stream => new _ControllerStream(this); | |
| 294 | |
| 295 /** | |
| 296 * Returns a view of this object that only exposes the [StreamSink] interface. | |
| 297 */ | |
| 298 StreamSink<T> get sink => new _StreamSinkWrapper<T>(this); | |
| 299 | |
| 300 /** | |
| 301 * Whether a listener has existed and been canceled. | |
| 302 * | |
| 303 * After this, adding more events will be ignored. | |
| 304 */ | |
| 305 bool get _isCanceled => (_state & _STATE_CANCELED) != 0; | |
| 306 | |
| 307 /** Whether there is an active listener. */ | |
| 308 bool get hasListener => (_state & _STATE_SUBSCRIBED) != 0; | |
| 309 | |
| 310 /** Whether there has not been a listener yet. */ | |
| 311 bool get _isInitialState => | |
| 312 (_state & _STATE_SUBSCRIPTION_MASK) == _STATE_INITIAL; | |
| 313 | |
| 314 bool get isClosed => (_state & _STATE_CLOSED) != 0; | |
| 315 | |
| 316 bool get isPaused => hasListener ? _subscription._isInputPaused | |
| 317 : !_isCanceled; | |
| 318 | |
| 319 bool get _isAddingStream => (_state & _STATE_ADDSTREAM) != 0; | |
| 320 | |
| 321 /** New events may not be added after close, or during addStream. */ | |
| 322 bool get _mayAddEvent => (_state < _STATE_CLOSED); | |
| 323 | |
| 324 // Returns the pending events. | |
| 325 // Pending events are events added before a subscription exists. | |
| 326 // They are added to the subscription when it is created. | |
| 327 // Pending events, if any, are kept in the _varData field until the | |
| 328 // stream is listened to. | |
| 329 // While adding a stream, pending events are moved into the | |
| 330 // state object to allow the state object to use the _varData field. | |
| 331 _PendingEvents get _pendingEvents { | |
| 332 assert(_isInitialState); | |
| 333 if (!_isAddingStream) { | |
| 334 return _varData; | |
| 335 } | |
| 336 _StreamControllerAddStreamState state = _varData; | |
| 337 return state.varData; | |
| 338 } | |
| 339 | |
| 340 // Returns the pending events, and creates the object if necessary. | |
| 341 _StreamImplEvents _ensurePendingEvents() { | |
| 342 assert(_isInitialState); | |
| 343 if (!_isAddingStream) { | |
| 344 if (_varData == null) _varData = new _StreamImplEvents(); | |
| 345 return _varData; | |
| 346 } | |
| 347 _StreamControllerAddStreamState state = _varData; | |
| 348 if (state.varData == null) state.varData = new _StreamImplEvents(); | |
| 349 return state.varData; | |
| 350 } | |
| 351 | |
| 352 // Get the current subscription. | |
| 353 // If we are adding a stream, the subscription is moved into the state | |
| 354 // object to allow the state object to use the _varData field. | |
| 355 _ControllerSubscription get _subscription { | |
| 356 assert(hasListener); | |
| 357 if (_isAddingStream) { | |
| 358 _StreamControllerAddStreamState addState = _varData; | |
| 359 return addState.varData; | |
| 360 } | |
| 361 return _varData; | |
| 362 } | |
| 363 | |
| 364 /** | |
| 365 * Creates an error describing why an event cannot be added. | |
| 366 * | |
| 367 * The reason, and therefore the error message, depends on the current state. | |
| 368 */ | |
| 369 Error _badEventState() { | |
| 370 if (isClosed) { | |
| 371 return new StateError("Cannot add event after closing"); | |
| 372 } | |
| 373 assert(_isAddingStream); | |
| 374 return new StateError("Cannot add event while adding a stream"); | |
| 375 } | |
| 376 | |
| 377 // StreamSink interface. | |
| 378 Future addStream(Stream<T> source, {bool cancelOnError: true}) { | |
| 379 if (!_mayAddEvent) throw _badEventState(); | |
| 380 if (_isCanceled) return new _Future.immediate(null); | |
| 381 _StreamControllerAddStreamState addState = | |
| 382 new _StreamControllerAddStreamState(this, | |
| 383 _varData, | |
| 384 source, | |
| 385 cancelOnError); | |
| 386 _varData = addState; | |
| 387 _state |= _STATE_ADDSTREAM; | |
| 388 return addState.addStreamFuture; | |
| 389 } | |
| 390 | |
| 391 /** | |
| 392 * Returns a future that is completed when the stream is done | |
| 393 * processing events. | |
| 394 * | |
| 395 * This happens either when the done event has been sent, or if the | |
| 396 * subscriber of a single-subscription stream is cancelled. | |
| 397 */ | |
| 398 Future get done => _ensureDoneFuture(); | |
| 399 | |
| 400 Future _ensureDoneFuture() { | |
| 401 if (_doneFuture == null) { | |
| 402 _doneFuture = _isCanceled ? Future._nullFuture : new _Future(); | |
| 403 } | |
| 404 return _doneFuture; | |
| 405 } | |
| 406 | |
| 407 /** | |
| 408 * Send or enqueue a data event. | |
| 409 */ | |
| 410 void add(T value) { | |
| 411 if (!_mayAddEvent) throw _badEventState(); | |
| 412 _add(value); | |
| 413 } | |
| 414 | |
| 415 /** | |
| 416 * Send or enqueue an error event. | |
| 417 */ | |
| 418 void addError(Object error, [StackTrace stackTrace]) { | |
| 419 error = _nonNullError(error); | |
| 420 if (!_mayAddEvent) throw _badEventState(); | |
| 421 AsyncError replacement = Zone.current.errorCallback(error, stackTrace); | |
| 422 if (replacement != null) { | |
| 423 error = _nonNullError(replacement.error); | |
| 424 stackTrace = replacement.stackTrace; | |
| 425 } | |
| 426 _addError(error, stackTrace); | |
| 427 } | |
| 428 | |
| 429 /** | |
| 430 * Closes this controller and sends a done event on the stream. | |
| 431 * | |
| 432 * The first time a controller is closed, a "done" event is added to its | |
| 433 * stream. | |
| 434 * | |
| 435 * You are allowed to close the controller more than once, but only the first | |
| 436 * call has any effect. | |
| 437 * | |
| 438 * After closing, no further events may be added using [add] or [addError]. | |
| 439 * | |
| 440 * The returned future is completed when the done event has been delivered. | |
| 441 */ | |
| 442 Future close() { | |
| 443 if (isClosed) { | |
| 444 return _ensureDoneFuture(); | |
| 445 } | |
| 446 if (!_mayAddEvent) throw _badEventState(); | |
| 447 _closeUnchecked(); | |
| 448 return _ensureDoneFuture(); | |
| 449 } | |
| 450 | |
| 451 void _closeUnchecked() { | |
| 452 _state |= _STATE_CLOSED; | |
| 453 if (hasListener) { | |
| 454 _sendDone(); | |
| 455 } else if (_isInitialState) { | |
| 456 _ensurePendingEvents().add(const _DelayedDone()); | |
| 457 } | |
| 458 } | |
| 459 | |
| 460 // EventSink interface. Used by the [addStream] events. | |
| 461 | |
| 462 // Add data event, used both by the [addStream] events and by [add]. | |
| 463 void _add(T value) { | |
| 464 if (hasListener) { | |
| 465 _sendData(value); | |
| 466 } else if (_isInitialState) { | |
| 467 _ensurePendingEvents().add(new _DelayedData<T>(value)); | |
| 468 } | |
| 469 } | |
| 470 | |
| 471 void _addError(Object error, StackTrace stackTrace) { | |
| 472 if (hasListener) { | |
| 473 _sendError(error, stackTrace); | |
| 474 } else if (_isInitialState) { | |
| 475 _ensurePendingEvents().add(new _DelayedError(error, stackTrace)); | |
| 476 } | |
| 477 } | |
| 478 | |
| 479 void _close() { | |
| 480 // End of addStream stream. | |
| 481 assert(_isAddingStream); | |
| 482 _StreamControllerAddStreamState addState = _varData; | |
| 483 _varData = addState.varData; | |
| 484 _state &= ~_STATE_ADDSTREAM; | |
| 485 addState.complete(); | |
| 486 } | |
| 487 | |
| 488 // _StreamControllerLifeCycle interface | |
| 489 | |
| 490 StreamSubscription<T> _subscribe( | |
| 491 void onData(T data), | |
| 492 Function onError, | |
| 493 void onDone(), | |
| 494 bool cancelOnError) { | |
| 495 if (!_isInitialState) { | |
| 496 throw new StateError("Stream has already been listened to."); | |
| 497 } | |
| 498 _ControllerSubscription subscription = | |
| 499 new _ControllerSubscription(this, onData, onError, onDone, | |
| 500 cancelOnError); | |
| 501 | |
| 502 _PendingEvents pendingEvents = _pendingEvents; | |
| 503 _state |= _STATE_SUBSCRIBED; | |
| 504 if (_isAddingStream) { | |
| 505 _StreamControllerAddStreamState addState = _varData; | |
| 506 addState.varData = subscription; | |
| 507 addState.resume(); | |
| 508 } else { | |
| 509 _varData = subscription; | |
| 510 } | |
| 511 subscription._setPendingEvents(pendingEvents); | |
| 512 subscription._guardCallback(() { | |
| 513 _runGuarded(_onListen); | |
| 514 }); | |
| 515 | |
| 516 return subscription; | |
| 517 } | |
| 518 | |
| 519 Future _recordCancel(StreamSubscription<T> subscription) { | |
| 520 // When we cancel, we first cancel any stream being added, | |
| 521 // Then we call _onCancel, and finally the _doneFuture is completed. | |
| 522 // If either of addStream's cancel or _onCancel returns a future, | |
| 523 // we wait for it before continuing. | |
| 524 // Any error during this process ends up in the returned future. | |
| 525 // If more errors happen, we act as if it happens inside nested try/finallys | |
| 526 // or whenComplete calls, and only the last error ends up in the | |
| 527 // returned future. | |
| 528 Future result; | |
| 529 if (_isAddingStream) { | |
| 530 _StreamControllerAddStreamState addState = _varData; | |
| 531 result = addState.cancel(); | |
| 532 } | |
| 533 _varData = null; | |
| 534 _state = | |
| 535 (_state & ~(_STATE_SUBSCRIBED | _STATE_ADDSTREAM)) | _STATE_CANCELED; | |
| 536 | |
| 537 if (_onCancel != null) { | |
| 538 if (result == null) { | |
| 539 // Only introduce a future if one is needed. | |
| 540 // If _onCancel returns null, no future is needed. | |
| 541 try { | |
| 542 result = _onCancel(); | |
| 543 } catch (e, s) { | |
| 544 // Return the error in the returned future. | |
| 545 // Complete it asynchronously, so there is time for a listener | |
| 546 // to handle the error. | |
| 547 result = new _Future().._asyncCompleteError(e, s); | |
| 548 } | |
| 549 } else { | |
| 550 // Simpler case when we already know that we will return a future. | |
| 551 result = result.whenComplete(_onCancel); | |
| 552 } | |
| 553 } | |
| 554 | |
| 555 void complete() { | |
| 556 if (_doneFuture != null && _doneFuture._mayComplete) { | |
| 557 _doneFuture._asyncComplete(null); | |
| 558 } | |
| 559 } | |
| 560 | |
| 561 if (result != null) { | |
| 562 result = result.whenComplete(complete); | |
| 563 } else { | |
| 564 complete(); | |
| 565 } | |
| 566 | |
| 567 return result; | |
| 568 } | |
| 569 | |
| 570 void _recordPause(StreamSubscription<T> subscription) { | |
| 571 if (_isAddingStream) { | |
| 572 _StreamControllerAddStreamState addState = _varData; | |
| 573 addState.pause(); | |
| 574 } | |
| 575 _runGuarded(_onPause); | |
| 576 } | |
| 577 | |
| 578 void _recordResume(StreamSubscription<T> subscription) { | |
| 579 if (_isAddingStream) { | |
| 580 _StreamControllerAddStreamState addState = _varData; | |
| 581 addState.resume(); | |
| 582 } | |
| 583 _runGuarded(_onResume); | |
| 584 } | |
| 585 } | |
| 586 | |
| 587 abstract class _SyncStreamControllerDispatch<T> | |
| 588 implements _StreamController<T> { | |
| 589 void _sendData(T data) { | |
| 590 _subscription._add(data); | |
| 591 } | |
| 592 | |
| 593 void _sendError(Object error, StackTrace stackTrace) { | |
| 594 _subscription._addError(error, stackTrace); | |
| 595 } | |
| 596 | |
| 597 void _sendDone() { | |
| 598 _subscription._close(); | |
| 599 } | |
| 600 } | |
| 601 | |
| 602 abstract class _AsyncStreamControllerDispatch<T> | |
| 603 implements _StreamController<T> { | |
| 604 void _sendData(T data) { | |
| 605 _subscription._addPending(new _DelayedData(data)); | |
| 606 } | |
| 607 | |
| 608 void _sendError(Object error, StackTrace stackTrace) { | |
| 609 _subscription._addPending(new _DelayedError(error, stackTrace)); | |
| 610 } | |
| 611 | |
| 612 void _sendDone() { | |
| 613 _subscription._addPending(const _DelayedDone()); | |
| 614 } | |
| 615 } | |
| 616 | |
| 617 // TODO(lrn): Use common superclass for callback-controllers when VM supports | |
| 618 // constructors in mixin superclasses. | |
| 619 | |
| 620 class _AsyncStreamController<T> extends _StreamController<T> | |
| 621 with _AsyncStreamControllerDispatch<T> { | |
| 622 final _NotificationHandler _onListen; | |
| 623 final _NotificationHandler _onPause; | |
| 624 final _NotificationHandler _onResume; | |
| 625 final _NotificationHandler _onCancel; | |
| 626 | |
| 627 _AsyncStreamController(void this._onListen(), | |
| 628 void this._onPause(), | |
| 629 void this._onResume(), | |
| 630 this._onCancel()); | |
| 631 } | |
| 632 | |
| 633 class _SyncStreamController<T> extends _StreamController<T> | |
| 634 with _SyncStreamControllerDispatch<T> { | |
| 635 final _NotificationHandler _onListen; | |
| 636 final _NotificationHandler _onPause; | |
| 637 final _NotificationHandler _onResume; | |
| 638 final _NotificationHandler _onCancel; | |
| 639 | |
| 640 _SyncStreamController(void this._onListen(), | |
| 641 void this._onPause(), | |
| 642 void this._onResume(), | |
| 643 this._onCancel()); | |
| 644 } | |
| 645 | |
| 646 abstract class _NoCallbacks { | |
| 647 _NotificationHandler get _onListen => null; | |
| 648 _NotificationHandler get _onPause => null; | |
| 649 _NotificationHandler get _onResume => null; | |
| 650 _NotificationHandler get _onCancel => null; | |
| 651 } | |
| 652 | |
| 653 class _NoCallbackAsyncStreamController/*<T>*/ = _StreamController/*<T>*/ | |
| 654 with _AsyncStreamControllerDispatch/*<T>*/, _NoCallbacks; | |
| 655 | |
| 656 class _NoCallbackSyncStreamController/*<T>*/ = _StreamController/*<T>*/ | |
| 657 with _SyncStreamControllerDispatch/*<T>*/, _NoCallbacks; | |
| 658 | |
| 659 typedef _NotificationHandler(); | |
| 660 | |
| 661 Future _runGuarded(_NotificationHandler notificationHandler) { | |
| 662 if (notificationHandler == null) return null; | |
| 663 try { | |
| 664 var result = notificationHandler(); | |
| 665 if (result is Future) return result; | |
| 666 return null; | |
| 667 } catch (e, s) { | |
| 668 Zone.current.handleUncaughtError(e, s); | |
| 669 } | |
| 670 } | |
| 671 | |
| 672 class _ControllerStream<T> extends _StreamImpl<T> { | |
| 673 _StreamControllerLifecycle<T> _controller; | |
| 674 | |
| 675 _ControllerStream(this._controller); | |
| 676 | |
| 677 StreamSubscription<T> _createSubscription( | |
| 678 void onData(T data), | |
| 679 Function onError, | |
| 680 void onDone(), | |
| 681 bool cancelOnError) => | |
| 682 _controller._subscribe(onData, onError, onDone, cancelOnError); | |
| 683 | |
| 684 // Override == and hashCode so that new streams returned by the same | |
| 685 // controller are considered equal. The controller returns a new stream | |
| 686 // each time it's queried, but doesn't have to cache the result. | |
| 687 | |
| 688 int get hashCode => _controller.hashCode ^ 0x35323532; | |
| 689 | |
| 690 bool operator==(Object other) { | |
| 691 if (identical(this, other)) return true; | |
| 692 if (other is! _ControllerStream) return false; | |
| 693 _ControllerStream otherStream = other; | |
| 694 return identical(otherStream._controller, this._controller); | |
| 695 } | |
| 696 } | |
| 697 | |
| 698 class _ControllerSubscription<T> extends _BufferingStreamSubscription<T> { | |
| 699 final _StreamControllerLifecycle<T> _controller; | |
| 700 | |
| 701 _ControllerSubscription(this._controller, void onData(T data), | |
| 702 Function onError, void onDone(), bool cancelOnError) | |
| 703 : super(onData, onError, onDone, cancelOnError); | |
| 704 | |
| 705 Future _onCancel() { | |
| 706 return _controller._recordCancel(this); | |
| 707 } | |
| 708 | |
| 709 void _onPause() { | |
| 710 _controller._recordPause(this); | |
| 711 } | |
| 712 | |
| 713 void _onResume() { | |
| 714 _controller._recordResume(this); | |
| 715 } | |
| 716 } | |
| 717 | |
| 718 | |
| 719 /** A class that exposes only the [StreamSink] interface of an object. */ | |
| 720 class _StreamSinkWrapper<T> implements StreamSink<T> { | |
| 721 final StreamController _target; | |
| 722 _StreamSinkWrapper(this._target); | |
| 723 void add(T data) { _target.add(data); } | |
| 724 void addError(Object error, [StackTrace stackTrace]) { | |
| 725 _target.addError(error, stackTrace); | |
| 726 } | |
| 727 Future close() => _target.close(); | |
| 728 Future addStream(Stream<T> source, {bool cancelOnError: true}) => | |
| 729 _target.addStream(source, cancelOnError: cancelOnError); | |
| 730 Future get done => _target.done; | |
| 731 } | |
| 732 | |
| 733 /** | |
| 734 * Object containing the state used to handle [StreamController.addStream]. | |
| 735 */ | |
| 736 class _AddStreamState<T> { | |
| 737 // [_Future] returned by call to addStream. | |
| 738 final _Future addStreamFuture; | |
| 739 | |
| 740 // Subscription on stream argument to addStream. | |
| 741 final StreamSubscription addSubscription; | |
| 742 | |
| 743 _AddStreamState(_EventSink<T> controller, Stream source, bool cancelOnError) | |
| 744 : addStreamFuture = new _Future(), | |
| 745 addSubscription = source.listen(controller._add, | |
| 746 onError: cancelOnError | |
| 747 ? makeErrorHandler(controller) | |
| 748 : controller._addError, | |
| 749 onDone: controller._close, | |
| 750 cancelOnError: cancelOnError); | |
| 751 | |
| 752 static makeErrorHandler(_EventSink controller) => | |
| 753 (e, StackTrace s) { | |
| 754 controller._addError(e, s); | |
| 755 controller._close(); | |
| 756 }; | |
| 757 | |
| 758 void pause() { | |
| 759 addSubscription.pause(); | |
| 760 } | |
| 761 | |
| 762 void resume() { | |
| 763 addSubscription.resume(); | |
| 764 } | |
| 765 | |
| 766 /** | |
| 767 * Stop adding the stream. | |
| 768 * | |
| 769 * Complete the future returned by `StreamController.addStream` when | |
| 770 * the cancel is complete. | |
| 771 * | |
| 772 * Return a future if the cancel takes time, otherwise return `null`. | |
| 773 */ | |
| 774 Future cancel() { | |
| 775 var cancel = addSubscription.cancel(); | |
| 776 if (cancel == null) { | |
| 777 addStreamFuture._asyncComplete(null); | |
| 778 return null; | |
| 779 } | |
| 780 return cancel.whenComplete(() { addStreamFuture._asyncComplete(null); }); | |
| 781 } | |
| 782 | |
| 783 void complete() { | |
| 784 addStreamFuture._asyncComplete(null); | |
| 785 } | |
| 786 } | |
| 787 | |
| 788 class _StreamControllerAddStreamState<T> extends _AddStreamState<T> { | |
| 789 // The subscription or pending data of a _StreamController. | |
| 790 // Stored here because we reuse the `_varData` field in the _StreamController | |
| 791 // to store this state object. | |
| 792 var varData; | |
| 793 | |
| 794 _StreamControllerAddStreamState(_StreamController controller, | |
| 795 this.varData, | |
| 796 Stream source, | |
| 797 bool cancelOnError) | |
| 798 : super(controller, source, cancelOnError) { | |
| 799 if (controller.isPaused) { | |
| 800 addSubscription.pause(); | |
| 801 } | |
| 802 } | |
| 803 } | |
| 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