Dart: Better handle leak checks. close() is async.

mojo/public/dart/src/event_stream.dart

 // Copyright 2014 The Chromium Authors. 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 core;
 
 class MojoEventStream extends Stream<List<int>> {
   // The underlying Mojo handle.
   MojoHandle _handle;
 
@@ skipping 19 matching lines @@
       [MojoHandleSignals signals = MojoHandleSignals.PEER_CLOSED_READABLE])
       : _handle = handle,
         _signals = signals,
         _isListening = false {
     MojoResult result = MojoHandle.register(this);
     if (!result.isOk) {
       throw "Failed to register the MojoHandle: $result.";
     }
   }
 
-  void close() {
+  Future close() {
     if (_handle != null) {
       if (_isListening) {
-        MojoHandleWatcher.close(_handle);
+        return _handleWatcherClose();
       } else {
-        _handle.close();
+        _localClose();
+        return new Future.value(null);
       }
-      _handle = null;
-    }
-    if (_receivePort != null) {
-      _receivePort.close();
-      _receivePort = null;
     }
   }
 
   StreamSubscription<List<int>> listen(void onData(List event),
       {Function onError, void onDone(), bool cancelOnError}) {
     if (_isListening) {
       throw "Listen has already been called: $_handle.";
     }
     _receivePort = new ReceivePort();
     _sendPort = _receivePort.sendPort;
@@ skipping 25 matching lines @@
         throw "MojoHandleWatcher add failed: $res";
       }
     }
   }
 
   void enableReadEvents() =>
       enableSignals(MojoHandleSignals.PEER_CLOSED_READABLE);
   void enableWriteEvents() => enableSignals(MojoHandleSignals.WRITABLE);
   void enableAllEvents() => enableSignals(MojoHandleSignals.READWRITE);
 
+  Future _handleWatcherClose() {
+    assert(_handle != null);
+    return MojoHandleWatcher.close(_handle, wait: true).then((_) {
+      if (_receivePort != null) {
+        _receivePort.close();
+        _receivePort = null;
+      }
+    });
+  }
+
+  void _localClose() {
+    assert(_handle != null);
+    _handle.close();
+    _handle = null;
+    if (_receivePort != null) {
+      _receivePort.close();
+      _receivePort = null;
+    }
+  }
+
   void _onSubscriptionStateChange() {
     if (!_controller.hasListener) {
       close();
     }
   }
 
   void _onPauseStateChange() {
     if (_controller.isPaused) {
       var res = MojoHandleWatcher.remove(_handle);
       if (!res.isOk) {
@@ skipping 53 matching lines @@
     assert(!isBound);
     _endpoint = new MojoMessagePipeEndpoint(handle);
     _eventStream = new MojoEventStream(handle);
     _isOpen = false;
   }
 
   StreamSubscription<List<int>> listen() {
     assert(isBound && (subscription == null));
     _isOpen = true;
     subscription = _eventStream.listen((List<int> event) {
+      if (!_isOpen) {
+        // The actual close of the underlying stream happens asynchronously
+        // after the call to close. However, we start to ignore incoming events
+        // immediately.
+        return;
+      }
       var signalsWatched = new MojoHandleSignals(event[0]);
       var signalsReceived = new MojoHandleSignals(event[1]);
       _isInHandler = true;
       if (signalsReceived.isReadable) {
         assert(_eventStream.readyRead);
         handleRead();
       }
       if (signalsReceived.isWritable) {
         assert(_eventStream.readyWrite);
         handleWrite();
       }
       if (_isOpen) {
         _eventStream.enableSignals(signalsWatched);
       }
       _isInHandler = false;
       if (signalsReceived.isPeerClosed) {
-        if (onError != null) onError();
+        if (onError != null) {
+          onError();
+        }
         close();
-        // The peer being closed obviates any other signal we might
-        // have received since we won't be able to read or write the handle.
-        // Thus, we just return before invoking other handlers.
-        return;
       }
     }, onDone: close);
     return subscription;
   }
 
-  void close() {
+  Future close() {
+    var result;
     _isOpen = false;
     _endpoint = null;
     subscription = null;
     if (_eventStream != null) {
-      _eventStream.close();
-      _eventStream = null;
+      result = _eventStream.close().then((_) {
+        _eventStream = null;
+      });
     }
+    return result != null ? result : new Future.value(null);
   }
 
   void handleRead() {}
   void handleWrite() {}
 
   MojoMessagePipeEndpoint get endpoint => _endpoint;
   bool get isOpen => _isOpen;
   bool get isInHandler => _isInHandler;
   bool get isBound => _endpoint != null;
 
   String toString() => "MojoEventStreamListener("
       "isOpen: $isOpen, isBound: $isBound, endpoint: $_endpoint)";
 }