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 import 'package:expect/expect.dart'; | 5 import 'package:expect/expect.dart'; |
6 import 'package:async_helper/async_helper.dart'; | 6 import 'package:async_helper/async_helper.dart'; |
7 import 'dart:async'; | 7 import 'dart:async'; |
8 import 'event_helper.dart'; | 8 import 'event_helper.dart'; |
9 | 9 |
10 | |
11 class DecrementingTransformerSink implements EventSink { | 10 class DecrementingTransformerSink implements EventSink { |
12 final outSink; | 11 final outSink; |
13 DecrementingTransformerSink(this.outSink); | 12 DecrementingTransformerSink(this.outSink); |
14 | 13 |
15 void add(int i) => outSink.add(i - 1); | 14 void add(int i) => outSink.add(i - 1); |
16 void addError(int e, [st]) => outSink.addError(e - 1, st); | 15 void addError(int e, [st]) => outSink.addError(e - 1, st); |
17 void close() => outSink.close(); | 16 void close() => outSink.close(); |
18 } | 17 } |
19 | 18 |
20 class FutureWaitingTransformerSink implements EventSink { | 19 class FutureWaitingTransformerSink implements EventSink { |
21 final outSink; | 20 final outSink; |
22 final closeFuture; | 21 final closeFuture; |
23 FutureWaitingTransformerSink(this.outSink, this.closeFuture); | 22 FutureWaitingTransformerSink(this.outSink, this.closeFuture); |
24 | 23 |
25 void add(Future future) { future.then(outSink.add); } | 24 void add(Future future) { |
26 void addError(Future e, [st]) { e.then((val) { outSink.addError(val, st); });
} | 25 future.then(outSink.add); |
27 void close() { closeFuture.whenComplete(outSink.close); } | 26 } |
| 27 |
| 28 void addError(Future e, [st]) { |
| 29 e.then((val) { |
| 30 outSink.addError(val, st); |
| 31 }); |
| 32 } |
| 33 |
| 34 void close() { |
| 35 closeFuture.whenComplete(outSink.close); |
| 36 } |
28 } | 37 } |
29 | 38 |
30 class ZoneTransformerSink implements EventSink { | 39 class ZoneTransformerSink implements EventSink { |
31 final outSink; | 40 final outSink; |
32 ZoneTransformerSink(this.outSink); | 41 ZoneTransformerSink(this.outSink); |
33 | 42 |
34 void add(_) { outSink.add(Zone.current); } | 43 void add(_) { |
35 void addError(_, [st]) { outSink.add(Zone.current); } | 44 outSink.add(Zone.current); |
| 45 } |
| 46 |
| 47 void addError(_, [st]) { |
| 48 outSink.add(Zone.current); |
| 49 } |
| 50 |
36 void close() { | 51 void close() { |
37 outSink.add(Zone.current); | 52 outSink.add(Zone.current); |
38 outSink.close(); | 53 outSink.close(); |
39 } | 54 } |
40 } | 55 } |
41 | 56 |
42 class TypeChangingSink implements EventSink<int> { | 57 class TypeChangingSink implements EventSink<int> { |
43 final EventSink<String> outSink; | 58 final EventSink<String> outSink; |
44 TypeChangingSink(this.outSink); | 59 TypeChangingSink(this.outSink); |
45 | 60 |
46 void add(int data) { outSink.add(data.toString()); } | 61 void add(int data) { |
47 void addError(error, [st]) { outSink.addError(error, st); } | 62 outSink.add(data.toString()); |
48 void close() { outSink.close(); } | 63 } |
| 64 |
| 65 void addError(error, [st]) { |
| 66 outSink.addError(error, st); |
| 67 } |
| 68 |
| 69 void close() { |
| 70 outSink.close(); |
| 71 } |
49 } | 72 } |
50 | 73 |
51 class SinkTransformer<S, T> implements StreamTransformer<S, T> { | 74 class SinkTransformer<S, T> implements StreamTransformer<S, T> { |
52 final Function sinkMapper; | 75 final Function sinkMapper; |
53 SinkTransformer(this.sinkMapper); | 76 SinkTransformer(this.sinkMapper); |
54 | 77 |
55 Stream<T> bind(Stream<S> stream) { | 78 Stream<T> bind(Stream<S> stream) { |
56 return new Stream<T>.eventTransformed(stream, sinkMapper); | 79 return new Stream<T>.eventTransformed(stream, sinkMapper); |
57 } | 80 } |
58 } | 81 } |
(...skipping 19 matching lines...) Expand all Loading... |
78 }); | 101 }); |
79 } | 102 } |
80 | 103 |
81 main() { | 104 main() { |
82 { | 105 { |
83 // Simple test: use the SinkTransformer (using the Stream.eventTransformed | 106 // Simple test: use the SinkTransformer (using the Stream.eventTransformed |
84 // constructor) to transform a sequence of numbers. This is basically | 107 // constructor) to transform a sequence of numbers. This is basically |
85 // similar to a map. | 108 // similar to a map. |
86 asyncStart(); | 109 asyncStart(); |
87 new Stream.fromIterable([1, 2, 3]) | 110 new Stream.fromIterable([1, 2, 3]) |
88 .transform( | 111 .transform(new SinkTransformer( |
89 new SinkTransformer((sink) => new DecrementingTransformerSink(sink))) | 112 (sink) => new DecrementingTransformerSink(sink))) |
90 .toList() | 113 .toList() |
91 .then((list) { | 114 .then((list) { |
92 Expect.listEquals([0, 1, 2], list); | 115 Expect.listEquals([0, 1, 2], list); |
93 asyncEnd(); | 116 asyncEnd(); |
94 }); | 117 }); |
95 } | 118 } |
96 | 119 |
97 { | 120 { |
98 // Similar test as above: but this time also transform errors. Also | 121 // Similar test as above: but this time also transform errors. Also |
99 // checks that the stack trace is correctly passed through. | 122 // checks that the stack trace is correctly passed through. |
100 asyncStart(); | 123 asyncStart(); |
101 var controller; | 124 var controller; |
102 var events = []; | 125 var events = []; |
103 var stackTrace = currentStackTrace; | 126 var stackTrace = currentStackTrace; |
104 controller = new StreamController(onListen: () { | 127 controller = new StreamController(onListen: () { |
105 controller.add(499); | 128 controller.add(499); |
106 controller.addError(42, stackTrace); | 129 controller.addError(42, stackTrace); |
107 controller.close(); | 130 controller.close(); |
108 }); | 131 }); |
109 controller.stream | 132 controller.stream |
110 .transform( | 133 .transform(new SinkTransformer( |
111 new SinkTransformer((sink) => new DecrementingTransformerSink(sink))) | 134 (sink) => new DecrementingTransformerSink(sink))) |
112 .listen((data) { | 135 .listen((data) { |
113 events.add(data); | 136 events.add(data); |
114 }, onError: (e, st) { | 137 }, onError: (e, st) { |
115 events.add(e); | 138 events.add(e); |
116 events.add(st); | 139 events.add(st); |
117 }, onDone: () { | 140 }, onDone: () { |
118 Expect.listEquals([498, 41, stackTrace], events); | 141 Expect.listEquals([498, 41, stackTrace], events); |
119 asyncEnd(); | 142 asyncEnd(); |
120 }); | 143 }); |
121 } | 144 } |
122 | 145 |
123 { | 146 { |
124 // Test that the output sink of the transformer can be used asynchronously. | 147 // Test that the output sink of the transformer can be used asynchronously. |
125 asyncStart(); | 148 asyncStart(); |
126 var controller; | 149 var controller; |
127 var events = []; | 150 var events = []; |
128 var stackTrace = currentStackTrace; | 151 var stackTrace = currentStackTrace; |
129 var completer1 = new Completer(); | 152 var completer1 = new Completer(); |
130 var completer2 = new Completer(); | 153 var completer2 = new Completer(); |
131 var completer3 = new Completer(); | 154 var completer3 = new Completer(); |
132 var closeCompleter = new Completer(); | 155 var closeCompleter = new Completer(); |
133 controller = new StreamController(onListen: () { | 156 controller = new StreamController(onListen: () { |
134 controller.add(completer1.future); | 157 controller.add(completer1.future); |
135 controller.addError(completer2.future, stackTrace); | 158 controller.addError(completer2.future, stackTrace); |
136 controller.add(completer3.future); | 159 controller.add(completer3.future); |
137 controller.close(); | 160 controller.close(); |
138 }); | 161 }); |
139 controller.stream | 162 controller.stream |
140 .transform( | 163 .transform(new SinkTransformer((sink) => |
141 new SinkTransformer((sink) => | |
142 new FutureWaitingTransformerSink(sink, closeCompleter.future))) | 164 new FutureWaitingTransformerSink(sink, closeCompleter.future))) |
143 .listen((data) { | 165 .listen((data) { |
144 events.add(data); | 166 events.add(data); |
145 }, onError: (e, st) { | 167 }, onError: (e, st) { |
146 events.add(e); | 168 events.add(e); |
147 events.add(st); | 169 events.add(st); |
148 }, onDone: () { | 170 }, onDone: () { |
149 Expect.listEquals(["error2", stackTrace, "future3", "future1"], events); | 171 Expect.listEquals(["error2", stackTrace, "future3", "future1"], events); |
150 asyncEnd(); | 172 asyncEnd(); |
151 }); | 173 }); |
152 Timer.run(() { | 174 Timer.run(() { |
153 completer2.complete("error2"); | 175 completer2.complete("error2"); |
154 Timer.run(() { | 176 Timer.run(() { |
155 completer3.complete("future3"); | 177 completer3.complete("future3"); |
156 Timer.run(() { | 178 Timer.run(() { |
157 completer1.complete("future1"); | 179 completer1.complete("future1"); |
158 scheduleMicrotask(closeCompleter.complete); | 180 scheduleMicrotask(closeCompleter.complete); |
159 }); | 181 }); |
160 }); | 182 }); |
161 }); | 183 }); |
(...skipping 10 matching lines...) Expand all Loading... |
172 var completer2 = new Completer.sync(); | 194 var completer2 = new Completer.sync(); |
173 var completer3 = new Completer.sync(); | 195 var completer3 = new Completer.sync(); |
174 var closeCompleter = new Completer(); | 196 var closeCompleter = new Completer(); |
175 controller = new StreamController(onListen: () { | 197 controller = new StreamController(onListen: () { |
176 controller.add(completer1.future); | 198 controller.add(completer1.future); |
177 controller.addError(completer2.future, stackTrace); | 199 controller.addError(completer2.future, stackTrace); |
178 controller.add(completer3.future); | 200 controller.add(completer3.future); |
179 controller.close(); | 201 controller.close(); |
180 }); | 202 }); |
181 var subscription; | 203 var subscription; |
182 completer1.future.then((_) { Expect.isTrue(subscription.isPaused); }); | 204 completer1.future.then((_) { |
183 completer2.future.then((_) { Expect.isTrue(subscription.isPaused); }); | 205 Expect.isTrue(subscription.isPaused); |
184 completer3.future.then((_) { Expect.isTrue(subscription.isPaused); }); | 206 }); |
| 207 completer2.future.then((_) { |
| 208 Expect.isTrue(subscription.isPaused); |
| 209 }); |
| 210 completer3.future.then((_) { |
| 211 Expect.isTrue(subscription.isPaused); |
| 212 }); |
185 subscription = controller.stream | 213 subscription = controller.stream |
186 .transform( | 214 .transform(new SinkTransformer((sink) => |
187 new SinkTransformer((sink) => | |
188 new FutureWaitingTransformerSink(sink, closeCompleter.future))) | 215 new FutureWaitingTransformerSink(sink, closeCompleter.future))) |
189 .listen((data) { | 216 .listen((data) { |
190 Expect.isFalse(subscription.isPaused); | 217 Expect.isFalse(subscription.isPaused); |
191 events.add(data); | 218 events.add(data); |
192 }, onError: (e, st) { | 219 }, onError: (e, st) { |
193 events.add(e); | 220 events.add(e); |
194 events.add(st); | 221 events.add(st); |
195 }, onDone: () { | 222 }, onDone: () { |
196 Expect.listEquals(["error2", stackTrace, "future3", "future1"], events); | 223 Expect.listEquals(["error2", stackTrace, "future3", "future1"], events); |
197 asyncEnd(); | 224 asyncEnd(); |
198 }); | 225 }); |
199 Timer.run(() { | 226 Timer.run(() { |
200 subscription.pause(); | 227 subscription.pause(); |
201 completer2.complete("error2"); | 228 completer2.complete("error2"); |
202 Timer.run(() { | 229 Timer.run(() { |
203 subscription.resume(); | 230 subscription.resume(); |
204 Timer.run(() { | 231 Timer.run(() { |
205 Expect.listEquals(["error2", stackTrace], events); | 232 Expect.listEquals(["error2", stackTrace], events); |
206 subscription.pause(); | 233 subscription.pause(); |
207 completer3.complete("future3"); | 234 completer3.complete("future3"); |
208 Timer.run(() { | 235 Timer.run(() { |
(...skipping 26 matching lines...) Expand all Loading... |
235 controller.add(completer1.future); | 262 controller.add(completer1.future); |
236 controller.addError(completer2.future, stackTrace); | 263 controller.addError(completer2.future, stackTrace); |
237 controller.add(completer3.future); | 264 controller.add(completer3.future); |
238 controller.close(); | 265 controller.close(); |
239 }); | 266 }); |
240 | 267 |
241 bool streamIsDone = false; | 268 bool streamIsDone = false; |
242 int errorCount = 0; | 269 int errorCount = 0; |
243 runZoned(() { | 270 runZoned(() { |
244 controller.stream | 271 controller.stream |
245 .transform( | 272 .transform(new SinkTransformer((sink) => |
246 new SinkTransformer((sink) => | |
247 new FutureWaitingTransformerSink(sink, closeCompleter.future))) | 273 new FutureWaitingTransformerSink(sink, closeCompleter.future))) |
248 .listen((data) { | 274 .listen((data) { |
249 events.add(data); | 275 events.add(data); |
250 }, onError: (e, st) { | 276 }, onError: (e, st) { |
251 events.add(e); | 277 events.add(e); |
252 events.add(st); | 278 events.add(st); |
253 }, onDone: () { | 279 }, onDone: () { |
254 Expect.listEquals([], events); | 280 Expect.listEquals([], events); |
255 streamIsDone = true; | 281 streamIsDone = true; |
256 }); | 282 }); |
257 }, onError: (e) { | 283 }, onError: (e) { |
258 Expect.isTrue(e is StateError); | 284 Expect.isTrue(e is StateError); |
259 errorCount++; | 285 errorCount++; |
260 }); | 286 }); |
261 closeCompleter.complete(); | 287 closeCompleter.complete(); |
262 Timer.run(() { | 288 Timer.run(() { |
263 Expect.isTrue(streamIsDone); | 289 Expect.isTrue(streamIsDone); |
264 // Each of the delayed completions should trigger an unhandled error | 290 // Each of the delayed completions should trigger an unhandled error |
265 // in the zone the stream was listened to. | 291 // in the zone the stream was listened to. |
266 Timer.run(() { completer1.complete(499); }); | 292 Timer.run(() { |
267 Timer.run(() { completer2.complete(42); }); | 293 completer1.complete(499); |
268 Timer.run(() { completer3.complete(99); }); | 294 }); |
| 295 Timer.run(() { |
| 296 completer2.complete(42); |
| 297 }); |
| 298 Timer.run(() { |
| 299 completer3.complete(99); |
| 300 }); |
269 delayCycles(() { | 301 delayCycles(() { |
270 Expect.equals(3, errorCount); | 302 Expect.equals(3, errorCount); |
271 asyncEnd(); | 303 asyncEnd(); |
272 }, 5); | 304 }, 5); |
273 }); | 305 }); |
274 } | 306 } |
275 | 307 |
276 { | 308 { |
277 // Test that the transformer is executed in the zone it was listened to. | 309 // Test that the transformer is executed in the zone it was listened to. |
278 asyncStart(); | 310 asyncStart(); |
279 var stackTrace = currentStackTrace; | 311 var stackTrace = currentStackTrace; |
280 var events = []; | 312 var events = []; |
281 var controller; | 313 var controller; |
282 controller = new StreamController(onListen: () { | 314 controller = new StreamController(onListen: () { |
283 // Events are added outside the zone. | 315 // Events are added outside the zone. |
284 controller.add(499); | 316 controller.add(499); |
285 controller.addError(42, stackTrace); | 317 controller.addError(42, stackTrace); |
286 controller.close(); | 318 controller.close(); |
287 }); | 319 }); |
288 Zone zone = Zone.current.fork(); | 320 Zone zone = Zone.current.fork(); |
289 var stream = controller.stream.transform( | 321 var stream = controller.stream.transform( |
290 new SinkTransformer((sink) => new ZoneTransformerSink(sink))); | 322 new SinkTransformer((sink) => new ZoneTransformerSink(sink))); |
291 zone.run(() { | 323 zone.run(() { |
292 stream.listen((data) { | 324 stream.listen((data) { |
293 events.add(data); | 325 events.add(data); |
294 }, onDone: () { | 326 }, onDone: () { |
295 Expect.listEquals([zone, zone, zone], events); | 327 Expect.listEquals([zone, zone, zone], events); |
296 delayCycles(asyncEnd, 3); | 328 delayCycles(asyncEnd, 3); |
297 }); | 329 }); |
298 }); | 330 }); |
299 } | 331 } |
300 | 332 |
301 { | 333 { |
302 // Just make sure that the generic types are correct everywhere. | 334 // Just make sure that the generic types are correct everywhere. |
303 asyncStart(); | 335 asyncStart(); |
304 new Stream.fromIterable([1, 2, 3]) | 336 new Stream.fromIterable([1, 2, 3]) |
305 .transform(new SinkTransformer<int, String>( | 337 .transform(new SinkTransformer<int, String>( |
306 (sink) => new TypeChangingSink(sink))) | 338 (sink) => new TypeChangingSink(sink))) |
307 .toList() | 339 .toList() |
308 .then((list) { | 340 .then((list) { |
309 Expect.listEquals(["1", "2", "3"], list); | 341 Expect.listEquals(["1", "2", "3"], list); |
310 asyncEnd(); | 342 asyncEnd(); |
311 }); | 343 }); |
312 } | 344 } |
313 } | 345 } |
OLD | NEW |