Let StreamSubscription.cancel return a Future.

sdk/lib/io/directory_impl.dart

 // Copyright (c) 2012, 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.
 
 part of dart.io;
 
 class _Directory implements Directory {
   static const CREATE_REQUEST = 0;
   static const DELETE_REQUEST = 1;
   static const EXISTS_REQUEST = 2;
@@ skipping 304 matching lines @@
   final String path;
   final bool recursive;
   final bool followLinks;
 
   StreamController controller;
   int id;
   bool canceled = false;
   bool nextRunning = false;
   bool closed = false;
 
+  var pending;
+  var cancelCompleter;
+
   _AsyncDirectoryLister(String this.path,
                         bool this.recursive,
                         bool this.followLinks) {
     controller = new StreamController(onListen: onListen,
                                       onResume: onResume,
                                       onCancel: onCancel,
                                       sync: true);
   }
 
   Stream get stream => controller.stream;
 
   void onListen() {
     var request = [_Directory.LIST_START_REQUEST, path, recursive, followLinks];
     _Directory._newServicePort().call(request)
         .then((response) {
           if (response is int) {
             id = response;
             next();
           } else {
             error(response);
             controller.close();
           }
         });
   }
 
   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() {
     if (canceled) {
       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();
+          }
         });
   }
 
   void error(message) {
     var errorType =
         message[RESPONSE_ERROR][_ERROR_RESPONSE_ERROR_TYPE];
     if (errorType == _ILLEGAL_ARGUMENT_RESPONSE) {
       controller.addError(new ArgumentError());
     } else if (errorType == _OSERROR_RESPONSE) {
       var responseError = message[RESPONSE_ERROR];
       var err = new OSError(
           responseError[_OSERROR_RESPONSE_MESSAGE],
           responseError[_OSERROR_RESPONSE_ERROR_CODE]);
       var errorPath = message[RESPONSE_PATH];
       if (errorPath == null) errorPath = path;
       controller.addError(
           new DirectoryException("Directory listing failed",
                                  errorPath,
                                  err));
     } else {
       controller.addError(
           new DirectoryException("Internal error"));
     }
   }
 }