Fix bug in subscription handling in mime package

pkg/mime/test/mime_multipart_transformer_test.dart

 // Copyright (c) 2013, the Dart project authors.  Please see the AUTHORS file
 // for details. All rights reserved. Use of this source code is governed by a
 // BSD-style license that can be found in the LICENSE file.
 
 import 'dart:async';
 import 'dart:math';
 
 import "package:unittest/unittest.dart";
 import "package:mime/mime.dart";
 
 void _writeInChunks(List<int> data,
                     int chunkSize,
                     StreamController<List<int>> controller) {
   if (chunkSize == -1) chunkSize = data.length;
 
   int written = 0;
   for (int pos = 0; pos < data.length; pos += chunkSize) {
     int remaining = data.length - pos;
     int writeLength = min(chunkSize, remaining);
     controller.add(data.sublist(pos, pos + writeLength));
     written += writeLength;
   }
   controller.close();
 }
 
 
-void _testParse(String message,
-               String boundary,
-               [List<Map> expectedHeaders,
-                List expectedParts,
-                bool expectError = false]) {
+enum TestMode {
+  IMMEDIATE_LISTEN,
+  DELAY_LISTEN,
+  PAUSE_RESUME
+}
+
+void _runParseTest(String message,
+                   String boundary,
+                   TestMode mode,
+                   [List<Map> expectedHeaders,
+                    List expectedParts,
+                    bool expectError = false]) {
   Future testWrite(List<int> data, [int chunkSize = -1]) {
     StreamController controller = new StreamController(sync: true);
 
     var stream = controller.stream.transform(
         new MimeMultipartTransformer(boundary));
     int i = 0;
     var completer = new Completer();
     var futures = [];
     stream.listen((multipart) {
       int part = i++;
       if (expectedHeaders != null) {
         expect(multipart.headers, equals(expectedHeaders[part]));
       }
-      futures.add(multipart.fold([], (buffer, data) => buffer..addAll(data))
-          .then((data) {
-            if (expectedParts[part] != null) {
-              expect(data, equals(expectedParts[part].codeUnits));
-            }
+      switch (mode) {
+        case TestMode.IMMEDIATE_LISTEN:
+          futures.add(multipart.fold([], (buffer, data) => buffer..addAll(data))
+              .then((data) {
+                if (expectedParts[part] != null) {
+                  expect(data, equals(expectedParts[part].codeUnits));
+                }
+              }));
+          break;
+
+        case TestMode.DELAY_LISTEN:
+          futures.add(new Future(() {
+            return multipart.fold([], (buffer, data) => buffer..addAll(data))
+                .then((data) {
+                  if (expectedParts[part] != null) {
+                    expect(data, equals(expectedParts[part].codeUnits));
+                  }
+                });
           }));
+          break;
+
+        case TestMode.PAUSE_RESUME:
+          var completer = new Completer();
+          futures.add(completer.future);
+          var buffer = [];
+          var subscription;
+          subscription = multipart.listen(
+              (data) {
+                buffer.addAll(data);
+                subscription.pause();
+                new Future(() => subscription.resume());
+              },
+              onDone: () {
+                if (expectedParts[part] != null) {
+                  expect(buffer, equals(expectedParts[part].codeUnits));
+                }
+                completer.complete();
+              });
+          break;
+      }
     }, onError: (error) {
       if (!expectError) throw error;
     }, onDone: () {
       if (expectedParts != null) {
         expect(i, equals(expectedParts.length));
       }
       Future.wait(futures).then(completer.complete);
     });
 
     _writeInChunks(data, chunkSize, controller);
@@ skipping 94 matching lines @@
       expect(Future.wait([
         testCompletePartAfterCancel(data, numPartsExpected),
         testCompletePartAfterCancel(data, numPartsExpected, 10),
         testCompletePartAfterCancel(data, numPartsExpected, 2),
         testCompletePartAfterCancel(data, numPartsExpected, 1),
       ]), completes);
     });
   }
 }
 
+void _testParse(String message,
+                String boundary,
+               [List<Map> expectedHeaders,
+                List expectedParts,
+                bool expectError = false]) {
+  _runParseTest(
+      message, boundary, TestMode.IMMEDIATE_LISTEN,
+      expectedHeaders, expectedParts, expectError);
+  _runParseTest(
+      message, boundary, TestMode.DELAY_LISTEN,
+      expectedHeaders, expectedParts, expectError);
+  _runParseTest(
+      message, boundary, TestMode.PAUSE_RESUME,
+      expectedHeaders, expectedParts, expectError);
+}
+
 void _testParseValid() {
   // Empty message from Chrome form post.
   var message = '------WebKitFormBoundaryU3FBruSkJKG0Yor1--\r\n';
   _testParse(message, "----WebKitFormBoundaryU3FBruSkJKG0Yor1", [], []);
 
   // Sample from Wikipedia.
   message = """
 This is a message with multiple parts in MIME format.\r
 --frontier\r
 Content-Type: text/plain\r
@@ skipping 224 matching lines @@
 \r
 Body2\r
 --xxx\r\n""";
   _testParse(message, "xxx", null, [null, null], true);
 }
 
 void main() {
   _testParseValid();
   _testParseInvalid();
 }