tests/lib_strong/async/slow_consumer2_test.dart - Issue 3001793002: Migrated test block 171 to Dart 2.0.

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

Issue 3001793002: Migrated test block 171 to Dart 2.0. (Closed)

Patch Set: Created 3 years, 4 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:

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_consumer2_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 class DataProvider {

64 final int chunkSize;

65 final int bytesPerSecond;

66 int sentCount = 0;

67 int targetCount;

68 StreamController controller;

69

70 DataProvider(int this.bytesPerSecond, int this.targetCount, this.chunkSize) {

71 controller = new StreamController(

72 sync: true, onPause: onPauseStateChange, onResume: onPauseStateChange);

73 Timer.run(send);

74 }

75

76 Stream get stream => controller.stream;

77

78 send() {

79 if (controller.isPaused) return;

80 if (sentCount == targetCount) {

81 controller.close();

82 return;

83 }

84 int listSize = chunkSize;

85 sentCount += listSize;

86 if (sentCount > targetCount) {

87 listSize -= sentCount - targetCount;

88 sentCount = targetCount;

89 }

90 controller.add(new List(listSize));

91 int ms = listSize * 1000 ~/ bytesPerSecond;

92 Duration duration = new Duration(milliseconds: ms);

93 if (!controller.isPaused) new Timer(duration, send);

94 }

95

96 onPauseStateChange() {

97 // We don't care if we just unpaused or paused. In either case we just

98 // call send which will test it for us.

99 send();

100 }

101 }

102

103 main() {

104 asyncStart();

105 // The data provider can deliver 800MB/s of data. It sends 100MB of data to

106 // the slower consumer who can only read 200MB/s. The data is sent in 1MB

107 // chunks. The consumer has a buffer of 5MB. That is, it can accept a few

108 // packages without pausing its input.

109 //

110 // This test is limited to 32MB of heap-space (see VMOptions on top of the

111 // file). If the consumer doesn't pause the data-provider it will run out of

112 // heap-space.

113

114 new DataProvider(800 * MB, 100 * MB, 1 * MB)

115 .stream

116 .pipe(new SlowConsumer(200 * MB, 5 * MB))

117 .then((count) {

118 Expect.equals(100 * MB, count);

119 asyncEnd();

120 });

121 }

OLD	NEW