| 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 |