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 "delegate/stream_subscription.dart"; | 7 import "delegate/stream_subscription.dart"; |
8 | 8 |
9 /// A [Stream] adapter for a [StreamSubscription]. | 9 /// A [Stream] adapter for a [StreamSubscription]. |
10 /// | 10 /// |
11 /// This class allows a `StreamSubscription` to be treated as a `Stream`. | 11 /// This class allows a `StreamSubscription` to be treated as a `Stream`. |
12 /// | 12 /// |
13 /// The subscription is paused until the stream is listened to, | 13 /// The subscription is paused until the stream is listened to, |
14 /// then it is resumed and the events are passed on to the | 14 /// then it is resumed and the events are passed on to the |
15 /// stream's new subscription. | 15 /// stream's new subscription. |
16 /// | 16 /// |
17 /// This class assumes that is has control over the original subscription. | 17 /// This class assumes that is has control over the original subscription. |
18 /// If other code is accessing the subscription, results may be unpredictable. | 18 /// If other code is accessing the subscription, results may be unpredictable. |
19 class SubscriptionStream<T> extends Stream<T> { | 19 class SubscriptionStream<T> extends Stream<T> { |
20 /// The subscription providing the events for this stream. | 20 /// The subscription providing the events for this stream. |
21 StreamSubscription _source; | 21 StreamSubscription<T> _source; |
22 | 22 |
23 /// Create a single-subscription `Stream` from [subscription]. | 23 /// Create a single-subscription `Stream` from [subscription]. |
24 /// | 24 /// |
25 /// The `subscription` should not be paused. This class will not resume prior | 25 /// The `subscription` should not be paused. This class will not resume prior |
26 /// pauses, so being paused is indistinguishable from not providing any | 26 /// pauses, so being paused is indistinguishable from not providing any |
27 /// events. | 27 /// events. |
28 /// | 28 /// |
29 /// If the `subscription` doesn't send any `done` events, neither will this | 29 /// If the `subscription` doesn't send any `done` events, neither will this |
30 /// stream. That may be an issue if `subscription` was made to cancel on | 30 /// stream. That may be an issue if `subscription` was made to cancel on |
31 /// an error. | 31 /// an error. |
32 SubscriptionStream(StreamSubscription subscription) | 32 SubscriptionStream(StreamSubscription<T> subscription) |
33 : _source = subscription { | 33 : _source = subscription { |
34 _source.pause(); | 34 _source.pause(); |
35 // Clear callbacks to avoid keeping them alive unnecessarily. | 35 // Clear callbacks to avoid keeping them alive unnecessarily. |
36 _source.onData(null); | 36 _source.onData(null); |
37 _source.onError(null); | 37 _source.onError(null); |
38 _source.onDone(null); | 38 _source.onDone(null); |
39 } | 39 } |
40 | 40 |
41 StreamSubscription<T> listen(void onData(T event), | 41 StreamSubscription<T> listen(void onData(T event), |
42 {Function onError, | 42 {Function onError, |
43 void onDone(), | 43 void onDone(), |
44 bool cancelOnError}) { | 44 bool cancelOnError}) { |
45 if (_source == null) { | 45 if (_source == null) { |
46 throw new StateError("Stream has already been listened to."); | 46 throw new StateError("Stream has already been listened to."); |
47 } | 47 } |
48 cancelOnError = (true == cancelOnError); | 48 cancelOnError = (true == cancelOnError); |
49 var subscription = _source; | 49 var subscription = _source; |
50 _source = null; | 50 _source = null; |
51 var result; | 51 |
52 if (cancelOnError) { | 52 var result = cancelOnError |
53 result = new _CancelOnErrorSubscriptionWrapper<T>(subscription); | 53 ? new _CancelOnErrorSubscriptionWrapper<T>(subscription) |
54 } else { | 54 : subscription; |
55 // Wrap the subscription to ensure correct type parameter. | |
56 result = new DelegatingStreamSubscription<T>(subscription); | |
57 } | |
58 result.onData(onData); | 55 result.onData(onData); |
59 result.onError(onError); | 56 result.onError(onError); |
60 result.onDone(onDone); | 57 result.onDone(onDone); |
61 subscription.resume(); | 58 subscription.resume(); |
62 return result; | 59 return result; |
63 } | 60 } |
64 } | 61 } |
65 | 62 |
66 /// Subscription wrapper that cancels on error. | 63 /// Subscription wrapper that cancels on error. |
67 /// | 64 /// |
(...skipping 22 matching lines...) Expand all Loading... |
90 } else { | 87 } else { |
91 if (handleError is ZoneBinaryCallback) { | 88 if (handleError is ZoneBinaryCallback) { |
92 handleError(error, stackTrace); | 89 handleError(error, stackTrace); |
93 } else { | 90 } else { |
94 handleError(error); | 91 handleError(error); |
95 } | 92 } |
96 } | 93 } |
97 }); | 94 }); |
98 } | 95 } |
99 } | 96 } |
OLD | NEW |