lib/src/stream_completer.dart

// Copyright (c) 2015, 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.

import 'dart:async';

/// A single-subscription [stream] where the contents are provided later.
///
/// It is generally recommended that you never create a `Future<Stream>`
/// because you can just directly create a stream that doesn't do anything
/// until it's ready to do so.
/// This class can be used to create such a stream.
///
/// The [stream] is a normal stream that you can listen to immediately,
/// but until either [setSourceStream] or [setEmpty] is called,
/// the stream won't produce any events.
///
/// The same effect can be achieved by using a [StreamController]
/// and adding the stream using `addStream` when both
/// the controller's stream is listened to and the source stream is ready.
/// This class attempts to shortcut some of the overhead when possible.
/// For example, if the [stream] is only listened to
/// after the source stream has been set,
/// the listen is performed directly on the source stream.
class StreamCompleter<T> {
  /// The stream doing the actual work, is returned by [stream].
  final _stream = _CompleterStream<T>();

  /// Convert a `Future<Stream>` to a `Stream`.
  ///
  /// This creates a stream using a stream completer,
  /// and sets the source stream to the result of the future when the
  /// future completes.
  ///
  /// If the future completes with an error, the returned stream will
  /// instead contain just that error.
  static Stream<T> fromFuture<T>(Future<Stream<T>> streamFuture) {
    var completer = StreamCompleter<T>();
    streamFuture.then(completer.setSourceStream, onError: completer.setError);
    return completer.stream;
  }

  /// The stream of this completer.
  ///
  /// This stream is always a single-subscription stream.
  ///
  /// When a source stream is provided, its events will be forwarded to
  /// listeners on this stream.
  ///
  /// The stream can be listened either before or after a source stream
  /// is set.
  Stream<T> get stream => _stream;

  /// Set a stream as the source of events for the [StreamCompleter]'s
  /// [stream].
  ///
  /// The completer's `stream` will act exactly as [sourceStream].
  ///
  /// If the source stream is set before [stream] is listened to,
  /// the listen call on [stream] is forwarded directly to [sourceStream].
  ///
  /// If [stream] is listened to before setting the source stream,
  /// an intermediate subscription is created. It looks like a completely
  /// normal subscription, and can be paused or canceled, but it won't
  /// produce any events until a source stream is provided.
  ///
  /// If the `stream` subscription is canceled before a source stream is set,
  /// the source stream will be listened to and immediately canceled again.
  ///
  /// Otherwise, when the source stream is then set,
  /// it is immediately listened to, and its events are forwarded to the
  /// existing subscription.
  ///
  /// Any one of [setSourceStream], [setEmpty], and [setError] may be called at
  /// most once. Trying to call any of them again will fail.
  void setSourceStream(Stream<T> sourceStream) {
    if (_stream._isSourceStreamSet) {
      throw StateError('Source stream already set');
    }
    _stream._setSourceStream(sourceStream);
  }

  /// Equivalent to setting an empty stream using [setSourceStream].
  ///
  /// Any one of [setSourceStream], [setEmpty], and [setError] may be called at
  /// most once. Trying to call any of them again will fail.
  void setEmpty() {
    if (_stream._isSourceStreamSet) {
      throw StateError('Source stream already set');
    }
    _stream._setEmpty();
  }

  /// Completes this to a stream that emits [error] and then closes.
  ///
  /// This is useful when the process of creating the data for the stream fails.
  ///
  /// Any one of [setSourceStream], [setEmpty], and [setError] may be called at
  /// most once. Trying to call any of them again will fail.
  void setError(Object error, [StackTrace? stackTrace]) {
    setSourceStream(Stream.fromFuture(Future.error(error, stackTrace)));
  }
}

/// Stream completed by [StreamCompleter].
class _CompleterStream<T> extends Stream<T> {
  /// Controller for an intermediate stream.
  ///
  /// Created if the user listens on this stream before the source stream
  /// is set, or if using [_setEmpty] so there is no source stream.
  StreamController<T>? _controller;

  /// Source stream for the events provided by this stream.
  ///
  /// Set when the completer sets the source stream using [_setSourceStream]
  /// or [_setEmpty].
  Stream<T>? _sourceStream;

  @override
  StreamSubscription<T> listen(void Function(T)? onData,
      {Function? onError, void Function()? onDone, bool? cancelOnError}) {
    if (_controller == null) {
      var sourceStream = _sourceStream;
      if (sourceStream != null && !sourceStream.isBroadcast) {
        // If the source stream is itself single subscription,
        // just listen to it directly instead of creating a controller.
        return sourceStream.listen(onData,
            onError: onError, onDone: onDone, cancelOnError: cancelOnError);
      }
      _ensureController();
      if (_sourceStream != null) {
        _linkStreamToController();
      }
    }
    return _controller!.stream.listen(onData,
        onError: onError, onDone: onDone, cancelOnError: cancelOnError);
  }

  /// Whether a source stream has been set.
  ///
  /// Used to throw an error if trying to set a source stream twice.
  bool get _isSourceStreamSet => _sourceStream != null;

  /// Sets the source stream providing the events for this stream.
  ///
  /// If set before the user listens, listen calls will be directed directly
  /// to the source stream. If the user listenes earlier, and intermediate
  /// stream is created using a stream controller, and the source stream is
  /// linked into that stream later.
  void _setSourceStream(Stream<T> sourceStream) {
    assert(_sourceStream == null);
    _sourceStream = sourceStream;
    if (_controller != null) {
      // User has already listened, so provide the data through controller.
      _linkStreamToController();
    }
  }

  /// Links source stream to controller when both are available.
  void _linkStreamToController() {
    var controller = _controller!;
    controller
        .addStream(_sourceStream!, cancelOnError: false)
        .whenComplete(controller.close);
  }

  /// Sets an empty source stream.
  ///
  /// Uses [_controller] for the stream, then closes the controller
  /// immediately.
  void _setEmpty() {
    assert(_sourceStream == null);
    var controller = _ensureController();
    _sourceStream = controller.stream; // Mark stream as set.
    controller.close();
  }

  // Creates the [_controller].
  StreamController<T> _ensureController() {
    return _controller ??= StreamController<T>(sync: true);
  }
}