| // Copyright (c) 2012, the Dart project authors. Please see the AUTHORS file |
| // for details. All rights reserved. Use of this source code is governed by a |
| // BSD-style license that can be found in the LICENSE file. |
| |
| part of dart.async; |
| |
| // ------------------------------------------------------------------- |
| // Controller for creating and adding events to a stream. |
| // ------------------------------------------------------------------- |
| |
| /// Type of a stream controller's `onListen`, `onPause` and `onResume` callbacks. |
| typedef void ControllerCallback(); |
| |
| /// Type of stream controller `onCancel` callbacks. |
| typedef FutureOr<void> ControllerCancelCallback(); |
| |
| /// A controller with the stream it controls. |
| /// |
| /// This controller allows sending data, error and done events on |
| /// its [stream]. |
| /// This class can be used to create a simple stream that others |
| /// can listen on, and to push events to that stream. |
| /// |
| /// It's possible to check whether the stream is paused or not, and whether |
| /// it has subscribers or not, as well as getting a callback when either of |
| /// these change. |
| abstract class StreamController<T> implements StreamSink<T> { |
| /// The stream that this controller is controlling. |
| Stream<T> get stream; |
| |
| /// A controller with a [stream] that supports only one single subscriber. |
| /// |
| /// If [sync] is true, the returned stream controller is a |
| /// [SynchronousStreamController], and must be used with the care |
| /// and attention necessary to not break the [Stream] contract. If in doubt, |
| /// use the non-sync version. |
| /// |
| /// Using an asynchronous controller will never give the wrong |
| /// behavior, but using a synchronous controller incorrectly can cause |
| /// otherwise correct programs to break. |
| /// |
| /// A synchronous controller is only intended for optimizing event |
| /// propagation when one asynchronous event immediately triggers another. |
| /// It should not be used unless the calls to [add] or [addError] |
| /// are guaranteed to occur in places where it won't break `Stream` invariants. |
| /// |
| /// Use synchronous controllers only to forward (potentially transformed) |
| /// events from another stream or a future. |
| /// |
| /// A Stream should be inert until a subscriber starts listening on it (using |
| /// the [onListen] callback to start producing events). Streams should not |
| /// leak resources (like websockets) when no user ever listens on the stream. |
| /// |
| /// The controller buffers all incoming events until a subscriber is |
| /// registered, but this feature should only be used in rare circumstances. |
| /// |
| /// The [onPause] function is called when the stream becomes |
| /// paused. [onResume] is called when the stream resumed. |
| /// |
| /// The [onListen] callback is called when the stream |
| /// receives its listener and [onCancel] when the listener ends |
| /// its subscription. If [onCancel] needs to perform an asynchronous operation, |
| /// [onCancel] should return a future that completes when the cancel operation |
| /// is done. |
| /// |
| /// If the stream is canceled before the controller needs data the |
| /// [onResume] call might not be executed. |
| factory StreamController( |
| {void onListen()?, |
| void onPause()?, |
| void onResume()?, |
| FutureOr<void> onCancel()?, |
| bool sync = false}) { |
| return sync |
| ? _SyncStreamController<T>(onListen, onPause, onResume, onCancel) |
| : _AsyncStreamController<T>(onListen, onPause, onResume, onCancel); |
| } |
| |
| /// A controller where [stream] can be listened to more than once. |
| /// |
| /// The [Stream] returned by [stream] is a broadcast stream. |
| /// It can be listened to more than once. |
| /// |
| /// A Stream should be inert until a subscriber starts listening on it (using |
| /// the [onListen] callback to start producing events). Streams should not |
| /// leak resources (like websockets) when no user ever listens on the stream. |
| /// |
| /// Broadcast streams do not buffer events when there is no listener. |
| /// |
| /// The controller distributes any events to all currently subscribed |
| /// listeners at the time when [add], [addError] or [close] is called. |
| /// It is not allowed to call `add`, `addError`, or `close` before a previous |
| /// call has returned. The controller does not have any internal queue of |
| /// events, and if there are no listeners at the time the event is added, |
| /// it will just be dropped, or, if it is an error, be reported as uncaught. |
| /// |
| /// Each listener subscription is handled independently, |
| /// and if one pauses, only the pausing listener is affected. |
| /// A paused listener will buffer events internally until unpaused or canceled. |
| /// |
| /// If [sync] is true, events may be fired directly by the stream's |
| /// subscriptions during an [add], [addError] or [close] call. |
| /// The returned stream controller is a [SynchronousStreamController], |
| /// and must be used with the care and attention necessary to not break |
| /// the [Stream] contract. |
| /// See [Completer.sync] for some explanations on when a synchronous |
| /// dispatching can be used. |
| /// If in doubt, keep the controller non-sync. |
| /// |
| /// If [sync] is false, the event will always be fired at a later time, |
| /// after the code adding the event has completed. |
| /// In that case, no guarantees are given with regard to when |
| /// multiple listeners get the events, except that each listener will get |
| /// all events in the correct order. Each subscription handles the events |
| /// individually. |
| /// If two events are sent on an async controller with two listeners, |
| /// one of the listeners may get both events |
| /// before the other listener gets any. |
| /// A listener must be subscribed both when the event is initiated |
| /// (that is, when [add] is called) |
| /// and when the event is later delivered, |
| /// in order to receive the event. |
| /// |
| /// The [onListen] callback is called when the first listener is subscribed, |
| /// and the [onCancel] is called when there are no longer any active listeners. |
| /// If a listener is added again later, after the [onCancel] was called, |
| /// the [onListen] will be called again. |
| factory StreamController.broadcast( |
| {void onListen()?, void onCancel()?, bool sync = false}) { |
| return sync |
| ? _SyncBroadcastStreamController<T>(onListen, onCancel) |
| : _AsyncBroadcastStreamController<T>(onListen, onCancel); |
| } |
| |
| /// The callback which is called when the stream is listened to. |
| /// |
| /// May be set to `null`, in which case no callback will happen. |
| abstract void Function()? onListen; |
| |
| /// The callback which is called when the stream is paused. |
| /// |
| /// May be set to `null`, in which case no callback will happen. |
| /// |
| /// Pause related callbacks are not supported on broadcast stream controllers. |
| abstract void Function()? onPause; |
| |
| /// The callback which is called when the stream is resumed. |
| /// |
| /// May be set to `null`, in which case no callback will happen. |
| /// |
| /// Pause related callbacks are not supported on broadcast stream controllers. |
| abstract void Function()? onResume; |
| |
| /// The callback which is called when the stream is canceled. |
| /// |
| /// May be set to `null`, in which case no callback will happen. |
| abstract FutureOr<void> Function()? onCancel; |
| |
| /// Returns a view of this object that only exposes the [StreamSink] interface. |
| StreamSink<T> get sink; |
| |
| /// Whether the stream controller is closed for adding more events. |
| /// |
| /// The controller becomes closed by calling the [close] method. |
| /// New events cannot be added, by calling [add] or [addError], |
| /// to a closed controller. |
| /// |
| /// If the controller is closed, |
| /// the "done" event might not have been delivered yet, |
| /// but it has been scheduled, and it is too late to add more events. |
| bool get isClosed; |
| |
| /// Whether the subscription would need to buffer events. |
| /// |
| /// This is the case if the controller's stream has a listener and it is |
| /// paused, or if it has not received a listener yet. In that case, the |
| /// controller is considered paused as well. |
| /// |
| /// A broadcast stream controller is never considered paused. It always |
| /// forwards its events to all uncanceled subscriptions, if any, |
| /// and let the subscriptions handle their own pausing and buffering. |
| bool get isPaused; |
| |
| /// Whether there is a subscriber on the [Stream]. |
| bool get hasListener; |
| |
| /// Sends a data [event]. |
| /// |
| /// Listeners receive this event in a later microtask. |
| /// |
| /// Note that a synchronous controller (created by passing true to the `sync` |
| /// parameter of the `StreamController` constructor) delivers events |
| /// immediately. Since this behavior violates the contract mentioned here, |
| /// synchronous controllers should only be used as described in the |
| /// documentation to ensure that the delivered events always *appear* as if |
| /// they were delivered in a separate microtask. |
| void add(T event); |
| |
| /// Sends or enqueues an error event. |
| /// |
| /// Listeners receive this event at a later microtask. This behavior can be |
| /// overridden by using `sync` controllers. Note, however, that sync |
| /// controllers have to satisfy the preconditions mentioned in the |
| /// documentation of the constructors. |
| void addError(Object error, [StackTrace? stackTrace]); |
| |
| /// Closes the stream. |
| /// |
| /// No further events can be added to a closed stream. |
| /// |
| /// The returned future is the same future provided by [done]. |
| /// It is completed when the stream listeners is done sending events, |
| /// This happens either when the done event has been sent, |
| /// or when the subscriber on a single-subscription stream is canceled. |
| /// |
| /// A broadcast stream controller will send the done event |
| /// even if listeners are paused, so some broadcast events may not have been |
| /// received yet when the returned future completes. |
| /// |
| /// If no one listens to a non-broadcast stream, |
| /// or the listener pauses and never resumes, |
| /// the done event will not be sent and this future will never complete. |
| Future close(); |
| |
| /// A future which is completed when the stream controller is done |
| /// sending events. |
| /// |
| /// This happens either when the done event has been sent, or if the |
| /// subscriber on a single-subscription stream is canceled. |
| /// |
| /// A broadcast stream controller will send the done event |
| /// even if listeners are paused, so some broadcast events may not have been |
| /// received yet when the returned future completes. |
| /// |
| /// If there is no listener on a non-broadcast stream, |
| /// or the listener pauses and never resumes, |
| /// the done event will not be sent and this future will never complete. |
| Future get done; |
| |
| /// Receives events from [source] and puts them into this controller's stream. |
| /// |
| /// Returns a future which completes when the source stream is done. |
| /// |
| /// Events must not be added directly to this controller using [add], |
| /// [addError], [close] or [addStream], until the returned future |
| /// is complete. |
| /// |
| /// Data and error events are forwarded to this controller's stream. A done |
| /// event on the source will end the `addStream` operation and complete the |
| /// returned future. |
| /// |
| /// If [cancelOnError] is `true`, only the first error on [source] is |
| /// forwarded to the controller's stream, and the `addStream` ends |
| /// after this. If [cancelOnError] is false, all errors are forwarded |
| /// and only a done event will end the `addStream`. |
| /// If [cancelOnError] is omitted or `null`, it defaults to `false`. |
| Future addStream(Stream<T> source, {bool? cancelOnError}); |
| } |
| |
| /// A stream controller that delivers its events synchronously. |
| /// |
| /// A synchronous stream controller is intended for cases where |
| /// an already asynchronous event triggers an event on a stream. |
| /// |
| /// Instead of adding the event to the stream in a later microtask, |
| /// causing extra latency, the event is instead fired immediately by the |
| /// synchronous stream controller, as if the stream event was |
| /// the current event or microtask. |
| /// |
| /// The synchronous stream controller can be used to break the contract |
| /// on [Stream], and it must be used carefully to avoid doing so. |
| /// |
| /// The only advantage to using a [SynchronousStreamController] over a |
| /// normal [StreamController] is the improved latency. |
| /// Only use the synchronous version if the improvement is significant, |
| /// and if its use is safe. Otherwise just use a normal stream controller, |
| /// which will always have the correct behavior for a [Stream], and won't |
| /// accidentally break other code. |
| /// |
| /// Adding events to a synchronous controller should only happen as the |
| /// very last part of the handling of the original event. |
| /// At that point, adding an event to the stream is equivalent to |
| /// returning to the event loop and adding the event in the next microtask. |
| /// |
| /// Each listener callback will be run as if it was a top-level event |
| /// or microtask. This means that if it throws, the error will be reported as |
| /// uncaught as soon as possible. |
| /// This is one reason to add the event as the last thing in the original event |
| /// handler - any action done after adding the event will delay the report of |
| /// errors in the event listener callbacks. |
| /// |
| /// If an event is added in a setting that isn't known to be another event, |
| /// it may cause the stream's listener to get that event before the listener |
| /// is ready to handle it. We promise that after calling [Stream.listen], |
| /// you won't get any events until the code doing the listen has completed. |
| /// Calling [add] in response to a function call of unknown origin may break |
| /// that promise. |
| /// |
| /// An [onListen] callback from the controller is *not* an asynchronous event, |
| /// and adding events to the controller in the `onListen` callback is always |
| /// wrong. The events will be delivered before the listener has even received |
| /// the subscription yet. |
| /// |
| /// The synchronous broadcast stream controller also has a restrictions that a |
| /// normal stream controller does not: |
| /// The [add], [addError], [close] and [addStream] methods *must not* be |
| /// called while an event is being delivered. |
| /// That is, if a callback on a subscription on the controller's stream causes |
| /// a call to any of the functions above, the call will fail. |
| /// A broadcast stream may have more than one listener, and if an |
| /// event is added synchronously while another is being also in the process |
| /// of being added, the latter event might reach some listeners before |
| /// the former. To prevent that, an event cannot be added while a previous |
| /// event is being fired. |
| /// This guarantees that an event is fully delivered when the |
| /// first [add], [addError] or [close] returns, |
| /// and further events will be delivered in the correct order. |
| /// |
| /// This still only guarantees that the event is delivered to the subscription. |
| /// If the subscription is paused, the actual callback may still happen later, |
| /// and the event will instead be buffered by the subscription. |
| /// Barring pausing, and the following buffered events that haven't been |
| /// delivered yet, callbacks will be called synchronously when an event is added. |
| /// |
| /// Adding an event to a synchronous non-broadcast stream controller while |
| /// another event is in progress may cause the second event to be delayed |
| /// and not be delivered synchronously, and until that event is delivered, |
| /// the controller will not act synchronously. |
| abstract class SynchronousStreamController<T> implements StreamController<T> { |
| /// Adds event to the controller's stream. |
| /// |
| /// As [StreamController.add], but must not be called while an event is |
| /// being added by [add], [addError] or [close]. |
| void add(T data); |
| |
| /// Adds error to the controller's stream. |
| /// |
| /// As [StreamController.addError], but must not be called while an event is |
| /// being added by [add], [addError] or [close]. |
| void addError(Object error, [StackTrace? stackTrace]); |
| |
| /// Closes the controller's stream. |
| /// |
| /// As [StreamController.close], but must not be called while an event is |
| /// being added by [add], [addError] or [close]. |
| Future close(); |
| } |
| |
| abstract class _StreamControllerLifecycle<T> { |
| StreamSubscription<T> _subscribe(void onData(T data)?, Function? onError, |
| void onDone()?, bool cancelOnError); |
| void _recordPause(StreamSubscription<T> subscription) {} |
| void _recordResume(StreamSubscription<T> subscription) {} |
| Future<void>? _recordCancel(StreamSubscription<T> subscription) => null; |
| } |
| |
| // Base type for implementations of stream controllers. |
| abstract class _StreamControllerBase<T> |
| implements |
| StreamController<T>, |
| _StreamControllerLifecycle<T>, |
| _EventSink<T>, |
| _EventDispatch<T> {} |
| |
| /// Default implementation of [StreamController]. |
| /// |
| /// Controls a stream that only supports a single controller. |
| abstract class _StreamController<T> implements _StreamControllerBase<T> { |
| // The states are bit-flags. More than one can be set at a time. |
| // |
| // The "subscription state" goes through the states: |
| // initial -> subscribed -> canceled. |
| // These are mutually exclusive. |
| // The "closed" state records whether the [close] method has been called |
| // on the controller. This can be done at any time. If done before |
| // subscription, the done event is queued. If done after cancel, the done |
| // event is ignored (just as any other event after a cancel). |
| |
| /// The controller is in its initial state with no subscription. |
| static const int _STATE_INITIAL = 0; |
| |
| /// The controller has a subscription, but hasn't been closed or canceled. |
| /// |
| /// Keep in sync with |
| /// runtime/vm/stack_trace.cc:kStreamController_StateSubscribed. |
| static const int _STATE_SUBSCRIBED = 1; |
| |
| /// The subscription is canceled. |
| static const int _STATE_CANCELED = 2; |
| |
| /// Mask for the subscription state. |
| static const int _STATE_SUBSCRIPTION_MASK = 3; |
| |
| // The following state relate to the controller, not the subscription. |
| // If closed, adding more events is not allowed. |
| // If executing an [addStream], new events are not allowed either, but will |
| // be added by the stream. |
| |
| /// The controller is closed due to calling [close]. |
| /// |
| /// When the stream is closed, you can neither add new events nor add new |
| /// listeners. |
| static const int _STATE_CLOSED = 4; |
| |
| /// The controller is in the middle of an [addStream] operation. |
| /// |
| /// While adding events from a stream, no new events can be added directly |
| /// on the controller. |
| static const int _STATE_ADDSTREAM = 8; |
| |
| /// Field containing different data depending on the current subscription |
| /// state. |
| /// |
| /// If [_state] is [_STATE_INITIAL], the field may contain a [_PendingEvents] |
| /// for events added to the controller before a subscription. |
| /// |
| /// While [_state] is [_STATE_SUBSCRIBED], the field contains the subscription. |
| /// |
| /// When [_state] is [_STATE_CANCELED] the field is currently not used, |
| /// and will contain `null`. |
| @pragma("vm:entry-point") |
| Object? _varData; |
| |
| /// Current state of the controller. |
| @pragma("vm:entry-point") |
| int _state = _STATE_INITIAL; |
| |
| /// Future completed when the stream sends its last event. |
| /// |
| /// This is also the future returned by [close]. |
| // TODO(lrn): Could this be stored in the varData field too, if it's not |
| // accessed until the call to "close"? Then we need to special case if it's |
| // accessed earlier, or if close is called before subscribing. |
| _Future<void>? _doneFuture; |
| |
| void Function()? onListen; |
| void Function()? onPause; |
| void Function()? onResume; |
| FutureOr<void> Function()? onCancel; |
| |
| _StreamController(this.onListen, this.onPause, this.onResume, this.onCancel); |
| |
| // Return a new stream every time. The streams are equal, but not identical. |
| Stream<T> get stream => _ControllerStream<T>(this); |
| |
| /// Returns a view of this object that only exposes the [StreamSink] interface. |
| StreamSink<T> get sink => _StreamSinkWrapper<T>(this); |
| |
| /// Whether a listener has existed and been canceled. |
| /// |
| /// After this, adding more events will be ignored. |
| bool get _isCanceled => (_state & _STATE_CANCELED) != 0; |
| |
| /// Whether there is an active listener. |
| bool get hasListener => (_state & _STATE_SUBSCRIBED) != 0; |
| |
| /// Whether there has not been a listener yet. |
| bool get _isInitialState => |
| (_state & _STATE_SUBSCRIPTION_MASK) == _STATE_INITIAL; |
| |
| bool get isClosed => (_state & _STATE_CLOSED) != 0; |
| |
| bool get isPaused => |
| hasListener ? _subscription._isInputPaused : !_isCanceled; |
| |
| bool get _isAddingStream => (_state & _STATE_ADDSTREAM) != 0; |
| |
| /// New events may not be added after close, or during addStream. |
| bool get _mayAddEvent => (_state < _STATE_CLOSED); |
| |
| // Returns the pending events. |
| // Pending events are events added before a subscription exists. |
| // They are added to the subscription when it is created. |
| // Pending events, if any, are kept in the _varData field until the |
| // stream is listened to. |
| // While adding a stream, pending events are moved into the |
| // state object to allow the state object to use the _varData field. |
| _PendingEvents<T>? get _pendingEvents { |
| assert(_isInitialState); |
| if (!_isAddingStream) { |
| return _varData as dynamic; |
| } |
| _StreamControllerAddStreamState<T> state = _varData as dynamic; |
| return state.varData; |
| } |
| |
| // Returns the pending events, and creates the object if necessary. |
| _StreamImplEvents<T> _ensurePendingEvents() { |
| assert(_isInitialState); |
| if (!_isAddingStream) { |
| Object? events = _varData; |
| if (events == null) { |
| _varData = events = _StreamImplEvents<T>(); |
| } |
| return events as dynamic; |
| } |
| _StreamControllerAddStreamState<T> state = _varData as dynamic; |
| Object? events = state.varData; |
| if (events == null) { |
| state.varData = events = _StreamImplEvents<T>(); |
| } |
| return events as dynamic; |
| } |
| |
| // Get the current subscription. |
| // If we are adding a stream, the subscription is moved into the state |
| // object to allow the state object to use the _varData field. |
| _ControllerSubscription<T> get _subscription { |
| assert(hasListener); |
| Object? varData = _varData; |
| if (_isAddingStream) { |
| _StreamControllerAddStreamState<Object?> streamState = varData as dynamic; |
| varData = streamState.varData; |
| } |
| return varData as dynamic; |
| } |
| |
| /// Creates an error describing why an event cannot be added. |
| /// |
| /// The reason, and therefore the error message, depends on the current state. |
| Error _badEventState() { |
| if (isClosed) { |
| return StateError("Cannot add event after closing"); |
| } |
| assert(_isAddingStream); |
| return StateError("Cannot add event while adding a stream"); |
| } |
| |
| // StreamSink interface. |
| Future addStream(Stream<T> source, {bool? cancelOnError}) { |
| if (!_mayAddEvent) throw _badEventState(); |
| if (_isCanceled) return _Future.immediate(null); |
| _StreamControllerAddStreamState<T> addState = |
| _StreamControllerAddStreamState<T>( |
| this, _varData, source, cancelOnError ?? false); |
| _varData = addState; |
| _state |= _STATE_ADDSTREAM; |
| return addState.addStreamFuture; |
| } |
| |
| /// Returns a future that is completed when the stream is done |
| /// processing events. |
| /// |
| /// This happens either when the done event has been sent, or if the |
| /// subscriber of a single-subscription stream is cancelled. |
| Future<void> get done => _ensureDoneFuture(); |
| |
| Future<void> _ensureDoneFuture() => |
| _doneFuture ??= _isCanceled ? Future._nullFuture : _Future<void>(); |
| |
| /// Send or enqueue a data event. |
| void add(T value) { |
| if (!_mayAddEvent) throw _badEventState(); |
| _add(value); |
| } |
| |
| /// Send or enqueue an error event. |
| void addError(Object error, [StackTrace? stackTrace]) { |
| checkNotNullable(error, "error"); |
| if (!_mayAddEvent) throw _badEventState(); |
| AsyncError? replacement = Zone.current.errorCallback(error, stackTrace); |
| if (replacement != null) { |
| error = replacement.error; |
| stackTrace = replacement.stackTrace; |
| } else { |
| stackTrace ??= AsyncError.defaultStackTrace(error); |
| } |
| if (stackTrace == null) throw "unreachable"; // TODO(40088) |
| _addError(error, stackTrace); |
| } |
| |
| /// Closes this controller and sends a done event on the stream. |
| /// |
| /// The first time a controller is closed, a "done" event is added to its |
| /// stream. |
| /// |
| /// You are allowed to close the controller more than once, but only the first |
| /// call has any effect. |
| /// |
| /// After closing, no further events may be added using [add], [addError] |
| /// or [addStream]. |
| /// |
| /// The returned future is completed when the done event has been delivered. |
| Future close() { |
| if (isClosed) { |
| return _ensureDoneFuture(); |
| } |
| if (!_mayAddEvent) throw _badEventState(); |
| _closeUnchecked(); |
| return _ensureDoneFuture(); |
| } |
| |
| void _closeUnchecked() { |
| _state |= _STATE_CLOSED; |
| if (hasListener) { |
| _sendDone(); |
| } else if (_isInitialState) { |
| _ensurePendingEvents().add(const _DelayedDone()); |
| } |
| } |
| |
| // EventSink interface. Used by the [addStream] events. |
| |
| // Add data event, used both by the [addStream] events and by [add]. |
| void _add(T value) { |
| if (hasListener) { |
| _sendData(value); |
| } else if (_isInitialState) { |
| _ensurePendingEvents().add(_DelayedData<T>(value)); |
| } |
| } |
| |
| void _addError(Object error, StackTrace stackTrace) { |
| if (hasListener) { |
| _sendError(error, stackTrace); |
| } else if (_isInitialState) { |
| _ensurePendingEvents().add(_DelayedError(error, stackTrace)); |
| } |
| } |
| |
| void _close() { |
| // End of addStream stream. |
| assert(_isAddingStream); |
| _StreamControllerAddStreamState<T> addState = _varData as dynamic; |
| _varData = addState.varData; |
| _state &= ~_STATE_ADDSTREAM; |
| addState.complete(); |
| } |
| |
| // _StreamControllerLifeCycle interface |
| |
| StreamSubscription<T> _subscribe(void onData(T data)?, Function? onError, |
| void onDone()?, bool cancelOnError) { |
| if (!_isInitialState) { |
| throw StateError("Stream has already been listened to."); |
| } |
| _ControllerSubscription<T> subscription = _ControllerSubscription<T>( |
| this, onData, onError, onDone, cancelOnError); |
| |
| _PendingEvents<T>? pendingEvents = _pendingEvents; |
| _state |= _STATE_SUBSCRIBED; |
| if (_isAddingStream) { |
| _StreamControllerAddStreamState<T> addState = _varData as dynamic; |
| addState.varData = subscription; |
| addState.resume(); |
| } else { |
| _varData = subscription; |
| } |
| subscription._setPendingEvents(pendingEvents); |
| subscription._guardCallback(() { |
| _runGuarded(onListen); |
| }); |
| |
| return subscription; |
| } |
| |
| Future<void>? _recordCancel(StreamSubscription<T> subscription) { |
| // When we cancel, we first cancel any stream being added, |
| // Then we call `onCancel`, and finally the _doneFuture is completed. |
| // If either of addStream's cancel or `onCancel` returns a future, |
| // we wait for it before continuing. |
| // Any error during this process ends up in the returned future. |
| // If more errors happen, we act as if it happens inside nested try/finallys |
| // or whenComplete calls, and only the last error ends up in the |
| // returned future. |
| Future<void>? result; |
| if (_isAddingStream) { |
| _StreamControllerAddStreamState<T> addState = _varData as dynamic; |
| result = addState.cancel(); |
| } |
| _varData = null; |
| _state = |
| (_state & ~(_STATE_SUBSCRIBED | _STATE_ADDSTREAM)) | _STATE_CANCELED; |
| |
| var onCancel = this.onCancel; |
| if (onCancel != null) { |
| if (result == null) { |
| // Only introduce a future if one is needed. |
| // If _onCancel returns null, no future is needed. |
| try { |
| var cancelResult = onCancel(); |
| if (cancelResult is Future<void>) { |
| result = cancelResult; |
| } |
| } catch (e, s) { |
| // Return the error in the returned future. |
| // Complete it asynchronously, so there is time for a listener |
| // to handle the error. |
| result = _Future().._asyncCompleteError(e, s); |
| } |
| } else { |
| // Simpler case when we already know that we will return a future. |
| result = result.whenComplete(onCancel); |
| } |
| } |
| |
| void complete() { |
| var doneFuture = _doneFuture; |
| if (doneFuture != null && doneFuture._mayComplete) { |
| doneFuture._asyncComplete(null); |
| } |
| } |
| |
| if (result != null) { |
| result = result.whenComplete(complete); |
| } else { |
| complete(); |
| } |
| |
| return result; |
| } |
| |
| void _recordPause(StreamSubscription<T> subscription) { |
| if (_isAddingStream) { |
| _StreamControllerAddStreamState<T> addState = _varData as dynamic; |
| addState.pause(); |
| } |
| _runGuarded(onPause); |
| } |
| |
| void _recordResume(StreamSubscription<T> subscription) { |
| if (_isAddingStream) { |
| _StreamControllerAddStreamState<T> addState = _varData as dynamic; |
| addState.resume(); |
| } |
| _runGuarded(onResume); |
| } |
| } |
| |
| abstract class _SyncStreamControllerDispatch<T> |
| implements _StreamController<T>, SynchronousStreamController<T> { |
| void _sendData(T data) { |
| _subscription._add(data); |
| } |
| |
| void _sendError(Object error, StackTrace stackTrace) { |
| _subscription._addError(error, stackTrace); |
| } |
| |
| void _sendDone() { |
| _subscription._close(); |
| } |
| } |
| |
| abstract class _AsyncStreamControllerDispatch<T> |
| implements _StreamController<T> { |
| void _sendData(T data) { |
| _subscription._addPending(_DelayedData<T>(data)); |
| } |
| |
| void _sendError(Object error, StackTrace stackTrace) { |
| _subscription._addPending(_DelayedError(error, stackTrace)); |
| } |
| |
| void _sendDone() { |
| _subscription._addPending(const _DelayedDone()); |
| } |
| } |
| |
| // TODO(lrn): Use common superclass for callback-controllers when VM supports |
| // constructors in mixin superclasses. |
| |
| @pragma("vm:entry-point") |
| class _AsyncStreamController<T> = _StreamController<T> |
| with _AsyncStreamControllerDispatch<T>; |
| |
| class _SyncStreamController<T> = _StreamController<T> |
| with _SyncStreamControllerDispatch<T>; |
| |
| void _runGuarded(void Function()? notificationHandler) { |
| if (notificationHandler == null) return; |
| try { |
| notificationHandler(); |
| } catch (e, s) { |
| Zone.current.handleUncaughtError(e, s); |
| } |
| } |
| |
| class _ControllerStream<T> extends _StreamImpl<T> { |
| _StreamControllerLifecycle<T> _controller; |
| |
| _ControllerStream(this._controller); |
| |
| StreamSubscription<T> _createSubscription(void onData(T data)?, |
| Function? onError, void onDone()?, bool cancelOnError) => |
| _controller._subscribe(onData, onError, onDone, cancelOnError); |
| |
| // Override == and hashCode so that new streams returned by the same |
| // controller are considered equal. The controller returns a new stream |
| // each time it's queried, but doesn't have to cache the result. |
| |
| int get hashCode => _controller.hashCode ^ 0x35323532; |
| |
| bool operator ==(Object other) { |
| if (identical(this, other)) return true; |
| return other is _ControllerStream && |
| identical(other._controller, this._controller); |
| } |
| } |
| |
| class _ControllerSubscription<T> extends _BufferingStreamSubscription<T> { |
| final _StreamControllerLifecycle<T> _controller; |
| |
| _ControllerSubscription(this._controller, void onData(T data)?, |
| Function? onError, void onDone()?, bool cancelOnError) |
| : super(onData, onError, onDone, cancelOnError); |
| |
| Future<void>? _onCancel() { |
| return _controller._recordCancel(this); |
| } |
| |
| void _onPause() { |
| _controller._recordPause(this); |
| } |
| |
| void _onResume() { |
| _controller._recordResume(this); |
| } |
| } |
| |
| /// A class that exposes only the [StreamSink] interface of an object. |
| class _StreamSinkWrapper<T> implements StreamSink<T> { |
| final StreamController _target; |
| _StreamSinkWrapper(this._target); |
| void add(T data) { |
| _target.add(data); |
| } |
| |
| void addError(Object error, [StackTrace? stackTrace]) { |
| _target.addError(error, stackTrace); |
| } |
| |
| Future close() => _target.close(); |
| |
| Future addStream(Stream<T> source) => _target.addStream(source); |
| |
| Future get done => _target.done; |
| } |
| |
| /// Object containing the state used to handle [StreamController.addStream]. |
| class _AddStreamState<T> { |
| // [_Future] returned by call to addStream. |
| final _Future addStreamFuture; |
| |
| // Subscription on stream argument to addStream. |
| final StreamSubscription addSubscription; |
| |
| _AddStreamState( |
| _EventSink<T> controller, Stream<T> source, bool cancelOnError) |
| : addStreamFuture = _Future(), |
| addSubscription = source.listen(controller._add, |
| onError: cancelOnError |
| ? makeErrorHandler(controller) |
| : controller._addError, |
| onDone: controller._close, |
| cancelOnError: cancelOnError); |
| |
| static makeErrorHandler(_EventSink controller) => (Object e, StackTrace s) { |
| controller._addError(e, s); |
| controller._close(); |
| }; |
| |
| void pause() { |
| addSubscription.pause(); |
| } |
| |
| void resume() { |
| addSubscription.resume(); |
| } |
| |
| /// Stop adding the stream. |
| /// |
| /// Complete the future returned by `StreamController.addStream` when |
| /// the cancel is complete. |
| /// |
| /// Return a future if the cancel takes time, otherwise return `null`. |
| Future<void> cancel() { |
| var cancel = addSubscription.cancel(); |
| if (cancel == null) { |
| addStreamFuture._asyncComplete(null); |
| return Future._nullFuture; |
| } |
| return cancel.whenComplete(() { |
| addStreamFuture._asyncComplete(null); |
| }); |
| } |
| |
| void complete() { |
| addStreamFuture._asyncComplete(null); |
| } |
| } |
| |
| class _StreamControllerAddStreamState<T> extends _AddStreamState<T> { |
| // The subscription or pending data of a _StreamController. |
| // Stored here because we reuse the `_varData` field in the _StreamController |
| // to store this state object. |
| var varData; |
| |
| _StreamControllerAddStreamState(_StreamController<T> controller, this.varData, |
| Stream<T> source, bool cancelOnError) |
| : super(controller, source, cancelOnError) { |
| if (controller.isPaused) { |
| addSubscription.pause(); |
| } |
| } |
| } |