Mojo: Add a way to create thread-safe interfaces in Java

This change adds Interface.Manager.buildThreadSafeProxy(), which is
roughly analogous to mojo::ThreadSafeInterfacePtr<T>. Given that Java
does not have move-only semantics, the Proxy object that is passed is
unbound and a new object is returned.

Bug: 810084
Test: cheets_ContainerSmokeTest in Chrome OS
Change-Id: I6565f9e526e3fa8f8cb222cb8cd11e95bb57f7d3
Reviewed-on: https://chromium-review.googlesource.com/1147320Reviewed-by: Ken Rockot <rockot@chromium.org>
Commit-Queue: Luis Hector Chavez <lhchavez@chromium.org>
Cr-Commit-Position: refs/heads/master@{#577429}

mojo/public/java/bindings/src/org/chromium/mojo/bindings/Interface.java

@@ -20,6 +20,7 @@ import org.chromium.mojo.system.MojoException;
 import org.chromium.mojo.system.Pair;
 
 import java.io.Closeable;
+import java.util.concurrent.Executor;
 
 /**
  * Base class for mojo generated interfaces.
@@ -317,6 +318,67 @@ public interface Interface extends ConnectionErrorHandler, Closeable {
 
     }
 
+    /**
+     * A {@link MessageReceiverWithResponder} implementation that forwards all calls to the thread
+     * the ThreadSafeForwarder was created.
+     */
+    class ThreadSafeForwarder implements MessageReceiverWithResponder {
+
+        /**
+         * The {@link MessageReceiverWithResponder} that will receive a serialized message for
+         * each method call.
+         */
+        private final MessageReceiverWithResponder mMessageReceiver;
+
+        /**
+         * The {@link Executor} to forward all tasks to.
+         */
+        private final Executor mExecutor;
+
+        /**
+         * Constructor.
+         *
+         * @param core the Core implementation used to create pipes and access the async waiter.
+         * @param messageReceiver the message receiver to send message to.
+         */
+        public ThreadSafeForwarder(Core core, MessageReceiverWithResponder messageReceiver) {
+            mMessageReceiver = messageReceiver;
+            mExecutor = ExecutorFactory.getExecutorForCurrentThread(core);
+        }
+
+        /**
+         * @see org.chromium.mojo.bindings.MessageReceiver#close()
+         */
+        @Override
+        public void close() {
+            mExecutor.execute(() -> {
+                mMessageReceiver.close();
+            });
+        }
+
+        /**
+         * @see org.chromium.mojo.bindings.MessageReceiver#accept()
+         */
+        @Override
+        public boolean accept(Message message) {
+            mExecutor.execute(() -> {
+                mMessageReceiver.accept(message);
+            });
+            return true;
+        }
+
+        /**
+         * @see org.chromium.mojo.bindings.MessageReceiverWithResponder#acceptWithResponder()
+         */
+        @Override
+        public boolean acceptWithResponder(Message message, MessageReceiver responder) {
+            mExecutor.execute(() -> {
+                mMessageReceiver.acceptWithResponder(message, responder);
+            });
+            return true;
+        }
+    }
+
     /**
      * The |Manager| object enables building of proxies and stubs for a given interface.
      *
@@ -385,6 +447,32 @@ public interface Interface extends ConnectionErrorHandler, Closeable {
             return new InterfaceRequest<I>(handle);
         }
 
+        /**
+         * Constructs a thread-safe Proxy forwarding the calls to the given message receiver.
+         * All calls can be performed from any thread and are posted to the {@link Executor} that
+         * is associated with the thread on which this method was called on.
+         *
+         * The original Proxy object is unbound.
+         */
+        public final P buildThreadSafeProxy(P proxy) {
+            HandlerImpl handlerImpl = (HandlerImpl) proxy.getProxyHandler();
+            Core core = handlerImpl.getCore();
+            int version = handlerImpl.getVersion();
+
+            Router router = new RouterImpl(handlerImpl.passHandle());
+            // Close the original proxy now that its handle has been passed.
+            proxy.close();
+
+            proxy = buildProxy(
+                core, new ThreadSafeForwarder(core, new AutoCloseableRouter(core, router)));
+            DelegatingConnectionErrorHandler handlers = new DelegatingConnectionErrorHandler();
+            handlers.addConnectionErrorHandler(proxy);
+            router.setErrorHandler(handlers);
+            router.start();
+            ((HandlerImpl) proxy.getProxyHandler()).setVersion(version);
+            return proxy;
+        }
+
         /**
          * Binds the implementation to the given |router|.
          */