OLD | NEW |
1 // Copyright (c) 2015, the Dart project authors. Please see the AUTHORS file | 1 // Copyright (c) 2015, 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 filevents. | 3 // BSD-style license that can be found in the LICENSE filevents. |
4 | 4 |
5 import "dart:async"; | 5 import "dart:async"; |
6 | 6 |
7 import "package:async/async.dart"; | 7 import "package:async/async.dart"; |
8 import "package:test/test.dart"; | 8 import "package:test/test.dart"; |
9 | 9 |
10 import "utils.dart"; | 10 import "utils.dart"; |
(...skipping 65 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... |
76 expect(events.eventsDispatched, equals(2)); | 76 expect(events.eventsDispatched, equals(2)); |
77 }); | 77 }); |
78 | 78 |
79 test("doesn't increment for rest requests", () async { | 79 test("doesn't increment for rest requests", () async { |
80 var events = new StreamQueue<int>(createStream()); | 80 var events = new StreamQueue<int>(createStream()); |
81 await events.rest.toList(); | 81 await events.rest.toList(); |
82 expect(events.eventsDispatched, equals(0)); | 82 expect(events.eventsDispatched, equals(0)); |
83 }); | 83 }); |
84 }); | 84 }); |
85 | 85 |
| 86 group("lookAhead operation", () { |
| 87 test("as simple list of events", () async { |
| 88 var events = new StreamQueue<int>(createStream()); |
| 89 expect(await events.lookAhead(4), [1, 2, 3, 4]); |
| 90 expect(await events.next, 1); |
| 91 expect(await events.lookAhead(2), [2, 3]); |
| 92 expect(await events.take(2), [2, 3]); |
| 93 expect(await events.next, 4); |
| 94 await events.cancel(); |
| 95 }); |
| 96 |
| 97 test("of 0 events", () async { |
| 98 var events = new StreamQueue<int>(createStream()); |
| 99 expect(events.lookAhead(0), completion([])); |
| 100 expect(events.next, completion(1)); |
| 101 expect(events.lookAhead(0), completion([])); |
| 102 expect(events.next, completion(2)); |
| 103 expect(events.lookAhead(0), completion([])); |
| 104 expect(events.next, completion(3)); |
| 105 expect(events.lookAhead(0), completion([])); |
| 106 expect(events.next, completion(4)); |
| 107 expect(events.lookAhead(0), completion([])); |
| 108 expect(events.lookAhead(5), completion([])); |
| 109 expect(events.next, throwsStateError); |
| 110 await events.cancel(); |
| 111 }); |
| 112 |
| 113 test("with bad arguments throws", () async { |
| 114 var events = new StreamQueue<int>(createStream()); |
| 115 expect(() => events.lookAhead(-1), throwsArgumentError); |
| 116 expect(await events.next, 1); // Did not consume event. |
| 117 expect(() => events.lookAhead(-1), throwsArgumentError); |
| 118 expect(await events.next, 2); // Did not consume event. |
| 119 await events.cancel(); |
| 120 }); |
| 121 |
| 122 test("of too many arguments", () async { |
| 123 var events = new StreamQueue<int>(createStream()); |
| 124 expect(await events.lookAhead(6), [1, 2, 3, 4]); |
| 125 await events.cancel(); |
| 126 }); |
| 127 |
| 128 test("too large later", () async { |
| 129 var events = new StreamQueue<int>(createStream()); |
| 130 expect(await events.next, 1); |
| 131 expect(await events.next, 2); |
| 132 expect(await events.lookAhead(6), [3, 4]); |
| 133 await events.cancel(); |
| 134 }); |
| 135 |
| 136 test("error", () async { |
| 137 var events = new StreamQueue<int>(createErrorStream()); |
| 138 expect(events.lookAhead(4), throwsA("To err is divine!")); |
| 139 expect(events.take(4), throwsA("To err is divine!")); |
| 140 expect(await events.next, 4); |
| 141 await events.cancel(); |
| 142 }); |
| 143 }); |
| 144 |
86 group("next operation", () { | 145 group("next operation", () { |
87 test("simple sequence of requests", () async { | 146 test("simple sequence of requests", () async { |
88 var events = new StreamQueue<int>(createStream()); | 147 var events = new StreamQueue<int>(createStream()); |
89 for (int i = 1; i <= 4; i++) { | 148 for (int i = 1; i <= 4; i++) { |
90 expect(await events.next, i); | 149 expect(await events.next, i); |
91 } | 150 } |
92 expect(events.next, throwsStateError); | 151 expect(events.next, throwsStateError); |
93 }); | 152 }); |
94 | 153 |
95 test("multiple requests at the same time", () async { | 154 test("multiple requests at the same time", () async { |
(...skipping 271 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... |
367 await flushMicrotasks(); | 426 await flushMicrotasks(); |
368 expect(lastEvent, 3); | 427 expect(lastEvent, 3); |
369 | 428 |
370 var cancelFuture = subscription.cancel(); | 429 var cancelFuture = subscription.cancel(); |
371 expect(controller.hasListener, isFalse); | 430 expect(controller.hasListener, isFalse); |
372 cancel.complete(42); | 431 cancel.complete(42); |
373 expect(cancelFuture, completion(42)); | 432 expect(cancelFuture, completion(42)); |
374 }); | 433 }); |
375 }); | 434 }); |
376 | 435 |
| 436 group("peek operation", () { |
| 437 test("peeks one event", () async { |
| 438 var events = new StreamQueue<int>(createStream()); |
| 439 expect(await events.peek, 1); |
| 440 expect(await events.next, 1); |
| 441 expect(await events.peek, 2); |
| 442 expect(await events.take(2), [2, 3]); |
| 443 expect(await events.peek, 4); |
| 444 expect(await events.next, 4); |
| 445 // Throws at end. |
| 446 expect(events.peek, throws); |
| 447 await events.cancel(); |
| 448 }); |
| 449 test("multiple requests at the same time", () async { |
| 450 var events = new StreamQueue<int>(createStream()); |
| 451 var result = await Future.wait( |
| 452 [events.peek, events.peek, events.next, events.peek, events.peek]); |
| 453 expect(result, [1, 1, 1, 2, 2]); |
| 454 await events.cancel(); |
| 455 }); |
| 456 test("sequence of requests with error", () async { |
| 457 var events = new StreamQueue<int>(createErrorStream()); |
| 458 expect(await events.next, 1); |
| 459 expect(await events.next, 2); |
| 460 expect(events.peek, throwsA("To err is divine!")); |
| 461 // Error stays in queue. |
| 462 expect(events.peek, throwsA("To err is divine!")); |
| 463 expect(events.next, throwsA("To err is divine!")); |
| 464 expect(await events.next, 4); |
| 465 await events.cancel(); |
| 466 }); |
| 467 }); |
| 468 |
377 group("cancel operation", () { | 469 group("cancel operation", () { |
378 test("closes the events, prevents any other operation", () async { | 470 test("closes the events, prevents any other operation", () async { |
379 var events = new StreamQueue<int>(createStream()); | 471 var events = new StreamQueue<int>(createStream()); |
380 await events.cancel(); | 472 await events.cancel(); |
| 473 expect(() => events.lookAhead(1), throwsStateError); |
381 expect(() => events.next, throwsStateError); | 474 expect(() => events.next, throwsStateError); |
| 475 expect(() => events.peek, throwsStateError); |
382 expect(() => events.skip(1), throwsStateError); | 476 expect(() => events.skip(1), throwsStateError); |
383 expect(() => events.take(1), throwsStateError); | 477 expect(() => events.take(1), throwsStateError); |
384 expect(() => events.rest, throwsStateError); | 478 expect(() => events.rest, throwsStateError); |
385 expect(() => events.cancel(), throwsStateError); | 479 expect(() => events.cancel(), throwsStateError); |
386 }); | 480 }); |
387 | 481 |
388 test("cancels underlying subscription when called before any event", | 482 test("cancels underlying subscription when called before any event", |
389 () async { | 483 () async { |
390 var cancelFuture = new Future.value(42); | 484 var cancelFuture = new Future.value(42); |
391 var controller = new StreamController<int>(onCancel: () => cancelFuture); | 485 var controller = new StreamController<int>(onCancel: () => cancelFuture); |
(...skipping 616 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... |
1008 controller.add(4); | 1102 controller.add(4); |
1009 await flushMicrotasks(); | 1103 await flushMicrotasks(); |
1010 controller.close(); | 1104 controller.close(); |
1011 }(); | 1105 }(); |
1012 return controller.stream; | 1106 return controller.stream; |
1013 } | 1107 } |
1014 | 1108 |
1015 Stream<int> createLongStream(int eventCount) async* { | 1109 Stream<int> createLongStream(int eventCount) async* { |
1016 for (int i = 0; i < eventCount; i++) yield i; | 1110 for (int i = 0; i < eventCount; i++) yield i; |
1017 } | 1111 } |
OLD | NEW |