| OLD | NEW |
| 1 // Copyright (c) 2014, the Dart project authors. Please see the AUTHORS file | 1 // Copyright (c) 2014, 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 import "package:expect/expect.dart"; | 6 import "package:expect/expect.dart"; |
| 7 import "package:async_helper/async_helper.dart"; | 7 import "package:async_helper/async_helper.dart"; |
| 8 | 8 |
| 9 class Trace { | 9 class Trace { |
| 10 String trace; | 10 String trace = ""; |
| 11 Trace(this.trace); | 11 record(x) { |
| 12 void record(x) { | |
| 13 trace += x.toString(); | 12 trace += x.toString(); |
| 14 } | 13 } |
| 15 String toString() => trace; | 14 toString() => trace; |
| 16 } | 15 } |
| 17 | 16 |
| 18 | 17 |
| 19 Stream makeMeAStream() { | 18 Stream makeMeAStream() { |
| 20 return timedCounter(5); | 19 return timedCounter(5); |
| 21 } | 20 } |
| 22 | 21 |
| 22 Trace t1 = new Trace(); |
| 23 | 23 |
| 24 consumeOne(trace) async { | 24 consumeOne() async { |
| 25 // Equivalent to await for (x in makeMeAStream()) { ... } | 25 // Equivalent to await for (x in makeMeAStream()) { ... } |
| 26 var s = makeMeAStream(); | 26 var s = makeMeAStream(); |
| 27 var it = new StreamIterator(s); | 27 var it = new StreamIterator(s); |
| 28 while (await it.moveNext()) { | 28 while (await it.moveNext()) { |
| 29 var x = it.current; | 29 var x = it.current; |
| 30 trace.record(x); | 30 t1.record(x); |
| 31 } | 31 } |
| 32 trace.record("X"); | 32 t1.record("X"); |
| 33 } | 33 } |
| 34 | 34 |
| 35 Trace t2 = new Trace(); |
| 35 | 36 |
| 36 consumeTwo(trace) async { | 37 consumeTwo() async { |
| 37 await for (var x in makeMeAStream()) { | 38 await for (var x in makeMeAStream()) { |
| 38 trace.record(x); | 39 t2.record(x); |
| 39 } | 40 } |
| 40 trace.record("Y"); | 41 t2.record("Y"); |
| 41 } | 42 } |
| 42 | 43 |
| 44 Trace t3 = new Trace(); |
| 43 | 45 |
| 44 consumeNested(trace) async { | 46 consumeNested() async { |
| 45 await for (var x in makeMeAStream()) { | 47 await for (var x in makeMeAStream()) { |
| 46 trace.record(x); | 48 t3.record(x); |
| 47 await for (var y in makeMeAStream()) { | 49 await for (var y in makeMeAStream()) { |
| 48 trace.record(y); | 50 t3.record(y); |
| 49 } | 51 } |
| 50 trace.record("|"); | 52 t3.record("|"); |
| 51 } | 53 } |
| 52 trace.record("Z"); | 54 t3.record("Z"); |
| 53 } | 55 } |
| 54 | 56 |
| 55 consumeSomeOfInfinite(trace) async { | 57 Trace t4 = new Trace(); |
| 58 |
| 59 consumeSomeOfInfinite() async { |
| 56 int i = 0; | 60 int i = 0; |
| 57 await for (var x in infiniteStream()) { | 61 await for (var x in infiniteStream()) { |
| 58 i++; | 62 i++; |
| 59 if (i > 10) break; | 63 if (i > 10) break; |
| 60 trace.record(x); | 64 t4.record(x); |
| 61 } | 65 } |
| 62 trace.record("U"); | 66 t4.record("U"); |
| 63 } | |
| 64 | |
| 65 const String cancelError = | |
| 66 "ERROR: Error in future returned by .cancel() must be caught"; | |
| 67 | |
| 68 /// Creates a stream that yields integers forever, but throws when canceled. | |
| 69 /// | |
| 70 /// The thrown error should end up in the future returned by `cancel`. | |
| 71 Stream<int> errorOnCancelStream(int n) async* { | |
| 72 try { | |
| 73 while (true) yield n++; | |
| 74 } finally { | |
| 75 throw cancelError; | |
| 76 } | |
| 77 } | |
| 78 | |
| 79 | |
| 80 // Sanity-check that the errorOnCancelStream behaves as expected. | |
| 81 testErrorOnCancel() { | |
| 82 var stream = errorOnCancelStream(0); | |
| 83 var subscription = stream.listen(null); | |
| 84 return subscription.cancel().then((_) { | |
| 85 Expect.fail("Cancel future did not contain error"); | |
| 86 }, onError: (e) { | |
| 87 Expect.equals(cancelError, e); | |
| 88 }); | |
| 89 } | |
| 90 | |
| 91 testCancelAwaited() async { | |
| 92 return runZoned(() async { | |
| 93 var stream = errorOnCancelStream(0); | |
| 94 try { | |
| 95 var n = 0; | |
| 96 await for (var x in stream) { | |
| 97 Expect.equals(n++, x); | |
| 98 if (x == 5) break; | |
| 99 } | |
| 100 Expect.fail("Didn't await the cancel future."); | |
| 101 } on String catch (e) { | |
| 102 Expect.equals(cancelError, e); | |
| 103 } | |
| 104 }, onError: (e) { | |
| 105 // Catch the error if it's uncaught. | |
| 106 if (cancelError == e) { | |
| 107 Expect.fail("Error in cancel is considered uncaught"); | |
| 108 } | |
| 109 throw e; | |
| 110 }); | |
| 111 } | 67 } |
| 112 | 68 |
| 113 main() { | 69 main() { |
| 114 Trace t1 = new Trace("T1:"); | 70 var f1 = consumeOne(); |
| 115 var f1 = consumeOne(t1); | 71 t1.record("T1:"); |
| 116 | 72 |
| 117 Trace t2 = new Trace("T2:"); | 73 var f2 = consumeTwo(); |
| 118 var f2 = consumeTwo(t2); | 74 t2.record("T2:"); |
| 119 | 75 |
| 120 Trace t3 = new Trace("T3:"); | 76 var f3 = consumeNested(); |
| 121 var f3 = consumeNested(t3); | 77 t3.record("T3:"); |
| 122 | 78 |
| 123 Trace t4 = new Trace("T4:"); | 79 var f4 = consumeSomeOfInfinite(); |
| 124 var f4 = consumeSomeOfInfinite(t4); | 80 t4.record("T4:"); |
| 125 | |
| 126 var f5 = testErrorOnCancel(); | |
| 127 | |
| 128 var f6 = testCancelAwaited(); | |
| 129 | 81 |
| 130 asyncStart(); | 82 asyncStart(); |
| 131 Future.wait([f1, f2, f3, f4, f5, f6]).then((_) { | 83 Future.wait([f1, f2, f3, f4]).then((_) { |
| 132 Expect.equals("T1:12345X", t1.toString()); | 84 Expect.equals("T1:12345X", t1.toString()); |
| 133 Expect.equals("T2:12345Y", t2.toString()); | 85 Expect.equals("T2:12345Y", t2.toString()); |
| 134 Expect.equals("T3:112345|212345|312345|412345|512345|Z", t3.toString()); | 86 Expect.equals("T3:112345|212345|312345|412345|512345|Z", t3.toString()); |
| 135 Expect.equals("T4:12345678910U", t4.toString()); | 87 Expect.equals("T4:12345678910U", t4.toString()); |
| 136 asyncEnd(); | 88 asyncEnd(); |
| 137 }); | 89 }); |
| 138 } | 90 } |
| 139 | 91 |
| 140 // Create a stream that produces numbers [1, 2, ... maxCount] | 92 // Create a stream that produces numbers [1, 2, ... maxCount] |
| 141 Stream timedCounter(int maxCount) { | 93 Stream timedCounter(int maxCount) { |
| (...skipping 53 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... |
| 195 } | 147 } |
| 196 | 148 |
| 197 controller = new StreamController( | 149 controller = new StreamController( |
| 198 onListen: startTimer, | 150 onListen: startTimer, |
| 199 onPause: stopTimer, | 151 onPause: stopTimer, |
| 200 onResume: startTimer, | 152 onResume: startTimer, |
| 201 onCancel: stopTimer); | 153 onCancel: stopTimer); |
| 202 | 154 |
| 203 return controller.stream; | 155 return controller.stream; |
| 204 } | 156 } |
| OLD | NEW |