blob: 00a7086cf978cba631c2f46fdb3dfaabde98eaf7 [file] [log] [blame]
// 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);
}
}
}