Fix File stream on cancel.

sdk/lib/io/file_impl.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.
 
 part of dart.io;
 
 // Read the file in blocks of size 64k.
 const int _BLOCK_SIZE = 64 * 1024;
 
 
 class _FileStream extends Stream<List<int>> {
   // Stream controller.
   StreamController<List<int>> _controller;
 
   // Information about the underlying file.
   String _path;
   RandomAccessFile _openedFile;
   int _position;
   int _end;
+  final Completer _closeCompleter = new Completer();
 
   // Has the stream been paused or unsubscribed?
   bool _paused = false;
   bool _unsubscribed = false;
 
   // Is there a read currently in progress?
   bool _readInProgress = false;
 
   // Block read but not yet send because stream is paused.
   List<int> _currentBlock;
@@ skipping 21 matching lines @@
         onListen: _start,
         onPause: () => _paused = true,
         onResume: _resume,
         onCancel: () {
           _unsubscribed = true;
           _closeFile();
         });
   }
 
   Future _closeFile() {
-    Future closeFuture;
+    if (_readInProgress) {
+      return _closeCompleter.future;
+    }
     if (_openedFile != null) {
-      Future closeFuture = _openedFile.close();
+      _openedFile.close()
+          .then(_closeCompleter.complete,
+                onError: _closeCompleter.completeError);
       _openedFile = null;
-      return closeFuture;
-    } else {
-      return new Future.value();
     }
+    return _closeCompleter.future;
   }
 
   void _readBlock() {
     // Don't start a new read if one is already in progress.
     if (_readInProgress) return;
     _readInProgress = true;
     int readBytes = _BLOCK_SIZE;
     if (_end != null) {
       readBytes = min(readBytes, _end - _position);
       if (readBytes < 0) {
+        _readInProgress = false;
         if (!_unsubscribed) {
           _controller.addError(new RangeError("Bad end position: $_end"));
           _closeFile().then((_) { _controller.close(); });
           _unsubscribed = true;
         }
         return;
       }
     }
     _openedFile.read(readBytes)
+      .whenComplete(() {
+        _readInProgress = false;
+      })
       .then((block) {
-        _readInProgress = false;
+        if (_unsubscribed) {
+          _closeFile();
+          return;
+        }
         if (block.length == 0) {
           if (!_unsubscribed) {
             _closeFile().then((_) { _controller.close(); });
             _unsubscribed = true;
           }
           return;
         }
         _position += block.length;
         if (_paused) {
           _currentBlock = block;
@@ skipping 756 matching lines @@
   void _checkNotClosed() {
     if (closed) {
       throw new FileException("File closed", path);
     }
   }
 
   Future _closedException() {
     return new Future.error(new FileException("File closed", path));
   }
 }