Let StreamSubscription.cancel return a Future.

sdk/lib/io/directory_impl.dart

Index: sdk/lib/io/directory_impl.dart
diff --git a/sdk/lib/io/directory_impl.dart b/sdk/lib/io/directory_impl.dart
index 5429a294ecc96ca0fb92e0f02f8c6ce5456d573e..e24636d5aacbfaa0a3d9e104cd5e38c6d2d16811 100644
--- a/sdk/lib/io/directory_impl.dart
+++ b/sdk/lib/io/directory_impl.dart
@@ -322,6 +322,9 @@ class _AsyncDirectoryLister {
   bool nextRunning = false;
   bool closed = false;
 
+  var pending;
+  var cancelCompleter;
+
   _AsyncDirectoryLister(String this.path,
                         bool this.recursive,
                         bool this.followLinks) {
@@ -348,15 +351,15 @@ class _AsyncDirectoryLister {
   }
 
   void onResume() {
-    if (!nextRunning) next();
+    handlePending();
   }
 
-  void onCancel() {
+  onCancel() {
     canceled = true;
     // If we are active, but not requesting, close.
-    if (!nextRunning) {
-      close();
-    }
+    cancelCompleter = new Completer();
+    close();
+    return cancelCompleter.future;
   }
 
   void next() {
@@ -364,49 +367,70 @@ class _AsyncDirectoryLister {
       close();
       return;
     }
+    if (nextRunning) return;
     if (id == null) return;
-    if (controller.isPaused) return;
-    assert(!nextRunning);
+    if (pending != null) return;
     nextRunning = true;
     _Directory._newServicePort().call([_Directory.LIST_NEXT_REQUEST, id])
         .then((result) {
-          if (result is List) {
-            assert(result.length % 2 == 0);
-            for (int i = 0; i < result.length; i++) {
-              assert(i % 2 == 0);
-              switch (result[i++]) {
-                case LIST_FILE:
-                  controller.add(new File(result[i]));
-                  break;
-                case LIST_DIRECTORY:
-                  controller.add(new Directory(result[i]));
-                  break;
-                case LIST_LINK:
-                  controller.add(new Link(result[i]));
-                  break;
-                case LIST_ERROR:
-                  error(result[i]);
-                  break;
-                case LIST_DONE:
-                  close();
-                  return;
-              }
-            }
-          } else {
-            controller.addError(new DirectoryException("Internal error"));
-          }
           nextRunning = false;
-          next();
+          pending = result;
+          handlePending();
         });
   }
 
+  void handlePending() {
+    if (canceled) {
+      close();
+      return;
+    }
+    if (pending != null) {
+      if (controller.isPaused) return;
+      var result = pending;
+      pending = null;
+      next();
+      if (result is List) {
+        assert(result.length % 2 == 0);
+        for (int i = 0; i < result.length; i++) {
+          assert(i % 2 == 0);
+          switch (result[i++]) {
+            case LIST_FILE:
+              controller.add(new File(result[i]));
+              break;
+            case LIST_DIRECTORY:
+              controller.add(new Directory(result[i]));
+              break;
+            case LIST_LINK:
+              controller.add(new Link(result[i]));
+              break;
+            case LIST_ERROR:
+              error(result[i]);
+              break;
+            case LIST_DONE:
+              close();
+              return;
+          }
+        }
+      } else {
+        controller.addError(new DirectoryException("Internal error"));
+      }
+    } else {
+      next();
+    }
+  }
+
   void close() {
+    if (nextRunning) return; // Wait for it to complete.
     if (closed) return;
     if (id == null) return;
     closed = true;
     _Directory._newServicePort().call([_Directory.LIST_STOP_REQUEST, id])
         .then((_) {
-          controller.close();
+          if (canceled) {
+            cancelCompleter.complete();
+          } else {
+            controller.close();
+          }
         });
   }