Dart: Use a RawReceivePort to receive events for Mojo handles.

mojo/dart/embedder/io/socket_patch.dart

 // Copyright 2015 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.
 
 //
 // Implementation of Socket and RawSocket for Mojo.
 //
 
 patch class Socket {
   /* patch */ static Future<Socket> connect(host, int port, {sourceAddress}) {
@@ skipping 16 matching lines @@
   // Constructing a new MojoDataPipe allocates two handles. All failure paths
   // must be sure that these handles are closed so we do not leak any handles.
   final _pipeOut = new MojoDataPipe();
   bool _outClosed = false;
   // Constructing a new MojoDataPipe allocates two handles. All failure paths
   // must be sure that these handles are closed so we do not leak any handles.
   final _pipeIn = new MojoDataPipe();
   bool _inClosed = false;
   bool _readEventsEnabled = true;
   bool _writeEventsEnabled = true;
-  MojoEventStream _pipeOutEvents;
-  MojoEventStream _pipeInEvents;
+  MojoEventSubscription _pipeOutEvents;
+  MojoEventSubscription _pipeInEvents;
   InternetAddress _localAddress;
   int _localPort;
   InternetAddress _remoteAddress;
   int _remotePort;
   var _owner;
 
   bool _trace = false;
   int _traceId;
 
   _tracePrint(String message) {
@@ skipping 211 matching lines @@
       if (_trace) {
         _tracePrint('<- READ_CLOSED');
       }
       _controller.add(RawSocketEvent.READ_CLOSED);
       // Once we are closed, stop reporting events.
       _inClosed = true;
       return;
     }
   }
 
-  _onInputError(e, st) {
-    _controller.addError(e);
-    _onInputDone();
-  }
-
-  _onInputDone() {
-    if (_inClosed) {
-      return;
-    }
-    if (_trace) {
-      _tracePrint('<- READ_CLOSED (done)');
-    }
-    _controller.add(RawSocketEvent.READ_CLOSED);
-    _inClosed = true;
-  }
-
   _onOutputData(List<int> event) {
     if (_outClosed) {
       return;
     }
     var signalsWatched = new MojoHandleSignals(event[0]);
     var signalsReceived = new MojoHandleSignals(event[1]);
     if (_trace) {
       _tracePrint('<- OUT: ${signalsReceived}');
     }
     if (signalsReceived.isPeerClosed) {
       if (_trace) {
         _tracePrint('<- CLOSED');
       }
       _controller.add(RawSocketEvent.CLOSED);
       // Once we are closed, stop reporting events.
       _outClosed = true;
       return;
     }
     if (signalsReceived.isWritable) {
       if (_trace) {
         _tracePrint('<- WRITE');
       }
       _controller.add(RawSocketEvent.WRITE);
     }
   }
 
-  _onOutputError(e, st) {
-    _controller.addError(e);
-    _onOutputDone();
-  }
-
-  _onOutputDone() {
-    if (_outClosed) {
-      return;
-    }
-    if (_trace) {
-      _tracePrint('<- CLOSED (done)');
-    }
-    _controller.add(RawSocketEvent.CLOSED);
-    _outClosed = true;
-  }
-
   _setupIn() {
     assert(_pipeInEvents == null);
-    _pipeInEvents = new MojoEventStream(_pipeIn.consumer.handle,
+    _pipeInEvents = new MojoEventSubscription(_pipeIn.consumer.handle,
                                         MojoHandleSignals.READABLE +
                                         MojoHandleSignals.PEER_CLOSED);
-    _pipeInEvents.listen(_onInputData,
-                         onError: _onInputError,
-                         onDone: _onInputDone);
+    _pipeInEvents.subscribe(_onInputData);
   }
 
   _setupOut() {
     assert(_pipeOutEvents == null);
-    _pipeOutEvents = new MojoEventStream(_pipeOut.producer.handle,
+    _pipeOutEvents = new MojoEventSubscription(_pipeOut.producer.handle,
                                          MojoHandleSignals.WRITABLE +
                                          MojoHandleSignals.PEER_CLOSED);
-    _pipeOutEvents.listen(_onOutputData,
-                          onError: _onOutputError,
-                          onDone: _onOutputDone);
+    _pipeOutEvents.subscribe(_onOutputData);
   }
 
   _shutdownIn([bool force = false]) {
     _inClosed = true;
     if (_trace) {
       _tracePrint('shutdown IN');
       _tracePipeIn();
     }
     if (_pipeInEvents != null) {
       if (force) {
@@ skipping 135 matching lines @@
 
   StreamSubscription<RawSocketEvent> listen(void onData(RawSocketEvent event),
                                             {Function onError,
                                              void onDone(),
                                              bool cancelOnError}) {
     return _controller.stream.listen(onData, onError: onError, onDone: onDone,
                                      cancelOnError: cancelOnError);
   }
 
 
-  static _enableReadEvents(MojoEventStream stream) {
-    if (stream == null) {
+  static _enableReadEvents(MojoEventSubscription subscription) {
+    if (subscription == null) {
       return;
     }
-    stream.enableSignals(MojoHandleSignals.PEER_CLOSED +
-                         MojoHandleSignals.READABLE);
+    subscription.enableSignals(MojoHandleSignals.PEER_CLOSED +
+                               MojoHandleSignals.READABLE);
   }
 
-  static _enableWriteEvents(MojoEventStream stream) {
-    if (stream == null) {
+  static _enableWriteEvents(MojoEventSubscription subscription) {
+    if (subscription == null) {
       return;
     }
-    stream.enableSignals(MojoHandleSignals.PEER_CLOSED +
-                         MojoHandleSignals.WRITABLE);
+    subscription.enableSignals(MojoHandleSignals.PEER_CLOSED +
+                               MojoHandleSignals.WRITABLE);
   }
 
-  static _disableEvents(MojoEventStream stream) {
-    if (stream == null) {
+  static _disableEvents(MojoEventSubscription subscription) {
+    if (subscription == null) {
       return;
     }
-    stream.enableSignals(MojoHandleSignals.PEER_CLOSED);
+    subscription.enableSignals(MojoHandleSignals.PEER_CLOSED);
   }
 
   _pause() {
     _disableEvents(_pipeInEvents);
     _disableEvents(_pipeOutEvents);
   }
 
   void _resume() {
     if (_pipeInEvents != null) {
       if (_readEventsEnabled) {
@@ skipping 320 matching lines @@
       if (_raw != null) {
         _raw.shutdown(SocketDirection.SEND);
         _disableWriteEvent();
       }
     }
   }
 
   Map _toJSON(bool ref) => _raw._toJSON(ref);
   void set _owner(owner) { _raw._owner = owner; }
 }