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

Side by Side Diff: tests/lib/async/slow_consumer2_test.dart

Issue 12213092: Rework Timer interface. (Closed) Base URL: https://dart.googlecode.com/svn/branches/bleeding_edge/dart
Patch Set: Address comments. Created 7 years, 10 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 | Annotate | Revision Log
« no previous file with comments | « tests/lib/async/future_test.dart ('k') | tests/lib/async/slow_consumer3_test.dart » ('j') | no next file with comments »
Toggle Intra-line Diffs ('i') | Expand Comments ('e') | Collapse Comments ('c') | Show Comments Hide Comments ('s')
OLDNEW
1 // Copyright (c) 2012, the Dart project authors. Please see the AUTHORS file 1 // Copyright (c) 2012, 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 // VMOptions=--old_gen_heap_size=32 5 // VMOptions=--old_gen_heap_size=32
6 6
7 library slow_consumer2_test; 7 library slow_consumer2_test;
8 8
9 import 'dart:async'; 9 import 'dart:async';
10 import 'dart:isolate'; 10 import 'dart:isolate';
(...skipping 17 matching lines...) Expand all
28 subscription = stream.listen( 28 subscription = stream.listen(
29 (List<int> data) { 29 (List<int> data) {
30 receivedCount += data.length; 30 receivedCount += data.length;
31 usedBufferSize += data.length; 31 usedBufferSize += data.length;
32 bufferedData.add(data); 32 bufferedData.add(data);
33 int currentBufferedDataLength = bufferedData.length; 33 int currentBufferedDataLength = bufferedData.length;
34 if (usedBufferSize > bufferSize) { 34 if (usedBufferSize > bufferSize) {
35 subscription.pause(); 35 subscription.pause();
36 usedBufferSize = 0; 36 usedBufferSize = 0;
37 int ms = data.length * 1000 ~/ bytesPerSecond; 37 int ms = data.length * 1000 ~/ bytesPerSecond;
38 new Timer(ms, (_) { 38 Duration duration = new Duration(milliseconds: ms);
39 new Timer(duration, () {
39 for (int i = 0; i < currentBufferedDataLength; i++) { 40 for (int i = 0; i < currentBufferedDataLength; i++) {
40 bufferedData[i] = null; 41 bufferedData[i] = null;
41 } 42 }
42 subscription.resume(); 43 subscription.resume();
43 }); 44 });
44 } 45 }
45 }, 46 },
46 onDone: () { result.complete(receivedCount); }); 47 onDone: () { result.complete(receivedCount); });
47 return result.future; 48 return result.future;
48 } 49 }
49 } 50 }
50 51
51 class DataProvider { 52 class DataProvider {
52 final int chunkSize; 53 final int chunkSize;
53 final int bytesPerSecond; 54 final int bytesPerSecond;
54 int sentCount = 0; 55 int sentCount = 0;
55 int targetCount; 56 int targetCount;
56 StreamController controller; 57 StreamController controller;
57 58
58 DataProvider(int this.bytesPerSecond, int this.targetCount, this.chunkSize) { 59 DataProvider(int this.bytesPerSecond, int this.targetCount, this.chunkSize) {
59 controller = new StreamController(onPauseStateChange: onPauseStateChange); 60 controller = new StreamController(onPauseStateChange: onPauseStateChange);
60 new Timer(0, (_) => send()); 61 Timer.run(send);
61 } 62 }
62 63
63 Stream get stream => controller.stream; 64 Stream get stream => controller.stream;
64 65
65 send() { 66 send() {
66 if (controller.isPaused) return; 67 if (controller.isPaused) return;
67 if (sentCount == targetCount) { 68 if (sentCount == targetCount) {
68 controller.close(); 69 controller.close();
69 return; 70 return;
70 } 71 }
71 int listSize = chunkSize; 72 int listSize = chunkSize;
72 sentCount += listSize; 73 sentCount += listSize;
73 if (sentCount > targetCount) { 74 if (sentCount > targetCount) {
74 listSize -= sentCount - targetCount; 75 listSize -= sentCount - targetCount;
75 sentCount = targetCount; 76 sentCount = targetCount;
76 } 77 }
77 controller.add(new List.fixedLength(listSize)); 78 controller.add(new List.fixedLength(listSize));
78 int ms = listSize * 1000 ~/ bytesPerSecond; 79 int ms = listSize * 1000 ~/ bytesPerSecond;
79 if (!controller.isPaused) new Timer(ms, (_) => send()); 80 Duration duration = new Duration(milliseconds: ms);
81 if (!controller.isPaused) new Timer(duration, send);
80 } 82 }
81 83
82 onPauseStateChange() { 84 onPauseStateChange() {
83 // We don't care if we just unpaused or paused. In either case we just 85 // We don't care if we just unpaused or paused. In either case we just
84 // call send which will test it for us. 86 // call send which will test it for us.
85 send(); 87 send();
86 } 88 }
87 } 89 }
88 90
89 main() { 91 main() {
90 var port = new ReceivePort(); 92 var port = new ReceivePort();
91 // The data provider can deliver 800MB/s of data. It sends 100MB of data to 93 // The data provider can deliver 800MB/s of data. It sends 100MB of data to
92 // the slower consumer who can only read 200MB/s. The data is sent in 1MB 94 // the slower consumer who can only read 200MB/s. The data is sent in 1MB
93 // chunks. The consumer has a buffer of 5MB. That is, it can accept a few 95 // chunks. The consumer has a buffer of 5MB. That is, it can accept a few
94 // packages without pausing its input. 96 // packages without pausing its input.
95 // 97 //
96 // This test is limited to 32MB of heap-space (see VMOptions on top of the 98 // This test is limited to 32MB of heap-space (see VMOptions on top of the
97 // file). If the consumer doesn't pause the data-provider it will run out of 99 // file). If the consumer doesn't pause the data-provider it will run out of
98 // heap-space. 100 // heap-space.
99 101
100 new DataProvider(800 * MB, 100 * MB, 1 * MB).stream 102 new DataProvider(800 * MB, 100 * MB, 1 * MB).stream
101 .pipe(new SlowConsumer(200 * MB, 5 * MB)) 103 .pipe(new SlowConsumer(200 * MB, 5 * MB))
102 .then((count) { 104 .then((count) {
103 port.close(); 105 port.close();
104 Expect.equals(100 * MB, count); 106 Expect.equals(100 * MB, count);
105 }); 107 });
106 } 108 }
OLDNEW
« no previous file with comments | « tests/lib/async/future_test.dart ('k') | tests/lib/async/slow_consumer3_test.dart » ('j') | no next file with comments »

Powered by Google App Engine
This is Rietveld 408576698