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

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

 // Copyright 2013 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.
 
 #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()),
       incoming_receiver_(nullptr),
       async_wait_id_(0),
       error_(false),
       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();
 }
 
 Connector::~Connector() {
   if (destroyed_flag_)
     *destroyed_flag_ = true;
 
   CancelWait();
 }
 
 void Connector::CloseMessagePipe() {
   CancelWait();
+  MayAutoLock locker(lock_.get());
   Close(message_pipe_.Pass());
 }
 
 ScopedMessagePipeHandle Connector::PassMessagePipe() {
   CancelWait();
+  MayAutoLock locker(lock_.get());
   return message_pipe_.Pass();
 }
 
 void Connector::RaiseError() {
   HandleError(true, true);
 }
 
 bool Connector::WaitForIncomingMessage(MojoDeadline deadline) {
   if (error_)
     return false;
@@ skipping 24 matching lines @@
 
 void Connector::ResumeIncomingMethodCallProcessing() {
   if (!paused_)
     return;
 
   paused_ = false;
   WaitToReadMore();
 }
 
 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 =
       WriteMessageRaw(message_pipe_.get(),
                       message->data(),
                       message->data_num_bytes(),
                       message->mutable_handles()->empty()
                           ? nullptr
                           : reinterpret_cast<const MojoHandle*>(
                                 &message->mutable_handles()->front()),
@@ skipping 127 matching lines @@
     force_pipe_reset = true;
 
   if (paused_) {
     // If the user has paused receiving messages, we shouldn't call the error
     // handler right away. We need to wait until the user starts receiving
     // messages again.
     force_async_handler = true;
   }
 
   if (force_pipe_reset) {
-    CloseMessagePipe();
+    CancelWait();
+    MayAutoLock locker(lock_.get());
+    Close(message_pipe_.Pass());
     MessagePipe dummy_pipe;
     message_pipe_ = dummy_pipe.handle0.Pass();
   } else {
     CancelWait();
   }
 
   if (force_async_handler) {
     // |dummy_pipe.handle1| has been destructed. Reading the pipe will
     // eventually cause a read error on |message_pipe_| and set error state.
     if (!paused_)
       WaitToReadMore();
   } else {
     error_ = true;
     connection_error_handler_.Run();
   }
 }
 
 }  // namespace internal
 }  // namespace mojo