Provide mechanism to close immediately to Dart bindings

mojo/public/dart/src/event_stream.dart

Index: mojo/public/dart/src/event_stream.dart
diff --git a/mojo/public/dart/src/event_stream.dart b/mojo/public/dart/src/event_stream.dart
index 712d0259f8308868891e9a6f65bdb2b3d612e7ee..b040f7c7294b3def06e21a5905952bc58092321f 100644
--- a/mojo/public/dart/src/event_stream.dart
+++ b/mojo/public/dart/src/event_stream.dart
@@ -37,12 +37,12 @@ class MojoEventStream extends Stream<List<int>> {
     }
   }
 
-  Future close() {
+  Future close({bool immediate: false}) {
     if (_handle != null) {
       if (_isListening) {
-        return _handleWatcherClose();
+        return _handleWatcherClose(immediate: immediate);
       } else {
-        _localClose();
+        _localClose(immediate: immediate);
         return new Future.value(null);
       }
     }
@@ -92,10 +92,10 @@ class MojoEventStream extends Stream<List<int>> {
   void enableWriteEvents() => enableSignals(MojoHandleSignals.WRITABLE);
   void enableAllEvents() => enableSignals(MojoHandleSignals.READWRITE);
 
-  Future _handleWatcherClose() {
+  Future _handleWatcherClose({bool immediate: false}) {
     assert(_handle != null);
     assert(MojoHandle._removeUnclosedHandle(_handle));
-    return MojoHandleWatcher.close(_handle.h, wait: true).then((r) {
+    return MojoHandleWatcher.close(_handle.h, wait: !immediate).then((r) {
       if (_receivePort != null) {
         _receivePort.close();
         _receivePort = null;
@@ -104,7 +104,7 @@ class MojoEventStream extends Stream<List<int>> {
     });
   }
 
-  void _localClose() {
+  void _localClose({bool immediate: false}) {

[zra, 2015/04/07 19:49:46]

Is this needed for something?

[Cutch, 2015/04/07 19:52:12]

It's not needed, I was just passing the value down. I'll remove it for now.

     assert(_handle != null);
     _handle.close();
     _handle = null;
@@ -116,7 +116,8 @@ class MojoEventStream extends Stream<List<int>> {
 
   void _onSubscriptionStateChange() {
     if (!_controller.hasListener) {
-      close();
+      // No one is listening, close it immediately.
+      close(immediate: true);
     }
   }
 
@@ -210,9 +211,9 @@ class MojoEventStreamListener {
       }
       _isInHandler = false;
       if (signalsReceived.isPeerClosed) {
-        // nodefer is true here because there is no need to wait to close until
-        // outstanding messages are sent. The other side is gone.
-        close(nodefer: true).then((_) {
+        // immediate is true here because there is no need to wait to close
+        // until outstanding messages are sent. The other side is gone.
+        close(immediate: true).then((_) {
           if (onError != null) {
             onError();
           }
@@ -222,13 +223,13 @@ class MojoEventStreamListener {
     return subscription;
   }
 
-  Future close({bool nodefer: false}) {
+  Future close({bool immediate: false}) {
     var result;
     _isOpen = false;
     _endpoint = null;
     subscription = null;
     if (_eventStream != null) {
-      result = _eventStream.close().then((_) {
+      result = _eventStream.close(immediate: immediate).then((_) {
         _eventStream = null;
       });
     }