Adding a router class to handle messages that expect responses.

mojo/bindings/java/src/org/chromium/mojo/bindings/RouterImpl.java

+// 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.
+
+package org.chromium.mojo.bindings;
+
+import org.chromium.mojo.system.AsyncWaiter;
+import org.chromium.mojo.system.MessagePipeHandle;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Implementation of Router.

[rmcilroy, 2014/07/10 19:03:59]

{@link Router}

[qsr, 2014/07/11 11:42:08]

Done.

+ */
+public class RouterImpl implements Router {
+
+    /**
+     * {@link MessageReceiver} used as the {@link Connector} callback.
+     */
+    private class Thunk implements MessageReceiver {

[rmcilroy, 2014/07/10 19:03:59]

Thunk is not a very clear name for what the class does - I would just call it
MessageReceiverImpl or something similar.

[qsr, 2014/07/11 11:42:08]

 It is the name the C++ implementation use. I'd like to keep the naming close.

+
+        /**
+         * @see MessageReceiver#accept(Message)
+         */
+        @Override
+        public boolean accept(Message message) {
+            return handleIncomingMessage(message);
+        }
+
+    }
+
+    /**
+     * The {@link Connector} connected to the handle.

[rmcilroy, 2014/07/10 19:03:59]

/s/connected/which is connected

[qsr, 2014/07/11 11:42:08]

Done.

+     */
+    private final Connector mConnector;
+
+    /**
+     * The {@link MessageReceiverWithResponder} that will deserialize and use the message received

[rmcilroy, 2014/07/10 19:03:59]

/s/deserialize and use/consume
/s/message/messages

[qsr, 2014/07/11 11:42:08]

Done.

+     * from the pipe.
+     */
+    private MessageReceiverWithResponder mIncomingMessageReceiver;
+
+    /**
+     * The id of the next request that needs a response. It is auto-incremented.

[rmcilroy, 2014/07/10 19:03:59]

The next id to use for a request id which needs a response...

[qsr, 2014/07/11 11:42:09]

Done.

+     */
+    private long mNextRequestId = 1;
+
+    /**
+     * The map from request ids to {@link MessageReceiver} of request currently in flight.
+     */
+    private Map<Long, MessageReceiver> mResponders = new HashMap<Long, MessageReceiver>();
+
+    /**
+     * Constructor.

[rmcilroy, 2014/07/10 19:03:59]

Detail parameters and that it uses the default asyncwaiter for the
messagePipeHandle

[qsr, 2014/07/11 11:42:09]

Done.

+     */
+    public RouterImpl(MessagePipeHandle messagePipeHandle) {
+        this(messagePipeHandle, BindingsHelper.getDefaultAsyncWaiterForHandle(messagePipeHandle));
+    }
+
+    /**
+     * Constructor.

[rmcilroy, 2014/07/10 19:03:59]

Detail parameters

[qsr, 2014/07/11 11:42:08]

Done.

+     */
+    public RouterImpl(MessagePipeHandle messagePipeHandle, AsyncWaiter asyncWaiter) {
+        mConnector = new Connector(messagePipeHandle, asyncWaiter);
+        mConnector.setIncomingMessageReceiver(new Thunk());
+    }
+
+    /**
+     * @see org.chromium.mojo.bindings.Router#start()
+     */
+    @Override
+    public void start() {
+        mConnector.start();
+    }
+
+    /**
+     * @see Router#setIncomingMessageReceiver(MessageReceiverWithResponder)
+     */
+    @Override
+    public void setIncomingMessageReceiver(MessageReceiverWithResponder incomingMessageReceiver) {
+        this.mIncomingMessageReceiver = incomingMessageReceiver;
+    }
+
+    /**
+     * @see MessageReceiver#accept(Message)
+     */
+    @Override
+    public boolean accept(Message message) {
+        // A message without receiver is directly forwarded to the connector.

[rmcilroy, 2014/07/10 19:03:59]

/s/receiever/a responder

[qsr, 2014/07/11 11:42:08]

Done.

+        return mConnector.accept(message);
+    }
+
+    /**
+     * @see MessageReceiverWithResponder#acceptWithResponder(Message, MessageReceiver)
+     */
+    @Override
+    public boolean acceptWithResponder(Message message, MessageReceiver receiver) {

[rmcilroy, 2014/07/10 19:03:59]

/s/receiver/responder

[qsr, 2014/07/11 11:42:09]

Done.

+        // Checking the message expects a response.
+        assert message.getHeader().hasFlag(MessageHeader.MESSAGE_EXPECTS_RESPONSE_FLAG);
+
+        // Compute a request id for being able to route the response.
+        long requestId = mNextRequestId++;
+        // Reserve 0 in case we want it to convey special meaning in the future.
+        if (requestId == 0) {
+            requestId = mNextRequestId++;
+        }
+        assert !mResponders.containsKey(requestId);

[rmcilroy, 2014/07/10 19:03:59]

I don't think it's enough to just assert this - even in release builds we want
to be sure that we never use a requestId which is inflight even when
mNextRequestId overflows.

[qsr, 2014/07/11 11:42:09]

Done. I don't think (famous last words) that it is expected that we will ever be
able to send 2^64 messages...

+        MessageHeader.setRequestId(message, requestId);
+        if (!mConnector.accept(message)) {
+            return false;
+        }
+        // Only keep the receiver is the message has been accepted.

[rmcilroy, 2014/07/10 19:03:59]

/s/receiever/responder

[qsr, 2014/07/11 11:42:09]

Done.

+        mResponders.put(requestId, receiver);
+        return true;
+    }
+
+    /**
+     * @see org.chromium.mojo.bindings.HandleOwner#passHandle()
+     */
+    @Override
+    public MessagePipeHandle passHandle() {
+        return mConnector.passHandle();
+    }
+
+    /**
+     * @see java.io.Closeable#close()
+     */
+    @Override
+    public void close() {
+        mConnector.close();
+    }
+
+    /**
+     * @see Router#setErrorHandler(ConnectionErrorHandler)
+     */
+    @Override
+    public void setErrorHandler(ConnectionErrorHandler errorHandler) {
+        mConnector.setErrorHandler(errorHandler);
+    }
+
+    /**
+     * Receive a message from the connector.

[rmcilroy, 2014/07/10 19:03:59]

describe what's returned

[qsr, 2014/07/11 11:42:08]

Done.

+     */
+    private boolean handleIncomingMessage(Message message) {
+        MessageHeader header = message.getHeader();
+        if (header.hasFlag(MessageHeader.MESSAGE_EXPECTS_RESPONSE_FLAG)) {
+            if (mIncomingMessageReceiver != null) {
+                return mIncomingMessageReceiver.acceptWithResponder(message, this);
+            }
+            // If we receive a request expecting a response when the client is not
+            // listening, then we have no choice but to tear down the pipe.
+            close();
+            return false;
+        } else if (header.hasFlag(MessageHeader.MESSAGE_IS_RESPONSE_FLAG)) {
+            long requestId = header.getRequestId();
+            MessageReceiver responder = mResponders.get(requestId);
+            if (responder == null) {
+                return false;

[rmcilroy, 2014/07/10 19:03:59]

should we teardown here too?

[qsr, 2014/07/11 11:42:08]

 Not really. The message is routed, but the problem is on the other end, not on
ours, Just dropping it should be fine.

+            }
+            return responder.accept(message);
+        } else {
+            if (mIncomingMessageReceiver != null) {
+                return mIncomingMessageReceiver.accept(message);
+            }
+            // OK to drop the message.
+        }
+        return false;
+    }
+
+}