1 files changed, 33 insertions, 14 deletions

diff --git a/mojo/public/dart/src/handle.dart b/mojo/public/dart/src/handle.dart
index bf9d87b..bf2f74e 100644
--- a/mojo/public/dart/src/handle.dart
+++ b/mojo/public/dart/src/handle.dart
@@ -5,6 +5,7 @@
 part of core;
 
 class _MojoHandleNatives {
+  static int register(MojoHandle handle) native "MojoHandle_Register";
   static int close(int handle) native "MojoHandle_Close";
   static int wait(int handle, int signals, int deadline)
       native "MojoHandle_Wait";
@@ -62,9 +63,17 @@ class RawMojoHandle {
         handles, signals, handles.length, deadline);
   }
 
-  static bool isValid(RawMojoHandle h) => (h.h != INVALID);
+  static MojoResult register(MojoHandle handle) {
+    return new MojoResult(_MojoHandleNatives.register(handle));
+  }
+
+  bool get isValid => (h != INVALID);
 
   String toString() => "$h";
+
+  bool operator ==(RawMojoHandle other) {
+    return h == other.h;
+  }
 }
 
 
@@ -92,25 +101,24 @@ class MojoHandle extends Stream<int> {
 
   MojoHandle(this._handle) : 
       _signals = MojoHandleSignals.READABLE,
-      _eventHandlerAdded = false,
-      _receivePort = new ReceivePort() {
-    _sendPort = _receivePort.sendPort;
-    _controller = new StreamController(sync: true,
-        onListen: _onSubscriptionStateChange,
-        onCancel: _onSubscriptionStateChange,
-        onPause: _onPauseStateChange,
-        onResume: _onPauseStateChange);
-    _controller.addStream(_receivePort);
+      _eventHandlerAdded = false {
+    MojoResult result = RawMojoHandle.register(this);
+    if (!result.isOk) {
+      throw new Exception("Failed to register the MojoHandle");
+    }
   }
 
   void close() {
     if (_eventHandlerAdded) {
       MojoHandleWatcher.close(_handle);
+      _eventHandlerAdded = false;
     } else {
       // If we're not in the handle watcher, then close the handle manually.
       _handle.close();
     }
-    _receivePort.close();
+    if (_receivePort != null) {
+      _receivePort.close();
+    }
   }
 
   // We wrap the callback provided by clients in listen() with some code to
@@ -125,11 +133,11 @@ class MojoHandle extends Stream<int> {
 
       // The callback could have closed the handle. If so, don't add it back to
       // the MojoHandleWatcher.
-      if (RawMojoHandle.isValid(_handle)) {
+      if (_handle.isValid) {
         assert(!_eventHandlerAdded);
         var res = MojoHandleWatcher.add(_handle, _sendPort, _signals);
         if (!res.isOk) {
-          throw new Exception("Failed to re-add handle: $_handle");
+          throw new Exception("Failed to re-add handle: $res");
         }
         _eventHandlerAdded = true;
       }
@@ -139,6 +147,14 @@ class MojoHandle extends Stream<int> {
   StreamSubscription<int> listen(
       void onData(int event),
       {Function onError, void onDone(), bool cancelOnError}) {
+    _receivePort = new ReceivePort();
+    _sendPort = _receivePort.sendPort;
+    _controller = new StreamController(sync: true,
+        onListen: _onSubscriptionStateChange,
+        onCancel: _onSubscriptionStateChange,
+        onPause: _onPauseStateChange,
+        onResume: _onPauseStateChange);
+    _controller.addStream(_receivePort);
 
     assert(!_eventHandlerAdded);
     var res = MojoHandleWatcher.add(_handle, _sendPort, _signals);
@@ -185,7 +201,10 @@ class MojoHandle extends Stream<int> {
   void _onPauseStateChange() {
     if (_controller.isPaused) {
       if (_eventHandlerAdded) {
-        MojoHandleWatcher.remove(_handle);
+        var res = MojoHandleWatcher.remove(_handle);
+        if (!res.isOk) {
+          throw new Exception("MojoHandleWatcher add failed: $res");
+        }
         _eventHandlerAdded = false;
       }
     } else {