OLD | NEW |
1 // Copyright (c) 2016, the Dart project authors. Please see the AUTHORS file | 1 // Copyright (c) 2016, 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 /// A stream that combines the values of other streams. | 7 /// A stream that combines the values of other streams. |
8 /// | 8 /// |
9 /// This emits lists of collected values from each input stream. The first list | 9 /// This emits lists of collected values from each input stream. The first list |
10 /// contains the first value emitted by each stream, the second contrains the | 10 /// contains the first value emitted by each stream, the second contrains the |
11 /// second value, and so on. The lists have the same ordering as the iterable | 11 /// second value, and so on. The lists have the same ordering as the iterable |
12 /// passed to [new StreamZip]. | 12 /// passed to [new StreamZip]. |
13 /// | 13 /// |
14 /// Any errors from any of the streams are forwarded directly to this stream. | 14 /// Any errors from any of the streams are forwarded directly to this stream. |
15 class StreamZip<T> extends Stream<List<T>> { | 15 class StreamZip<T> extends Stream<List<T>> { |
16 final Iterable<Stream<T>> _streams; | 16 final Iterable<Stream<T>> _streams; |
17 | 17 |
18 StreamZip(Iterable<Stream<T>> streams) : _streams = streams; | 18 StreamZip(Iterable<Stream<T>> streams) : _streams = streams; |
19 | 19 |
20 StreamSubscription<List<T>> listen(void onData(List data), { | 20 StreamSubscription<List<T>> listen(void onData(List<T> data), { |
21 Function onError, | 21 Function onError, |
22 void onDone(), | 22 void onDone(), |
23 bool cancelOnError}) { | 23 bool cancelOnError}) { |
24 cancelOnError = identical(true, cancelOnError); | 24 cancelOnError = identical(true, cancelOnError); |
25 List<StreamSubscription> subscriptions = <StreamSubscription>[]; | 25 var subscriptions = <StreamSubscription<T>>[]; |
26 StreamController controller; | 26 StreamController<List<T>> controller; |
27 List current; | 27 List<T> current; |
28 int dataCount = 0; | 28 int dataCount = 0; |
29 | 29 |
30 /// Called for each data from a subscription in [subscriptions]. | 30 /// Called for each data from a subscription in [subscriptions]. |
31 void handleData(int index, data) { | 31 void handleData(int index, T data) { |
32 current[index] = data; | 32 current[index] = data; |
33 dataCount++; | 33 dataCount++; |
34 if (dataCount == subscriptions.length) { | 34 if (dataCount == subscriptions.length) { |
35 List data = current; | 35 var data = current; |
36 current = new List(subscriptions.length); | 36 current = new List(subscriptions.length); |
37 dataCount = 0; | 37 dataCount = 0; |
38 for (int i = 0; i < subscriptions.length; i++) { | 38 for (int i = 0; i < subscriptions.length; i++) { |
39 if (i != index) subscriptions[i].resume(); | 39 if (i != index) subscriptions[i].resume(); |
40 } | 40 } |
41 controller.add(data); | 41 controller.add(data); |
42 } else { | 42 } else { |
43 subscriptions[index].pause(); | 43 subscriptions[index].pause(); |
44 } | 44 } |
45 } | 45 } |
(...skipping 17 matching lines...) Expand all Loading... |
63 } | 63 } |
64 | 64 |
65 void handleDone() { | 65 void handleDone() { |
66 for (int i = 0; i < subscriptions.length; i++) { | 66 for (int i = 0; i < subscriptions.length; i++) { |
67 subscriptions[i].cancel(); | 67 subscriptions[i].cancel(); |
68 } | 68 } |
69 controller.close(); | 69 controller.close(); |
70 } | 70 } |
71 | 71 |
72 try { | 72 try { |
73 for (Stream stream in _streams) { | 73 for (var stream in _streams) { |
74 int index = subscriptions.length; | 74 int index = subscriptions.length; |
75 subscriptions.add(stream.listen( | 75 subscriptions.add(stream.listen( |
76 (data) { handleData(index, data); }, | 76 (data) { handleData(index, data); }, |
77 onError: cancelOnError ? handleError : handleErrorCancel, | 77 onError: cancelOnError ? handleError : handleErrorCancel, |
78 onDone: handleDone, | 78 onDone: handleDone, |
79 cancelOnError: cancelOnError)); | 79 cancelOnError: cancelOnError)); |
80 } | 80 } |
81 } catch (e) { | 81 } catch (e) { |
82 for (int i = subscriptions.length - 1; i >= 0; i--) { | 82 for (int i = subscriptions.length - 1; i >= 0; i--) { |
83 subscriptions[i].cancel(); | 83 subscriptions[i].cancel(); |
84 } | 84 } |
85 rethrow; | 85 rethrow; |
86 } | 86 } |
87 | 87 |
88 current = new List(subscriptions.length); | 88 current = new List(subscriptions.length); |
89 | 89 |
90 controller = new StreamController<List>( | 90 controller = new StreamController<List<T>>( |
91 onPause: () { | 91 onPause: () { |
92 for (int i = 0; i < subscriptions.length; i++) { | 92 for (int i = 0; i < subscriptions.length; i++) { |
93 // This may pause some subscriptions more than once. | 93 // This may pause some subscriptions more than once. |
94 // These will not be resumed by onResume below, but must wait for the | 94 // These will not be resumed by onResume below, but must wait for the |
95 // next round. | 95 // next round. |
96 subscriptions[i].pause(); | 96 subscriptions[i].pause(); |
97 } | 97 } |
98 }, | 98 }, |
99 onResume: () { | 99 onResume: () { |
100 for (int i = 0; i < subscriptions.length; i++) { | 100 for (int i = 0; i < subscriptions.length; i++) { |
(...skipping 10 matching lines...) Expand all Loading... |
111 | 111 |
112 if (subscriptions.isEmpty) { | 112 if (subscriptions.isEmpty) { |
113 controller.close(); | 113 controller.close(); |
114 } | 114 } |
115 return controller.stream.listen(onData, | 115 return controller.stream.listen(onData, |
116 onError: onError, | 116 onError: onError, |
117 onDone: onDone, | 117 onDone: onDone, |
118 cancelOnError: cancelOnError); | 118 cancelOnError: cancelOnError); |
119 } | 119 } |
120 } | 120 } |
OLD | NEW |