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 watcher.utils; | 5 library watcher.utils; |
6 | 6 |
7 import 'dart:async'; | 7 import 'dart:async'; |
8 import 'dart:io'; | 8 import 'dart:io'; |
9 import 'dart:collection'; | 9 import 'dart:collection'; |
10 | 10 |
(...skipping 18 matching lines...) Expand all Loading... |
29 /// then close. | 29 /// then close. |
30 /// | 30 /// |
31 /// If [broadcast] is true, a broadcast stream is returned. This assumes that | 31 /// If [broadcast] is true, a broadcast stream is returned. This assumes that |
32 /// the stream returned by [future] will be a broadcast stream as well. | 32 /// the stream returned by [future] will be a broadcast stream as well. |
33 /// [broadcast] defaults to false. | 33 /// [broadcast] defaults to false. |
34 Stream futureStream(Future<Stream> future, {bool broadcast: false}) { | 34 Stream futureStream(Future<Stream> future, {bool broadcast: false}) { |
35 var subscription; | 35 var subscription; |
36 var controller; | 36 var controller; |
37 | 37 |
38 future = future.catchError((e, stackTrace) { | 38 future = future.catchError((e, stackTrace) { |
39 if (controller == null) return; | 39 // Since [controller] is synchronous, it's likely that emitting an error |
40 controller.addError(e, stackTrace); | 40 // will cause it to be cancelled before we call close. |
41 controller.close(); | 41 if (controller != null) controller.addError(e, stackTrace); |
| 42 if (controller != null) controller.close(); |
42 controller = null; | 43 controller = null; |
43 }); | 44 }); |
44 | 45 |
45 onListen() { | 46 onListen() { |
46 future.then((stream) { | 47 future.then((stream) { |
47 if (controller == null) return; | 48 if (controller == null) return; |
48 subscription = stream.listen( | 49 subscription = stream.listen( |
49 controller.add, | 50 controller.add, |
50 onError: controller.addError, | 51 onError: controller.addError, |
51 onDone: controller.close); | 52 onDone: controller.close); |
(...skipping 43 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... |
95 }); | 96 }); |
96 }, handleDone: (sink) { | 97 }, handleDone: (sink) { |
97 if (batch.isNotEmpty) { | 98 if (batch.isNotEmpty) { |
98 sink.add(batch.toList()); | 99 sink.add(batch.toList()); |
99 batch.clear(); | 100 batch.clear(); |
100 } | 101 } |
101 sink.close(); | 102 sink.close(); |
102 }).bind(input); | 103 }).bind(input); |
103 } | 104 } |
104 } | 105 } |
OLD | NEW |