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

examples/dart/netcat/lib/main.dart

 #!mojo mojo:dart_content_handler
 // 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.
 
 import 'dart:async';
 import 'dart:core';
 import 'dart:typed_data';
 
 import 'package:mojo/application.dart';
@@ skipping 31 matching lines @@
 // * Relatedly, we should listen for _socketSender's peer being closed (also
 //   _socket, I guess).
 // * Handle the socket send pipe being full (currently, we assume it's never
 //   full).
 class Connector {
   final Application _application;
   files.FileProxy _terminal;
   TcpConnectedSocketProxy _socket;
   MojoDataPipeProducer _socketSender;
   MojoDataPipeConsumer _socketReceiver;
-  MojoEventStream _socketReceiverEventStream;
+  MojoEventSubscription _socketReceiverEventSubscription;
   final ByteData _readBuffer;
   final ByteData _writeBuffer;
 
   // TODO(vtl): Don't just hard-code buffer sizes.
   Connector(this._application, this._terminal)
       : _readBuffer = new ByteData(16 * 1024),
         _writeBuffer = new ByteData(16 * 1024);
 
   Future connect(NetAddress remote_address) async {
     try {
@@ skipping 11 matching lines @@
       _socketReceiver = receiveDataPipe.consumer;
       _socket = new TcpConnectedSocketProxy.unbound();
       await boundSocket.ptr.connect(remote_address, sendDataPipe.consumer,
           receiveDataPipe.producer, _socket);
       await boundSocket.close();
 
       // Set up reading from the terminal.
       _startReadingFromTerminal();
 
       // Set up reading from the socket.
-      _socketReceiverEventStream = new MojoEventStream(_socketReceiver.handle);
-      _socketReceiverEventStream.listen(_onSocketReceiverEvent);
+      _socketReceiverEventSubscription =
+          new MojoEventSubscription(_socketReceiver.handle);
+      _socketReceiverEventSubscription.subscribe(_onSocketReceiverEvent);
     } catch (e) {
       _shutDown();
     }
   }
 
   void _startReadingFromTerminal() {
     // TODO(vtl): Do we have to do something on error?
     _terminal.ptr
         .read(_writeBuffer.lengthInBytes, 0, files.Whence.FROM_CURRENT)
         .then(_onReadFromTerminal)
-        .catchError((e) { _shutDown(); });
+        .catchError((e) {
+      _shutDown();
+    });
   }
 
   void _onReadFromTerminal(files.FileReadResponseParams p) {
     if (p.error != files.Error.OK) {
       // TODO(vtl): Do terminal errors.
       return;
     }
 
     // TODO(vtl): Verify that |bytesRead.length| is within the expected range.
     for (var i = 0, j = 0; i < p.bytesRead.length; i++, j++) {
@@ skipping 13 matching lines @@
     _startReadingFromTerminal();
   }
 
   void _onSocketReceiverEvent(List<int> event) {
     var mojoSignals = new MojoHandleSignals(event[1]);
     var shouldShutDown = false;
     if (mojoSignals.isReadable) {
       var numBytesRead = _socketReceiver.read(_readBuffer);
       if (_socketReceiver.status.isOk) {
         assert(numBytesRead > 0);
-        _terminal.ptr.write(_readBuffer.buffer.asUint8List(0, numBytesRead), 0,
-                            files.Whence.FROM_CURRENT)
-            .catchError((e) { _shutDown(); });
-        _socketReceiverEventStream.enableReadEvents();
+        _terminal.ptr
+            .write(_readBuffer.buffer.asUint8List(0, numBytesRead), 0,
+                files.Whence.FROM_CURRENT)
+            .catchError((e) {
+          _shutDown();
+        });
+        _socketReceiverEventSubscription.enableReadEvents();
       } else {
         shouldShutDown = true;
       }
     } else if (mojoSignals.isPeerClosed) {
       shouldShutDown = true;
     } else {
       throw 'Unexpected handle event: $mojoSignals';
     }
     if (shouldShutDown) {
       _shutDown();
     }
   }
 
   void _shutDown() {
-    if (_socketReceiverEventStream != null) {
-      ignoreFuture(_socketReceiverEventStream.close());
-      _socketReceiverEventStream = null;
+    if (_socketReceiverEventSubscription != null) {
+      ignoreFuture(_socketReceiverEventSubscription.close());
+      _socketReceiverEventSubscription = null;
     }
     if (_socketSender != null) {
-      if (_socketSender.handle.isValid)
-        _socketSender.handle.close();
+      if (_socketSender.handle.isValid) _socketSender.handle.close();
       _socketSender = null;
     }
     if (_socketReceiver != null) {
-      if (_socketReceiver.handle.isValid)
-        _socketReceiver.handle.close();
+      if (_socketReceiver.handle.isValid) _socketReceiver.handle.close();
       _socketReceiver = null;
     }
     if (_terminal != null) {
       ignoreFuture(_terminal.close());
       _terminal = null;
     }
   }
 }
 
 class TerminalClientImpl implements TerminalClient {
   TerminalClientStub _stub;
   Application _application;
   String _resolvedUrl;
 
   TerminalClientImpl(
       this._application, this._resolvedUrl, MojoMessagePipeEndpoint endpoint) {
     _stub = new TerminalClientStub.fromEndpoint(endpoint, this);
   }
 
   @override
   void connectToTerminal(files.FileProxy terminal) {
     var url = Uri.parse(_resolvedUrl);
     NetAddress remote_address;
     try {
       remote_address = _getNetAddressFromUrl(url);
     } catch (e) {
-      fputs(terminal.ptr, 'HALP: Add a query: ?host=<host>&port=<port>\n'
-          '(<host> must be "localhost" or n1.n2.n3.n4)\n\n'
-          'Got query parameters:\n' + url.queryParameters.toString());
+      fputs(
+          terminal.ptr,
+          'HALP: Add a query: ?host=<host>&port=<port>\n'
+              '(<host> must be "localhost" or n1.n2.n3.n4)\n\n'
+              'Got query parameters:\n' +
+              url.queryParameters.toString());
       ignoreFuture(terminal.close());
       return;
     }
 
     // TODO(vtl): Currently, we only do IPv4, so this should work.
-    fputs(terminal.ptr,
-          'Connecting to: ' + remote_address.ipv4.addr.join('.') + ':' +
-              remote_address.ipv4.port.toString() + '...');
+    fputs(
+        terminal.ptr,
+        'Connecting to: ' +
+            remote_address.ipv4.addr.join('.') +
+            ':' +
+            remote_address.ipv4.port.toString() +
+            '...');
 
     var connector = new Connector(_application, terminal);
     // TODO(vtl): Do we have to do something on error?
-    connector.connect(remote_address)
-        .catchError((e) {});
+    connector.connect(remote_address).catchError((e) {});
   }
 
   // Note: May throw all sorts of things.
   static NetAddress _getNetAddressFromUrl(Uri url) {
     var params = url.queryParameters;
     var host = params['host'];
     return makeIPv4NetAddress(
         (host == 'localhost') ? [127, 0, 0, 1] : Uri.parseIPv4Address(host),
         int.parse(params['port']));
   }
 }
 
 class NetcatApplication extends Application {
   NetcatApplication.fromHandle(MojoHandle handle) : super.fromHandle(handle);
 
   @override
   void acceptConnection(String requestorUrl, String resolvedUrl,
       ApplicationConnection connection) {
     connection.provideService(TerminalClientName,
         (endpoint) => new TerminalClientImpl(this, resolvedUrl, endpoint));
   }
 }
 
 main(List args) {
   MojoHandle appHandle = new MojoHandle(args[0]);
   String url = args[1];
   new NetcatApplication.fromHandle(appHandle)
-    ..onError = (() {
+    ..onError = ((Object e) {
       MojoHandle.reportLeakedHandles();
     });
 }