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

mojo/public/cpp/bindings/lib/connector.cc

Index: mojo/public/cpp/bindings/lib/connector.cc
diff --git a/mojo/public/cpp/bindings/lib/connector.cc b/mojo/public/cpp/bindings/lib/connector.cc
index 5bd94eb63ae94542a2d131affb8d97b6704e8e3b..2fa50af5cead3b0ff83484e28cae245a19dbc31c 100644
--- a/mojo/public/cpp/bindings/lib/connector.cc
+++ b/mojo/public/cpp/bindings/lib/connector.cc
@@ -4,14 +4,42 @@
 
 #include "mojo/public/cpp/bindings/lib/connector.h"
 
+#include "base/macros.h"
+#include "base/synchronization/lock.h"
 #include "mojo/public/cpp/environment/logging.h"
 
 namespace mojo {
 namespace internal {
 
+namespace {
+
+// Similar to base::AutoLock, except that it does nothing if |lock| passed into
+// the constructor is null.
+class MayAutoLock {
+ public:
+  explicit MayAutoLock(base::Lock* lock) : lock_(lock) {
+    if (lock_)
+      lock_->Acquire();
+  }
+
+  ~MayAutoLock() {
+    if (lock_) {
+      lock_->AssertAcquired();
+      lock_->Release();
+    }
+  }
+
+ private:
+  base::Lock* lock_;
+  DISALLOW_COPY_AND_ASSIGN(MayAutoLock);
+};
+
+}  // namespace
+
 // ----------------------------------------------------------------------------
 
 Connector::Connector(ScopedMessagePipeHandle message_pipe,
+                     SendingThreadSaftyType sending_thread_safty_type,
                      const MojoAsyncWaiter* waiter)
     : waiter_(waiter),
       message_pipe_(message_pipe.Pass()),
@@ -21,7 +49,9 @@ Connector::Connector(ScopedMessagePipeHandle message_pipe,
       drop_writes_(false),
       enforce_errors_from_incoming_receiver_(true),
       paused_(false),
-      destroyed_flag_(nullptr) {
+      destroyed_flag_(nullptr),
+      lock_(sending_thread_safty_type == MULTI_THREADED_SEND ? new base::Lock
+                                                             : nullptr) {
   // Even though we don't have an incoming receiver, we still want to monitor
   // the message pipe to know if is closed or encounters an error.
   WaitToReadMore();
@@ -36,11 +66,13 @@ Connector::~Connector() {
 
 void Connector::CloseMessagePipe() {
   CancelWait();
+  MayAutoLock locker(lock_.get());
   Close(message_pipe_.Pass());
 }
 
 ScopedMessagePipeHandle Connector::PassMessagePipe() {
   CancelWait();
+  MayAutoLock locker(lock_.get());
   return message_pipe_.Pass();
 }
 
@@ -85,11 +117,15 @@ void Connector::ResumeIncomingMethodCallProcessing() {
 }
 
 bool Connector::Accept(Message* message) {
+  // It shouldn't hurt even if |error_| may be changed by a different thread at
+  // the same time. The outcome is that we may write into |message_pipe_| after
+  // encountering an error, which should be fine.
   if (error_)
     return false;
 
-  MOJO_CHECK(message_pipe_.is_valid());
-  if (drop_writes_)
+  MayAutoLock locker(lock_.get());
+
+  if (!message_pipe_.is_valid() || drop_writes_)
     return true;
 
   MojoResult rv =
@@ -237,7 +273,9 @@ void Connector::HandleError(bool force_pipe_reset, bool force_async_handler) {
   }
 
   if (force_pipe_reset) {
-    CloseMessagePipe();
+    CancelWait();
+    MayAutoLock locker(lock_.get());
+    Close(message_pipe_.Pass());
     MessagePipe dummy_pipe;
     message_pipe_ = dummy_pipe.handle0.Pass();
   } else {