OLD | NEW |
---|---|
1 // Copyright (c) 2011, the Dart project authors. Please see the AUTHORS file | 1 // Copyright (c) 2011, 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 // Test the basic StreamController and StreamController.singleSubscription. | 5 // Test the basic StreamController and StreamController.singleSubscription. |
6 library stream_controller_async_test; | 6 library stream_controller_async_test; |
7 | 7 |
8 import "package:expect/expect.dart"; | 8 import "package:expect/expect.dart"; |
9 import 'dart:async'; | 9 import 'dart:async'; |
10 import 'dart:isolate'; | 10 import 'dart:isolate'; |
11 import '../../../pkg/unittest/lib/unittest.dart'; | 11 import '../../../pkg/unittest/lib/unittest.dart'; |
12 import 'event_helper.dart'; | 12 import 'event_helper.dart'; |
13 import 'stream_state_helper.dart'; | |
13 | 14 |
14 testController() { | 15 testController() { |
15 // Test fold | 16 // Test fold |
16 test("StreamController.fold", () { | 17 test("StreamController.fold", () { |
17 StreamController c = new StreamController(); | 18 StreamController c = new StreamController(); |
18 Stream stream = c.stream.asBroadcastStream(); | 19 Stream stream = c.stream.asBroadcastStream(); |
19 stream.fold(0, (a,b) => a + b) | 20 stream.fold(0, (a,b) => a + b) |
20 .then(expectAsync1((int v) { | 21 .then(expectAsync1((int v) { |
21 Expect.equals(42, v); | 22 Expect.equals(42, v); |
22 })); | 23 })); |
(...skipping 236 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... | |
259 Future f = c.stream.drain(); | 260 Future f = c.stream.drain(); |
260 f.catchError(expectAsync1((error) { Expect.equals("error", error); })); | 261 f.catchError(expectAsync1((error) { Expect.equals("error", error); })); |
261 Events errorEvents = new Events()..error("error")..error("error2")..close(); | 262 Events errorEvents = new Events()..error("error")..error("error2")..close(); |
262 errorEvents.replay(c); | 263 errorEvents.replay(c); |
263 }); | 264 }); |
264 | 265 |
265 } | 266 } |
266 | 267 |
267 testPause() { | 268 testPause() { |
268 test("pause event-unpause", () { | 269 test("pause event-unpause", () { |
269 StreamController c = new StreamController(); | 270 StreamProtocolTest test = new StreamProtocolTest(); |
270 Events actualEvents = new Events.capture(c.stream); | |
271 Events expectedEvents = new Events(); | |
272 expectedEvents.add(42); | |
273 c.add(42); | |
274 Expect.listEquals(expectedEvents.events, actualEvents.events); | |
275 Completer completer = new Completer(); | 271 Completer completer = new Completer(); |
276 actualEvents.pause(completer.future); | 272 test..expectListen() |
277 c..add(43)..add(44)..close(); | 273 ..expectData(42, () { test.pause(completer.future); }) |
278 Expect.listEquals(expectedEvents.events, actualEvents.events); | 274 ..expectPause(() { |
279 completer.complete(); | 275 completer.complete(null); |
280 expectedEvents..add(43)..add(44)..close(); | 276 }) |
281 actualEvents.onDone(expectAsync0(() { | 277 ..expectData(43) |
282 Expect.listEquals(expectedEvents.events, actualEvents.events); | 278 ..expectData(44) |
283 })); | 279 ..expectDone() |
280 ..expectCancel(); | |
281 test.listen(); | |
282 test.add(42); | |
283 test.add(43); | |
284 test.add(44); | |
285 test.close(); | |
284 }); | 286 }); |
285 | 287 |
286 test("pause twice event-unpause", () { | 288 test("pause twice event-unpause", () { |
287 StreamController c = new StreamController(); | 289 StreamProtocolTest test = new StreamProtocolTest(); |
288 Events actualEvents = new Events.capture(c.stream); | |
289 Events expectedEvents = new Events(); | |
290 expectedEvents.add(42); | |
291 c.add(42); | |
292 Expect.listEquals(expectedEvents.events, actualEvents.events); | |
293 Completer completer = new Completer(); | 290 Completer completer = new Completer(); |
294 Completer completer2 = new Completer(); | 291 Completer completer2 = new Completer(); |
295 actualEvents.pause(completer.future); | 292 test..expectListen() |
296 actualEvents.pause(completer2.future); | 293 ..expectData(42, () { |
297 c..add(43)..add(44)..close(); | 294 test.pause(completer.future); |
floitsch
2013/05/30 12:13:48
indent + 2
Lasse Reichstein Nielsen
2013/05/31 05:51:59
Done.
| |
298 Expect.listEquals(expectedEvents.events, actualEvents.events); | 295 test.pause(completer2.future); |
299 completer.complete(); | 296 }) |
300 Expect.listEquals(expectedEvents.events, actualEvents.events); | 297 ..expectPause(() { |
301 completer2.complete(); | 298 completer.future.then(completer2.complete); |
302 expectedEvents..add(43)..add(44)..close(); | 299 completer.complete(null); |
303 actualEvents.onDone(expectAsync0((){ | 300 }) |
304 Expect.listEquals(expectedEvents.events, actualEvents.events); | 301 ..expectData(43) |
305 })); | 302 ..expectData(44) |
303 ..expectDone() | |
304 ..expectCancel(); | |
305 test..listen() | |
306 ..add(42) | |
307 ..add(43) | |
308 ..add(44) | |
309 ..close(); | |
306 }); | 310 }); |
307 | 311 |
308 test("pause twice direct-unpause", () { | 312 test("pause twice direct-unpause", () { |
309 StreamController c = new StreamController(); | 313 StreamProtocolTest test = new StreamProtocolTest(); |
310 Events actualEvents = new Events.capture(c.stream); | 314 test..expectListen() |
311 Events expectedEvents = new Events(); | 315 ..expectData(42, () { |
312 expectedEvents.add(42); | 316 test.pause(); |
313 c.add(42); | 317 test.pause(); |
314 Expect.listEquals(expectedEvents.events, actualEvents.events); | 318 }) |
315 actualEvents.pause(); | 319 ..expectPause(() { |
316 actualEvents.pause(); | 320 test.resume(); |
317 c.add(43); | 321 test.resume(); |
318 c.add(44); | 322 }) |
319 c.close(); | 323 ..expectResume() // This one resumes inpuot because resume is synchrono us. |
floitsch
2013/05/30 12:13:48
... input... long line.
Lasse Reichstein Nielsen
2013/05/31 05:51:59
Done.
| |
320 Expect.listEquals(expectedEvents.events, actualEvents.events); | 324 ..expectData(43) |
321 actualEvents.resume(); | 325 ..expectData(44) |
322 Expect.listEquals(expectedEvents.events, actualEvents.events); | 326 ..expectDone() |
323 expectedEvents..add(43)..add(44)..close(); | 327 ..expectCancel(); |
324 actualEvents.onDone(expectAsync0(() { | 328 test..listen() |
325 Expect.listEquals(expectedEvents.events, actualEvents.events); | 329 ..add(42) |
326 })); | 330 ..add(43) |
327 actualEvents.resume(); | 331 ..add(44) |
332 ..close(); | |
328 }); | 333 }); |
329 | 334 |
330 test("pause twice direct-event-unpause", () { | 335 test("pause twice direct-event-unpause", () { |
331 StreamController c = new StreamController(); | 336 StreamProtocolTest test = new StreamProtocolTest(); |
332 Events actualEvents = new Events.capture(c.stream); | |
333 Events expectedEvents = new Events(); | |
334 expectedEvents.add(42); | |
335 c.add(42); | |
336 Expect.listEquals(expectedEvents.events, actualEvents.events); | |
337 Completer completer = new Completer(); | 337 Completer completer = new Completer(); |
338 actualEvents.pause(completer.future); | 338 test..expectListen() |
339 actualEvents.pause(); | 339 ..expectData(42, () { |
340 c.add(43); | 340 test.pause(); |
341 c.add(44); | 341 test.pause(completer.future); |
342 c.close(); | 342 test.add(43); |
343 Expect.listEquals(expectedEvents.events, actualEvents.events); | 343 test.add(44); |
344 actualEvents.resume(); | 344 test.close(); |
345 Expect.listEquals(expectedEvents.events, actualEvents.events); | 345 }) |
346 expectedEvents..add(43)..add(44)..close(); | 346 ..expectPause(() { |
347 actualEvents.onDone(expectAsync0(() { | 347 completer.future.then((v) => test.resume()); |
348 Expect.listEquals(expectedEvents.events, actualEvents.events); | 348 completer.complete(null); |
349 })); | 349 }) |
350 completer.complete(); | 350 ..expectData(43) |
351 }); | 351 ..expectData(44) |
352 | 352 ..expectDone() |
353 test("pause twice direct-unpause", () { | 353 ..expectCancel(); |
354 StreamController c = new StreamController(); | 354 test..listen() |
355 Events actualEvents = new Events.capture(c.stream); | 355 ..add(42); |
356 Events expectedEvents = new Events(); | |
357 expectedEvents.add(42); | |
358 c.add(42); | |
359 Expect.listEquals(expectedEvents.events, actualEvents.events); | |
360 Completer completer = new Completer(); | |
361 actualEvents.pause(completer.future); | |
362 actualEvents.pause(); | |
363 c.add(43); | |
364 c.add(44); | |
365 c.close(); | |
366 Expect.listEquals(expectedEvents.events, actualEvents.events); | |
367 completer.complete(); | |
368 Expect.listEquals(expectedEvents.events, actualEvents.events); | |
369 expectedEvents..add(43)..add(44)..close(); | |
370 actualEvents.onDone(expectAsync0(() { | |
371 Expect.listEquals(expectedEvents.events, actualEvents.events); | |
372 })); | |
373 actualEvents.resume(); | |
374 }); | 356 }); |
375 } | 357 } |
376 | 358 |
377 class TestError { const TestError(); } | 359 class TestError { const TestError(); } |
378 | 360 |
379 testRethrow() { | 361 testRethrow() { |
380 TestError error = const TestError(); | 362 TestError error = const TestError(); |
381 | 363 |
382 | 364 |
383 testStream(name, streamValueTransform) { | 365 testStream(name, streamValueTransform) { |
(...skipping 40 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... | |
424 testFuture("forEach", (s, act) => s.forEach(act)); | 406 testFuture("forEach", (s, act) => s.forEach(act)); |
425 testFuture("every", (s, act) => s.every(act)); | 407 testFuture("every", (s, act) => s.every(act)); |
426 testFuture("any", (s, act) => s.any(act)); | 408 testFuture("any", (s, act) => s.any(act)); |
427 testFuture("reduce", (s, act) => s.reduce((a,b) => act(b))); | 409 testFuture("reduce", (s, act) => s.reduce((a,b) => act(b))); |
428 testFuture("fold", (s, act) => s.fold(0, (a,b) => act(b))); | 410 testFuture("fold", (s, act) => s.fold(0, (a,b) => act(b))); |
429 testFuture("drain", (s, act) => s.drain().then(act)); | 411 testFuture("drain", (s, act) => s.drain().then(act)); |
430 } | 412 } |
431 | 413 |
432 void testBroadcastController() { | 414 void testBroadcastController() { |
433 test("broadcast-controller-basic", () { | 415 test("broadcast-controller-basic", () { |
434 StreamController<int> c = new StreamController.broadcast( | 416 StreamProtocolTest test = new StreamProtocolTest.broadcast(); |
435 onListen: expectAsync0(() {}), | 417 test..expectListen() |
436 onCancel: expectAsync0(() {}) | 418 ..expectData(42) |
437 ); | 419 ..expectDone() |
438 Stream<int> s = c.stream; | 420 ..expectCancel(test.terminate); |
439 s.listen(expectAsync1((x) { expect(x, equals(42)); })); | 421 test..listen() |
440 c.add(42); | 422 ..add(42) |
441 c.close(); | 423 ..close(); |
442 }); | 424 }); |
443 | 425 |
444 test("broadcast-controller-listen-twice", () { | 426 test("broadcast-controller-listen-twice", () { |
445 StreamController<int> c = new StreamController.broadcast( | 427 StreamProtocolTest test = new StreamProtocolTest.broadcast(); |
446 onListen: expectAsync0(() {}), | 428 test.expectListen(); |
447 onCancel: expectAsync0(() {}) | 429 var sub1 = test.listen(); |
448 ); | 430 sub1.expectData(42, () { |
449 c.stream.listen(expectAsync1((x) { expect(x, equals(42)); }, count: 2)); | 431 var sub2 = test.listen(); |
450 c.add(42); | 432 sub1.expectData(37); |
451 c.stream.listen(expectAsync1((x) { expect(x, equals(42)); })); | 433 sub2.expectData(37); |
452 c.add(42); | 434 sub1.expectDone(); |
453 c.close(); | 435 sub2.expectDone(); |
436 test.expectCancel(test.terminate); | |
437 | |
438 test.add(37); | |
439 test.close(); | |
440 }); | |
441 test.add(42); | |
454 }); | 442 }); |
455 | 443 |
456 test("broadcast-controller-listen-twice-non-overlap", () { | 444 test("broadcast-controller-listen-twice-non-overlap", () { |
457 StreamController<int> c = new StreamController.broadcast( | 445 StreamProtocolTest test = new StreamProtocolTest.broadcast(); |
458 onListen: expectAsync0(() {}, count: 2), | 446 test |
459 onCancel: expectAsync0(() {}, count: 2) | 447 ..expectListen(() { |
460 ); | 448 test.add(42); |
461 var sub = c.stream.listen(expectAsync1((x) { expect(x, equals(42)); })); | 449 }) |
462 c.add(42); | 450 ..expectData(42, () { |
463 sub.cancel(); | 451 test.cancel(); |
464 c.stream.listen(expectAsync1((x) { expect(x, equals(42)); })); | 452 }) |
465 c.add(42); | 453 ..expectCancel(() { |
466 c.close(); | 454 test.listen(); |
455 })..expectListen(() { | |
456 test.add(37); | |
457 }) | |
458 ..expectData(37, () { | |
459 test.close(); | |
460 }) | |
461 ..expectDone() | |
462 ..expectCancel(test.terminate); | |
463 test.listen(); | |
467 }); | 464 }); |
468 | 465 |
469 test("broadcast-controller-individual-pause", () { | 466 test("broadcast-controller-individual-pause", () { |
470 StreamController<int> c = new StreamController.broadcast( | 467 StreamProtocolTest test = new StreamProtocolTest.broadcast(); |
471 onListen: expectAsync0(() {}), | 468 test.expectListen(); |
472 onCancel: expectAsync0(() {}) | 469 var sub1 = test.listen(); |
473 ); | 470 var sub2 = test.listen(); |
474 var sub1 = c.stream.listen(expectAsync1((x) { expect(x, equals(42)); })); | 471 var sub3; |
475 var sub2 = c.stream.listen(expectAsync1((x) { expect(x, equals(42)); }, | 472 sub1.expectData(42, sub1.pause); |
476 count: 3)); | 473 sub2.expectData(42); |
477 c.add(42); | 474 sub2.expectData(43, () { |
478 sub1.pause(); | 475 sub1.cancel(); |
479 c.add(42); | 476 sub3 = test.listen(); |
480 sub1.cancel(); | 477 sub2.expectData(44); |
481 var sub3 = c.stream.listen(expectAsync1((x) { expect(x, equals(42)); })); | 478 sub3.expectData(44, test.terminate); |
482 c.add(42); | 479 }); |
483 c.close(); | 480 test.add(42); |
481 test.add(43); | |
482 test.add(44); | |
484 }); | 483 }); |
485 | 484 |
486 test("broadcast-controller-add-in-callback", () { | 485 test("broadcast-controller-add-in-callback", () { |
487 StreamController<int> c; | 486 StreamProtocolTest test = new StreamProtocolTest.broadcast(); |
488 c = new StreamController( | 487 test.expectListen(); |
489 onListen: expectAsync0(() {}), | 488 var sub = test.listen(); |
490 onCancel: expectAsync0(() { | 489 test.add(42); |
491 c.add(42); | 490 sub.expectData(42, () { |
492 }) | 491 test.add(87); |
493 ); | |
494 var sub; | |
495 sub = c.stream.asBroadcastStream().listen(expectAsync1((v) { | |
496 Expect.equals(37, v); | |
497 c.add(21); | |
498 sub.cancel(); | 492 sub.cancel(); |
499 })); | 493 }); |
500 c.add(37); // Triggers listener, which adds 21 and removes itself. | 494 test.expectCancel(() { |
501 // Removing listener triggers onCancel which adds another 42. | 495 test.add(37); |
502 // Both 21 and 42 are lost because there are no listeners. | 496 test.terminate(); |
497 }); | |
503 }); | 498 }); |
504 } | 499 } |
505 | 500 |
506 main() { | 501 main() { |
507 testController(); | 502 testController(); |
508 testSingleController(); | 503 testSingleController(); |
509 testExtraMethods(); | 504 testExtraMethods(); |
510 testPause(); | 505 testPause(); |
511 testRethrow(); | 506 testRethrow(); |
512 testBroadcastController(); | 507 testBroadcastController(); |
513 } | 508 } |
OLD | NEW |