| OLD | NEW |
| (Empty) |
| 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 | |
| 3 // BSD-style license that can be found in the LICENSE file. | |
| 4 | |
| 5 // VMOptions=--old_gen_heap_size=64 | |
| 6 | |
| 7 library slow_consumer3_test; | |
| 8 | |
| 9 import 'package:async_helper/async_helper.dart'; | |
| 10 import "package:expect/expect.dart"; | |
| 11 import 'dart:async'; | |
| 12 | |
| 13 const int KB = 1024; | |
| 14 const int MB = KB * KB; | |
| 15 const int GB = KB * KB * KB; | |
| 16 | |
| 17 class SlowConsumer extends StreamConsumer { | |
| 18 int receivedCount = 0; | |
| 19 final int bytesPerSecond; | |
| 20 final int bufferSize; | |
| 21 final List bufferedData = []; | |
| 22 int usedBufferSize = 0; | |
| 23 int finalCount; | |
| 24 | |
| 25 SlowConsumer(int this.bytesPerSecond, int this.bufferSize); | |
| 26 | |
| 27 Future consume(Stream stream) { | |
| 28 return addStream(stream).then((_) => close()); | |
| 29 } | |
| 30 | |
| 31 Future addStream(Stream stream) { | |
| 32 Completer result = new Completer(); | |
| 33 var subscription; | |
| 34 subscription = stream.listen((List<int> data) { | |
| 35 receivedCount += data.length; | |
| 36 usedBufferSize += data.length; | |
| 37 bufferedData.add(data); | |
| 38 int currentBufferedDataLength = bufferedData.length; | |
| 39 if (usedBufferSize > bufferSize) { | |
| 40 subscription.pause(); | |
| 41 usedBufferSize = 0; | |
| 42 int ms = data.length * 1000 ~/ bytesPerSecond; | |
| 43 Duration duration = new Duration(milliseconds: ms); | |
| 44 new Timer(duration, () { | |
| 45 for (int i = 0; i < currentBufferedDataLength; i++) { | |
| 46 bufferedData[i] = null; | |
| 47 } | |
| 48 subscription.resume(); | |
| 49 }); | |
| 50 } | |
| 51 }, onDone: () { | |
| 52 finalCount = receivedCount; | |
| 53 result.complete(receivedCount); | |
| 54 }); | |
| 55 return result.future; | |
| 56 } | |
| 57 | |
| 58 Future close() { | |
| 59 return new Future.value(finalCount); | |
| 60 } | |
| 61 } | |
| 62 | |
| 63 Stream<List> dataGenerator(int bytesTotal, int chunkSize) { | |
| 64 int chunks = bytesTotal ~/ chunkSize; | |
| 65 return new Stream.fromIterable(new Iterable.generate(chunks, (_) { | |
| 66 // This assumes one byte per entry. In practice it will be more. | |
| 67 return new List<int>(chunkSize); | |
| 68 })); | |
| 69 } | |
| 70 | |
| 71 main() { | |
| 72 asyncStart(); | |
| 73 // The data provider can deliver 800MBs of data as fast as it is | |
| 74 // requested. The data is sent in 0.5MB chunks. The consumer has a buffer of | |
| 75 // 3MB. That is, it can accept a few packages without pausing its input. | |
| 76 // | |
| 77 // Notice that we aren't really counting bytes, but words, since we use normal | |
| 78 // lists where each entry takes up a full word. In 64-bit VMs this will be | |
| 79 // 8 bytes per entry, so the 3*MB buffer is picked to stay below 32 actual | |
| 80 // MiB. | |
| 81 // | |
| 82 // This test is limited to 32MB of heap-space (see VMOptions on top of the | |
| 83 // file). If the consumer doesn't pause the data-provider it will run out of | |
| 84 // heap-space. | |
| 85 | |
| 86 dataGenerator(100 * MB, 512 * KB) | |
| 87 .pipe(new SlowConsumer(200 * MB, 3 * MB)) | |
| 88 .then((count) { | |
| 89 Expect.equals(100 * MB, count); | |
| 90 asyncEnd(); | |
| 91 }); | |
| 92 } | |
| OLD | NEW |