| Index: packages/async/lib/src/lazy_stream.dart
|
| diff --git a/packages/async/lib/src/lazy_stream.dart b/packages/async/lib/src/lazy_stream.dart
|
| new file mode 100644
|
| index 0000000000000000000000000000000000000000..f07c387e59f60bc37535624f264c6112c535b96e
|
| --- /dev/null
|
| +++ b/packages/async/lib/src/lazy_stream.dart
|
| @@ -0,0 +1,52 @@
|
| +// 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";
|
| +
|
| +import "delegate/stream.dart";
|
| +import "stream_completer.dart";
|
| +import "utils.dart";
|
| +
|
| +/// A [Stream] wrapper that forwards to another [Stream] that's initialized
|
| +/// lazily.
|
| +///
|
| +/// This class allows a concrete `Stream` to be created only once it has a
|
| +/// listener. It's useful to wrapping APIs that do expensive computation to
|
| +/// produce a `Stream`.
|
| +class LazyStream<T> extends Stream<T> {
|
| + /// The callback that's called to create the inner stream.
|
| + FutureOrCallback<Stream<T>> _callback;
|
| +
|
| + /// Creates a single-subscription `Stream` that calls [callback] when it gets
|
| + /// a listener and forwards to the returned stream.
|
| + LazyStream(FutureOr<Stream<T>> callback()) : _callback = callback {
|
| + // Explicitly check for null because we null out [_callback] internally.
|
| + if (_callback == null) throw new ArgumentError.notNull('callback');
|
| + }
|
| +
|
| + StreamSubscription<T> listen(void onData(T event),
|
| + {Function onError, void onDone(), bool cancelOnError}) {
|
| + if (_callback == null) {
|
| + throw new StateError("Stream has already been listened to.");
|
| + }
|
| +
|
| + // Null out the callback before we invoke it to ensure that even while
|
| + // running it, this can't be called twice.
|
| + var callback = _callback;
|
| + _callback = null;
|
| + var result = callback();
|
| +
|
| + Stream<T> stream;
|
| + if (result is Future<Stream<T>>) {
|
| + stream = StreamCompleter.fromFuture(result.then((stream) {
|
| + return DelegatingStream.typed<T>(stream);
|
| + }));
|
| + } else {
|
| + stream = DelegatingStream.typed<T>(result as Stream);
|
| + }
|
| +
|
| + return stream.listen(onData,
|
| + onError: onError, onDone: onDone, cancelOnError: cancelOnError);
|
| + }
|
| +}
|
|
|