Dart: Adds a content handler and a test.

mojo/public/dart/src/interface.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 bindings;
 
 abstract class Interface {
   core.MojoMessagePipeEndpoint _endpoint;
   core.MojoEventStream _eventStream;
   List _sendQueue;
   bool _isOpen;
+  bool _isClosing;
+  bool _isInHandler;
+  int _outstandingResponseFutures;
 
   Future<Message> handleMessage(MessageReader reader);
 
   Interface(core.MojoMessagePipeEndpoint endpoint) :
       _endpoint = endpoint,
       _sendQueue = [],
       _eventStream = new core.MojoEventStream(endpoint.handle),
-      _isOpen = false;
+      _isOpen = false,
+      _isClosing = false,
+      _isInHandler = false,
+      _outstandingResponseFutures = 0;
 
   Interface.fromHandle(int handle) {
     _endpoint =
         new core.MojoMessagePipeEndpoint(new core.MojoHandle(handle));
     _sendQueue = [];
     _eventStream = new core.MojoEventStream(_endpoint.handle);
     _isOpen = false;
+    _isClosing = false;
+    _isInHandler = false;
+    _outstandingResponseFutures = 0;
   }
 
   void _doRead() {
     assert(_eventStream.readyRead);
 
     // Query how many bytes are available.
     var result = _endpoint.query();
     assert(result.status.isOk || result.status.isResourceExhausted);
 
     // Read the data and view as a message.
     var bytes = new ByteData(result.bytesRead);
     var handles = new List<core.MojoHandle>(result.handlesRead);
     result = _endpoint.read(bytes, result.bytesRead, handles);
     assert(result.status.isOk || result.status.isResourceExhausted);
     var message = new Message(bytes, handles);
     var reader = new MessageReader(message);
 
-    // Prepare the response.
-    var responseFuture = handleMessage(reader);
+    // Prepare the response. Drop messages if we are closing.
+    var responseFuture = _isClosing ? null : handleMessage(reader);
 
     // If there's a response, queue it up for sending.
     if (responseFuture != null) {
+      _outstandingResponseFutures++;
       responseFuture.then((response) {
+        _outstandingResponseFutures--;
         _sendQueue.add(response);
         if (_sendQueue.length == 1) {
           _eventStream.enableWriteEvents();
         }
       });
+    } else if (_isClosing &&
+               (_sendQueue.length == 0) &&
+               (_outstandingResponseFutures == 0)) {
+      // We are closing, there is no response to send for this message, the send
+      // queue is empty, and there are no outstanding futures. Do the close now.
+      _close();
     }
   }
 
   void _doWrite() {
     if (_sendQueue.length > 0) {
       assert(_eventStream.readyWrite);
       var responseMessage = _sendQueue.removeAt(0);
       _endpoint.write(responseMessage.buffer,
                       responseMessage.buffer.lengthInBytes,
                       responseMessage.handles);
       if (!_endpoint.status.isOk) {
         throw "message pipe write failed: ${_endpoint.status}";
       }
+    } else if (_isClosing && (_outstandingResponseFutures == 0)) {
+      // We are closing, the send queue is empty, and there are no outstanding
+      // response futures. Really do the close, now.
+      _close();
     }
   }
 
   StreamSubscription<int> listen() {
     _isOpen = true;
     return _eventStream.listen((List<int> event) {
       var signalsWatched = new core.MojoHandleSignals(event[0]);
       var signalsReceived = new core.MojoHandleSignals(event[1]);
       if (signalsReceived.isPeerClosed) {
-        close();
+        // If the peer is closed, we can close the interface immediately.
+        _close();
         return;
       }
 
+      _isInHandler = true;
       if (signalsReceived.isReadable) {
         _doRead();
       }
 
       if (signalsReceived.isWritable) {
         _doWrite();
       }
 
       if (_sendQueue.length == 0) {
         var withoutWritable = signalsWatched - core.MojoHandleSignals.WRITABLE;
         _eventStream.enableSignals(withoutWritable);
       } else {
         _eventStream.enableSignals(signalsWatched);
       }
+      _isInHandler = false;
     });
   }
 
+  void _close() {
+    _eventStream.close();
+    _isClosing = false;
+    _isOpen = false;
+    _eventStream = null;
+  }
+
   void close() {
-    // TODO(zra): Cancel outstanding Futures started in _doRead?
-    if (_isOpen) {
-      _eventStream.close();
-      _isOpen = false;
-      _eventStream = null;
+    if (!_isOpen) return;
+    if (_isInHandler || (_sendQueue.length > 0) ||
+        (_outstandingResponseFutures > 0)) {
+      // If close is called from within the event handler, or if there are
+      // outstanding responses to send, then defer the close call until all
+      // everything is finished.
+      _isClosing = true;
+    } else {
+      _close();
     }
   }
 
   Message buildResponse(Type t, int name, Object response) {
     var builder = new MessageBuilder(name, align(getEncodedSize(t)));
     builder.encodeStruct(t, response);
     return builder.finish();
   }
 
   Message buildResponseWithID(
@@ skipping 13 matching lines @@
   }
 
   Future enqueueMessageWithRequestID(Type t, int name, int id, Object msg) {
     // TODO(zra): Is this correct?
     throw "The client interface should not expect a response";
   }
 
   bool get isOpen => _isOpen;
   core.MojoMessagePipeEndpoint get endpoint => _endpoint;
 }