Mojo C++ bindings: introduce MultiplexRouter and related classes.

mojo/public/cpp/bindings/lib/multiplex_router.h

Index: mojo/public/cpp/bindings/lib/multiplex_router.h
diff --git a/mojo/public/cpp/bindings/lib/multiplex_router.h b/mojo/public/cpp/bindings/lib/multiplex_router.h
new file mode 100644
index 0000000000000000000000000000000000000000..cc8dab2528d07c8416dd56664556ea8338736a80
--- /dev/null
+++ b/mojo/public/cpp/bindings/lib/multiplex_router.h
@@ -0,0 +1,194 @@
+// Copyright 2015 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.
+
+#ifndef MOJO_PUBLIC_CPP_BINDINGS_LIB_MULTIPLEX_ROUTER_H_
+#define MOJO_PUBLIC_CPP_BINDINGS_LIB_MULTIPLEX_ROUTER_H_
+
+#include <deque>
+#include <map>
+
+#include "base/logging.h"
+#include "base/memory/ref_counted_delete_on_message_loop.h"
+#include "base/memory/scoped_ptr.h"
+#include "base/memory/weak_ptr.h"
+#include "base/synchronization/lock.h"
+#include "base/threading/thread_checker.h"
+#include "mojo/public/cpp/bindings/callback.h"
+#include "mojo/public/cpp/bindings/lib/connector.h"
+#include "mojo/public/cpp/bindings/lib/interface_id.h"
+#include "mojo/public/cpp/bindings/lib/message_header_validator.h"
+#include "mojo/public/cpp/bindings/lib/pipe_control_message_handler.h"
+#include "mojo/public/cpp/bindings/lib/pipe_control_message_handler_delegate.h"
+#include "mojo/public/cpp/bindings/lib/pipe_control_message_proxy.h"
+#include "mojo/public/cpp/bindings/lib/scoped_interface_endpoint_handle.h"
+#include "mojo/public/cpp/environment/environment.h"
+
+namespace mojo {
+namespace internal {
+
+class InterfaceEndpointClient;
+
+// MultiplexRouter supports routing messages for multiple interfaces over a
+// single message pipe.

[sky, 2015/11/19 17:16:27]

Please add a more general description of threading. You clearly define which
methods are intended for which thread, but an overview would be helpful.

[yzshen1, 2015/11/19 22:00:40]

Done.

+class MultiplexRouter
+    : public MessageReceiver,
+      public base::RefCountedDeleteOnMessageLoop<MultiplexRouter>,
+      public PipeControlMessageHandlerDelegate {
+ public:
+  // If |set_interface_id_namespace_bit| is true, the interface IDs generated by
+  // this router will have the highest bit set.
+  MultiplexRouter(
+      bool set_interface_id_namespace_bit,
+      ScopedMessagePipeHandle message_pipe,
+      const MojoAsyncWaiter* waiter = Environment::GetDefaultAsyncWaiter());
+
+  // ---------------------------------------------------------------------------
+  // The following public methods are safe to call from any threads.
+
+  // Creates a pair of interface endpoint handles. The method generates a new
+  // interface ID and assigns it to the two handles. |local_endpoint| is
+  // supposed to be used locally; while |remote_endpoint| is supposed to be sent

[sky, 2015/11/19 17:16:27]

You can remove 'supposed' to everywhere you have it in this file. Without it the
text reads more natural.

[yzshen1, 2015/11/19 22:00:40]

I see. Thanks!

+  // over the message pipe.
+  void CreateEndpointHandlePair(ScopedInterfaceEndpointHandle* local_endpoint,
+                                ScopedInterfaceEndpointHandle* remote_endpoint);
+
+  // Creates an interface endpoint handle from a given interface ID. The handle
+  // is supposed to be used locally.
+  // Typically, this method is used to (1) create an endpoint handle for the
+  // master interface; or (2) create an endpoint handle on receiving an
+  // interface ID from the message pipe.
+  ScopedInterfaceEndpointHandle CreateLocalEndpointHandle(InterfaceId id);
+
+  // Closes an interface endpoint handle.
+  void CloseEndpointHandle(InterfaceId id, bool is_local);
+
+  // Attaches an client to the specified endpoint to send and receive messages.
+  void AttachEndpointClient(const ScopedInterfaceEndpointHandle& handle,
+                            InterfaceEndpointClient* endpoint_client);
+  // Detaches the client attached to the specified endpoint.
+  void DetachEndpointClient(const ScopedInterfaceEndpointHandle& handle);
+
+  bool SendMessage(const ScopedInterfaceEndpointHandle& handle,
+                   Message* message);
+
+  // Raises an error on the underlying message pipe. It disconnects the pipe
+  // and notifies all interfaces running on this pipe.
+  void RaiseError();
+
+  // ---------------------------------------------------------------------------
+  // The following public methods are supposed to be called on the creating
+  // thread.
+
+  // Please note that this method shouldn't be called unless it results from an
+  // explicit request of the user of bindings (e.g., the user sets an
+  // InterfacePtr to null or closes a Binding).
+  void CloseMessagePipe() {
+    DCHECK(thread_checker_.CalledOnValidThread());
+    connector_.CloseMessagePipe();
+  }
+
+  // Extracts the underlying message pipe.
+  //
+  // TODO(yzshen): For now, users need to make sure there is no one holding on
+  // to associated interface endpoint handles at both sides of the message pipe
+  // in order to call this method. We need a way to forcefully invalidate
+  // associated interface endpoint handles.
+  ScopedMessagePipeHandle PassMessagePipe();
+
+  // Blocks the current thread until the first incoming message, or |deadline|.
+  bool WaitForIncomingMessage(MojoDeadline deadline) {
+    DCHECK(thread_checker_.CalledOnValidThread());
+    return connector_.WaitForIncomingMessage(deadline);
+  }
+
+  // See Binding for details of pause/resume.
+  void PauseIncomingMethodCallProcessing() {
+    DCHECK(thread_checker_.CalledOnValidThread());
+    connector_.PauseIncomingMethodCallProcessing();
+  }
+  void ResumeIncomingMethodCallProcessing() {
+    DCHECK(thread_checker_.CalledOnValidThread());
+    connector_.ResumeIncomingMethodCallProcessing();
+  }
+
+  // Sets this object to testing mode.
+  // In testing mode, the object doesn't disconnect the underlying message pipe
+  // when it receives unexpected or invalid messages.
+  void EnableTestingMode();
+
+  // Is the router bound to a message pipe handle?
+  bool is_valid() const {
+    DCHECK(thread_checker_.CalledOnValidThread());
+    return connector_.is_valid();
+  }
+
+ private:
+  friend class base::RefCountedDeleteOnMessageLoop<MultiplexRouter>;
+  friend class base::DeleteHelper<MultiplexRouter>;
+
+  class InterfaceEndpoint;
+  struct Task;
+
+  ~MultiplexRouter() override;
+
+  // MessageReceiver implementation:
+  bool Accept(Message* message) override;
+
+  // PipeControlMessageHandlerDelegate implementation:
+  bool OnPeerAssociatedEndpointClosed(InterfaceId id) override;
+  bool OnAssociatedEndpointClosedBeforeSent(InterfaceId id) override;
+
+  void OnPipeConnectionError();
+
+  // Processes enqueued tasks (incoming messages and error notifications).
+  // If |force_async| is true, it guarantees not to call any
+  // InterfaceEndpointClient methods directly.
+  //
+  // Note: Because calling into InterfaceEndpointClient may lead to destruction
+  // of this object, if |force_async| is set to false, the caller needs to hold
+  // on to a ref outside of |lock_| before calling this method.
+  void ProcessTasks(bool force_async);
+
+  // Returns true to indicate that |task| has been processed. Otherwise the task
+  // will be added back to the front of the queue.
+  // |*force_async| may be set to true to force subsequent tasks being processed
+  // in an asynchronous manner.
+  bool ProcessNotifyErrorTask(Task* task, bool* force_async);
+  bool ProcessIncomingMessageTask(Task* task, bool* force_async);
+
+  void LockAndCallProcessTasks();
+
+  void OnEndpointDestructed(InterfaceId id);
+
+  void RaiseErrorInNonTestingMode();
+
+  // Whether to set the namespace bit when generating interface IDs. Please see
+  // comments of kInterfaceIdNamespaceMask.
+  const bool set_interface_id_namespace_bit_;
+
+  MessageHeaderValidator header_validator_;
+  Connector connector_;
+
+  base::ThreadChecker thread_checker_;
+
+  // Protects the following members.
+  mutable base::Lock lock_;
+  PipeControlMessageHandler control_message_handler_;
+  PipeControlMessageProxy control_message_proxy_;
+
+  // Doesn't own the values. When an InterfaceEndpoint object is destructed, it
+  // notifies this object by calling OnEndpointDestructed().
+  std::map<InterfaceId, InterfaceEndpoint*> endpoints_;
+  uint32_t next_interface_id_value_;
+
+  // Owned objects.
+  std::deque<Task*> tasks_;

[sky, 2015/11/19 17:16:27]

Can you make this a std::deque<scoped_ptr>?

[yzshen1, 2015/11/19 22:00:41]

I don't think we can do that. IIUC, scoped_ptr moves using Pass() which is not
understood by std::deque. We may want to use the counterpart of scoped_ptr in
the stl library. But I would like to wait util the discussion around what C++
library features can be used settles. :)

[sky, 2015/11/20 01:00:19]

I believe scoped_ptr now supports the move operator.

+
+  bool testing_mode_;
+};

[sky, 2015/11/19 17:16:27]

DISALLOW..

[yzshen1, 2015/11/19 22:00:40]

Done.

+
+}  // namespace internal
+}  // namespace mojo
+
+#endif  // MOJO_PUBLIC_CPP_BINDINGS_LIB_MULTIPLEX_ROUTER_H_