Fifo order should be preserved for messages on associated interfaces.

mojo/public/js/router.js

 // 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.
 
 define("mojo/public/js/router", [
   "mojo/public/js/connector",
   "mojo/public/js/core",
   "mojo/public/js/interface_types",
   "mojo/public/js/lib/interface_endpoint_handle",
   "mojo/public/js/lib/pipe_control_message_handler",
@@ skipping 54 matching lines @@
     this.connector_ = new Connector(handle);
 
     this.connector_.setIncomingReceiver({
         accept: this.accept.bind(this),
     });
     this.connector_.setErrorHandler({
         onError: this.onPipeConnectionError.bind(this),
     });
 
     this.setInterfaceIdNamespaceBit_ = setInterfaceIdNamespaceBit;
+    // |cachedMessageData| caches infomation about a message, so it can be
+    // processed later if a client is not yet attached to the target endpoint.
+    this.cachedMessageData = null;
     this.controlMessageHandler_ = new PipeControlMessageHandler(this);
     this.controlMessageProxy_ = new PipeControlMessageProxy(this.connector_);
     this.nextInterfaceIdValue_ = 1;
     this.encounteredError_ = false;
     this.endpoints_ = new Map();
   }
 
   Router.prototype.associateInterface = function(handleToSend) {
     if (!handleToSend.pendingAssociation()) {
       return types.kInvalidInterfaceId;
@@ skipping 40 matching lines @@
     check(endpoint);
     check(!endpoint.client);
     check(!endpoint.closed);
     endpoint.client = interfaceEndpointClient;
 
     if (endpoint.peerClosed) {
       timer.createOneShot(0,
           endpoint.client.notifyError.bind(endpoint.client));
     }
 
+    if (this.cachedMessageData && interfaceEndpointHandle.id() ===
+        this.cachedMessageData.message.getInterfaceId()) {
+      timer.createOneShot(0, (function() {
+        if (!this.cachedMessageData) {
+          return;
+        }
+
+        var targetEndpoint = this.endpoints_.get(
+            this.cachedMessageData.message.getInterfaceId());
+        // Check that the target endpoint's client still exists.
+        if (targetEndpoint && targetEndpoint.client) {
+          var ok = endpoint.client.handleIncomingMessage(
+              this.cachedMessageData.message,

[yzshen1, 2017/04/26 16:40:00]

nit: please move the contents of cachedMessageData to a local variable and
resumeIncomingMethodCallProcessing() before calling handleIncomingMessage().
Because that method may re-enter the router, it is a little more robust to do
things this way.

[wangjimmy, 2017/04/26 17:54:07]

Done.

+              this.cachedMessageData.messageValidator);
+
+          if (!ok) {
+            this.handleInvalidIncomingMessage_();
+          }
+          this.cachedMessageData = null;
+          this.connector_.resumeIncomingMethodCallProcessing();
+        }
+      }).bind(this));
+    }
+
     return endpoint;
   };
 
   Router.prototype.detachEndpointClient = function(
       interfaceEndpointHandle) {
     check(types.isValidInterfaceId(interfaceEndpointHandle.id()));
     var endpoint = this.endpoints_.get(interfaceEndpointHandle.id());
     check(endpoint);
     check(endpoint.client);
     check(!endpoint.closed);
@@ skipping 47 matching lines @@
       } else {
         var interfaceId = message.getInterfaceId();
         var endpoint = this.endpoints_.get(interfaceId);
         if (!endpoint || endpoint.closed) {
           return true;
         }
 
         if (!endpoint.client) {
           // We need to wait until a client is attached in order to dispatch
           // further messages.
-          // TODO(wangjimmy): Cache the message and send when the appropriate
-          // endpoint client is attached.
-          return false;
+          this.cachedMessageData = {message: message,
+              messageValidator: messageValidator};
+          this.connector_.pauseIncomingMethodCallProcessing();
+          return true;
         }
         ok = endpoint.client.handleIncomingMessage(message, messageValidator);
       }
     }
 
     if (!ok) {
       this.handleInvalidIncomingMessage_();
     }
     return ok;
   };
@@ skipping 68 matching lines @@
     check(endpoint);
     check(!endpoint.client);
     check(!endpoint.closed);
 
     this.updateEndpointStateMayRemove(endpoint,
         EndpointStateUpdateType.ENDPOINT_CLOSED);
 
     if (!types.isMasterInterfaceId(interfaceId) || reason) {
       this.controlMessageProxy_.notifyPeerEndpointClosed(interfaceId, reason);
     }
+
+    if (this.cachedMessageData && interfaceId ===
+        this.cachedMessageData.message.getInterfaceId()) {
+      timer.createOneShot(0, (function() {

[yzshen1, 2017/04/26 16:40:00]

no need to use timer: resumeIncomingMethodCallProcessing() won't receive message
synchronously.

[wangjimmy, 2017/04/26 17:54:07]

Done.

+        this.cachedMessageData = null;
+        this.connector_.resumeIncomingMethodCallProcessing();
+      }).bind(this));
+    }
   };
 
   Router.prototype.updateEndpointStateMayRemove = function(endpoint,
       endpointStateUpdateType) {
     if (endpointStateUpdateType === EndpointStateUpdateType.ENDPOINT_CLOSED) {
       endpoint.closed = true;
     } else {
       endpoint.peerClosed = true;
     }
     if (endpoint.closed && endpoint.peerClosed) {
       this.endpoints_.delete(endpoint.id);
     }
   };
 
   var exports = {};
   exports.Router = Router;
   return exports;
 });