OLD | NEW |
1 // Copyright (c) 2013, the Dart project authors. Please see the AUTHORS file | 1 // Copyright (c) 2013, 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 library barback.utils; | 5 library barback.utils; |
6 | 6 |
7 import 'dart:async'; | 7 import 'dart:async'; |
8 | 8 |
9 /// A pair of values. | 9 /// A pair of values. |
10 class Pair<E, F> { | 10 class Pair<E, F> { |
(...skipping 176 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... |
187 /// then close. | 187 /// then close. |
188 /// | 188 /// |
189 /// If [broadcast] is true, a broadcast stream is returned. This assumes that | 189 /// If [broadcast] is true, a broadcast stream is returned. This assumes that |
190 /// the stream returned by [future] will be a broadcast stream as well. | 190 /// the stream returned by [future] will be a broadcast stream as well. |
191 /// [broadcast] defaults to false. | 191 /// [broadcast] defaults to false. |
192 Stream futureStream(Future<Stream> future, {bool broadcast: false}) { | 192 Stream futureStream(Future<Stream> future, {bool broadcast: false}) { |
193 var subscription; | 193 var subscription; |
194 var controller; | 194 var controller; |
195 | 195 |
196 future = future.catchError((e, stackTrace) { | 196 future = future.catchError((e, stackTrace) { |
197 if (controller == null) return; | 197 // Since [controller] is synchronous, it's likely that emitting an error |
198 controller.addError(e, stackTrace); | 198 // will cause it to be cancelled before we call close. |
199 controller.close(); | 199 if (controller != null) controller.addError(e, stackTrace); |
| 200 if (controller != null) controller.close(); |
200 controller = null; | 201 controller = null; |
201 }); | 202 }); |
202 | 203 |
203 onListen() { | 204 onListen() { |
204 future.then((stream) { | 205 future.then((stream) { |
205 if (controller == null) return; | 206 if (controller == null) return; |
206 subscription = stream.listen( | 207 subscription = stream.listen( |
207 controller.add, | 208 controller.add, |
208 onError: controller.addError, | 209 onError: controller.addError, |
209 onDone: controller.close); | 210 onDone: controller.close); |
(...skipping 27 matching lines...) Expand all Loading... |
237 subscription = callback().listen(controller.add, | 238 subscription = callback().listen(controller.add, |
238 onError: controller.addError, | 239 onError: controller.addError, |
239 onDone: controller.close); | 240 onDone: controller.close); |
240 }, | 241 }, |
241 onCancel: () => subscription.cancel(), | 242 onCancel: () => subscription.cancel(), |
242 onPause: () => subscription.pause(), | 243 onPause: () => subscription.pause(), |
243 onResume: () => subscription.resume(), | 244 onResume: () => subscription.resume(), |
244 sync: true); | 245 sync: true); |
245 return controller.stream; | 246 return controller.stream; |
246 } | 247 } |
OLD | NEW |