OLD | NEW |
1 // Copyright (c) 2015, the Dart project authors. Please see the AUTHORS file | 1 // Copyright (c) 2015, the Dart project authors. Please see the AUTHORS file |
2 // for details. All rights reserved. Use of this source code is governed by a | 2 // for details. All rights reserved. Use of this source code is governed by a |
3 // BSD-style license that can be found in the LICENSE file. | 3 // BSD-style license that can be found in the LICENSE file. |
4 | 4 |
5 import "dart:async"; | 5 import "dart:async"; |
6 | 6 |
7 import "stream_completer.dart"; | 7 import "stream_completer.dart"; |
8 import "delegate/stream.dart"; | 8 import "delegate/stream.dart"; |
9 | 9 |
10 /// A [Stream] wrapper that forwards to another [Stream] that's initialized | 10 /// A [Stream] wrapper that forwards to another [Stream] that's initialized |
11 /// lazily. | 11 /// lazily. |
12 /// | 12 /// |
13 /// This class allows a concrete `Stream` to be created only once it has a | 13 /// This class allows a concrete `Stream` to be created only once it has a |
14 /// listener. It's useful to wrapping APIs that do expensive computation to | 14 /// listener. It's useful to wrapping APIs that do expensive computation to |
15 /// produce a `Stream`. | 15 /// produce a `Stream`. |
16 class LazyStream<T> extends Stream<T> { | 16 class LazyStream<T> extends Stream<T> { |
17 /// The callback that's called to create the inner stream. | 17 /// The callback that's called to create the inner stream. |
18 ZoneCallback _callback; | 18 ZoneCallback _callback; |
19 | 19 |
20 /// Creates a single-subscription `Stream` that calls [callback] when it gets | 20 /// Creates a single-subscription `Stream` that calls [callback] when it gets |
21 /// a listener and forwards to the returned stream. | 21 /// a listener and forwards to the returned stream. |
22 /// | 22 /// |
23 /// The [callback] may return a `Stream` or a `Future<Stream>`. | 23 /// The [callback] may return a `Stream` or a `Future<Stream>`. |
24 LazyStream(callback()) : _callback = callback { | 24 LazyStream(callback()) : _callback = callback { |
25 // Explicitly check for null because we null out [_callback] internally. | 25 // Explicitly check for null because we null out [_callback] internally. |
26 if (_callback == null) throw new ArgumentError.notNull('callback'); | 26 if (_callback == null) throw new ArgumentError.notNull('callback'); |
27 } | 27 } |
28 | 28 |
29 StreamSubscription<T> listen(void onData(T event), | 29 StreamSubscription<T> listen(void onData(T event), |
30 {Function onError, | 30 {Function onError, void onDone(), bool cancelOnError}) { |
31 void onDone(), | |
32 bool cancelOnError}) { | |
33 if (_callback == null) { | 31 if (_callback == null) { |
34 throw new StateError("Stream has already been listened to."); | 32 throw new StateError("Stream has already been listened to."); |
35 } | 33 } |
36 | 34 |
37 // Null out the callback before we invoke it to ensure that even while | 35 // Null out the callback before we invoke it to ensure that even while |
38 // running it, this can't be called twice. | 36 // running it, this can't be called twice. |
39 var callback = _callback; | 37 var callback = _callback; |
40 _callback = null; | 38 _callback = null; |
41 var result = callback(); | 39 var result = callback(); |
42 | 40 |
43 Stream<T> stream; | 41 Stream<T> stream; |
44 if (result is Future) { | 42 if (result is Future) { |
45 stream = StreamCompleter.fromFuture(result.then((stream) { | 43 stream = StreamCompleter.fromFuture(result.then((stream) { |
46 return DelegatingStream.typed/*<T>*/(stream as Stream); | 44 return DelegatingStream.typed<T>(stream as Stream); |
47 })); | 45 })); |
48 } else { | 46 } else { |
49 stream = DelegatingStream.typed/*<T>*/(result as Stream); | 47 stream = DelegatingStream.typed<T>(result as Stream); |
50 } | 48 } |
51 | 49 |
52 return stream.listen(onData, | 50 return stream.listen(onData, |
53 onError: onError, onDone: onDone, cancelOnError: cancelOnError); | 51 onError: onError, onDone: onDone, cancelOnError: cancelOnError); |
54 } | 52 } |
55 } | 53 } |
OLD | NEW |