Fix File.openStream where it sometimes would keep the file open.

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;
+  bool _closed = false;
 
   // Block read but not yet send because stream is paused.
   List<int> _currentBlock;
 
   _FileStream(String this._path, this._position, this._end) {
     _setupController();
   }
 
   _FileStream.forStdin() : _position = 0 {
     _setupController();
@@ skipping 14 matching lines @@
         onListen: _start,
         onPause: () => _paused = true,
         onResume: _resume,
         onCancel: () {
           _unsubscribed = true;
           return _closeFile();
         });
   }
 
   Future _closeFile() {
-    if (_readInProgress) {
+    if (_readInProgress || _closed) {
       return _closeCompleter.future;
     }
+    _closed = true;
+    void done() {
+      _closeCompleter.complete();
+      _controller.close();
+    }
     if (_openedFile != null) {
       _openedFile.close()
-          .then(_closeCompleter.complete,
-                onError: _closeCompleter.completeError);
+          .catchError(_controller.addError)
+          .whenComplete(done);
       _openedFile = null;
+    } else {
+      done();
     }
     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) {
@@ skipping 12 matching lines @@
       .whenComplete(() {
         _readInProgress = false;
       })
       .then((block) {
         if (_unsubscribed) {
           _closeFile();
           return;
         }
         if (block.length == 0) {
           if (!_unsubscribed) {
-            _closeFile().then((_) { _controller.close(); });
+            _closeFile();
             _unsubscribed = true;
           }
           return;
         }
         _position += block.length;
         if (_paused) {
           _currentBlock = block;
         } else {
           _controller.add(block);
           _readBlock();
         }
       })
       .catchError((e) {
         if (!_unsubscribed) {
           _controller.addError(e);
-          _closeFile().then((_) { _controller.close(); });
+          _closeFile();
           _unsubscribed = true;
         }
       });
   }
 
   void _start() {
     if (_position == null) {
       _position = 0;
     } else if (_position < 0) {
       _controller.addError(new RangeError("Bad start position: $_position"));
@@ skipping 732 matching lines @@
 
   void _checkAvailable() {
     if (_asyncDispatched) {
       throw new FileSystemException("An async operation is currently pending", path);
     }
     if (closed) {
       throw new FileSystemException("File closed", path);
     }
   }
 }