| Index: packages/async/lib/src/stream_sink_completer.dart
|
| diff --git a/packages/async/lib/src/stream_sink_completer.dart b/packages/async/lib/src/stream_sink_completer.dart
|
| new file mode 100644
|
| index 0000000000000000000000000000000000000000..4219126817583657297af3d2af2cbce681385c16
|
| --- /dev/null
|
| +++ b/packages/async/lib/src/stream_sink_completer.dart
|
| @@ -0,0 +1,177 @@
|
| +// Copyright (c) 2016, 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';
|
| +
|
| +import 'null_stream_sink.dart';
|
| +
|
| +/// A [sink] where the destination is provided later.
|
| +///
|
| +/// The [sink] is a normal sink that you can add events to to immediately, but
|
| +/// until [setDestinationSink] is called, the events will be buffered.
|
| +///
|
| +/// The same effect can be achieved by using a [StreamController] and adding it
|
| +/// to the sink using [Sink.addStream] when the destination sink is ready. This
|
| +/// class attempts to shortcut some of the overhead when possible. For example,
|
| +/// if the [sink] only has events added after the destination sink has been set,
|
| +/// those events are added directly to the sink.
|
| +class StreamSinkCompleter<T> {
|
| + /// The sink for this completer.
|
| + ///
|
| + /// When a destination sink is provided, events that have been passed to the
|
| + /// sink will be forwarded to the destination.
|
| + ///
|
| + /// Events can be added to the sink either before or after a destination sink
|
| + /// is set.
|
| + final StreamSink<T> sink = new _CompleterSink<T>();
|
| +
|
| + /// Returns [sink] typed as a [_CompleterSink].
|
| + _CompleterSink<T> get _sink => sink;
|
| +
|
| + /// Convert a `Future<StreamSink>` to a `StreamSink`.
|
| + ///
|
| + /// This creates a sink using a sink completer, and sets the destination sink
|
| + /// to the result of the future when the future completes.
|
| + ///
|
| + /// If the future completes with an error, the returned sink will instead
|
| + /// be closed. Its [Sink.done] future will contain the error.
|
| + static StreamSink<T> fromFuture<T>(Future<StreamSink<T>> sinkFuture) {
|
| + var completer = new StreamSinkCompleter<T>();
|
| + sinkFuture.then(completer.setDestinationSink, onError: completer.setError);
|
| + return completer.sink;
|
| + }
|
| +
|
| + /// Sets a sink as the destination for events from the [StreamSinkCompleter]'s
|
| + /// [sink].
|
| + ///
|
| + /// The completer's [sink] will act exactly as [destinationSink].
|
| + ///
|
| + /// If the destination sink is set before events are added to [sink], further
|
| + /// events are forwarded directly to [destinationSink].
|
| + ///
|
| + /// If events are added to [sink] before setting the destination sink, they're
|
| + /// buffered until the destination is available.
|
| + ///
|
| + /// A destination sink may be set at most once.
|
| + ///
|
| + /// Either of [setDestinationSink] or [setError] may be called at most once.
|
| + /// Trying to call either of them again will fail.
|
| + void setDestinationSink(StreamSink<T> destinationSink) {
|
| + if (_sink._destinationSink != null) {
|
| + throw new StateError("Destination sink already set");
|
| + }
|
| + _sink._setDestinationSink(destinationSink);
|
| + }
|
| +
|
| + /// Completes this to a closed sink whose [Sink.done] future emits [error].
|
| + ///
|
| + /// This is useful when the process of loading the sink fails.
|
| + ///
|
| + /// Either of [setDestinationSink] or [setError] may be called at most once.
|
| + /// Trying to call either of them again will fail.
|
| + void setError(error, [StackTrace stackTrace]) {
|
| + setDestinationSink(new NullStreamSink.error(error, stackTrace));
|
| + }
|
| +}
|
| +
|
| +/// [StreamSink] completed by [StreamSinkCompleter].
|
| +class _CompleterSink<T> implements StreamSink<T> {
|
| + /// Controller for an intermediate sink.
|
| + ///
|
| + /// Created if the user adds events to this sink before the destination sink
|
| + /// is set.
|
| + StreamController<T> _controller;
|
| +
|
| + /// Completer for [done].
|
| + ///
|
| + /// Created if the user requests the [done] future before the destination sink
|
| + /// is set.
|
| + Completer _doneCompleter;
|
| +
|
| + /// Destination sink for the events added to this sink.
|
| + ///
|
| + /// Set when [StreamSinkCompleter.setDestinationSink] is called.
|
| + StreamSink<T> _destinationSink;
|
| +
|
| + /// Whether events should be sent directly to [_destinationSink], as opposed
|
| + /// to going through [_controller].
|
| + bool get _canSendDirectly => _controller == null && _destinationSink != null;
|
| +
|
| + Future get done {
|
| + if (_doneCompleter != null) return _doneCompleter.future;
|
| + if (_destinationSink == null) {
|
| + _doneCompleter = new Completer.sync();
|
| + return _doneCompleter.future;
|
| + }
|
| + return _destinationSink.done;
|
| + }
|
| +
|
| + void add(T event) {
|
| + if (_canSendDirectly) {
|
| + _destinationSink.add(event);
|
| + } else {
|
| + _ensureController();
|
| + _controller.add(event);
|
| + }
|
| + }
|
| +
|
| + void addError(error, [StackTrace stackTrace]) {
|
| + if (_canSendDirectly) {
|
| + _destinationSink.addError(error, stackTrace);
|
| + } else {
|
| + _ensureController();
|
| + _controller.addError(error, stackTrace);
|
| + }
|
| + }
|
| +
|
| + Future addStream(Stream<T> stream) {
|
| + if (_canSendDirectly) return _destinationSink.addStream(stream);
|
| +
|
| + _ensureController();
|
| + return _controller.addStream(stream, cancelOnError: false);
|
| + }
|
| +
|
| + Future close() {
|
| + if (_canSendDirectly) {
|
| + _destinationSink.close();
|
| + } else {
|
| + _ensureController();
|
| + _controller.close();
|
| + }
|
| + return done;
|
| + }
|
| +
|
| + /// Create [_controller] if it doesn't yet exist.
|
| + void _ensureController() {
|
| + if (_controller == null) _controller = new StreamController(sync: true);
|
| + }
|
| +
|
| + /// Sets the destination sink to which events from this sink will be provided.
|
| + ///
|
| + /// If set before the user adds events, events will be added directly to the
|
| + /// destination sink. If the user adds events earlier, an intermediate sink is
|
| + /// created using a stream controller, and the destination sink is linked to
|
| + /// it later.
|
| + void _setDestinationSink(StreamSink<T> sink) {
|
| + assert(_destinationSink == null);
|
| + _destinationSink = sink;
|
| +
|
| + // If the user has already added data, it's buffered in the controller, so
|
| + // we add it to the sink.
|
| + if (_controller != null) {
|
| + // Catch any error that may come from [addStream] or [sink.close]. They'll
|
| + // be reported through [done] anyway.
|
| + sink
|
| + .addStream(_controller.stream)
|
| + .whenComplete(sink.close)
|
| + .catchError((_) {});
|
| + }
|
| +
|
| + // If the user has already asked when the sink is done, connect the sink's
|
| + // done callback to that completer.
|
| + if (_doneCompleter != null) {
|
| + _doneCompleter.complete(sink.done);
|
| + }
|
| + }
|
| +}
|
|
|