Chromium Code Reviews
chromiumcodereview-hr@appspot.gserviceaccount.com (chromiumcodereview-hr) | Please choose your nickname with Settings | Help | Chromium Project | Gerrit Changes | Sign out
(208)

Side by Side Diff: tests/lib_strong/async/stream_event_transformed_test.dart

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

Powered by Google App Engine
This is Rietveld 408576698