Mojo C++ bindings: support sync methods - part 2

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

 // 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.
 
 #include "mojo/public/cpp/bindings/lib/router.h"
 
 #include <stdint.h>
 #include <utility>
 
+#include "base/bind.h"
 #include "base/logging.h"
+#include "base/message_loop/message_loop.h"
+#include "base/stl_util.h"
 
 namespace mojo {
 namespace internal {
 
 // ----------------------------------------------------------------------------
 
 namespace {
 
 class ResponderThunk : public MessageReceiverWithStatus {
  public:
@@ skipping 32 matching lines @@
 
  private:
   base::WeakPtr<Router> router_;
   bool accept_was_invoked_;
 };
 
 }  // namespace
 
 // ----------------------------------------------------------------------------
 
+Router::SyncResponseInfo::SyncResponseInfo(bool* in_response_received)
+    : response_received(in_response_received) {}
+
+Router::SyncResponseInfo::~SyncResponseInfo() {}
+
+// ----------------------------------------------------------------------------
+
 Router::HandleIncomingMessageThunk::HandleIncomingMessageThunk(Router* router)
     : router_(router) {
 }
 
 Router::HandleIncomingMessageThunk::~HandleIncomingMessageThunk() {
 }
 
 bool Router::HandleIncomingMessageThunk::Accept(Message* message) {
   return router_->HandleIncomingMessage(message);
 }
 
 // ----------------------------------------------------------------------------
 
 Router::Router(ScopedMessagePipeHandle message_pipe,
                FilterChain filters,
+               bool expects_sync_requests,
                const MojoAsyncWaiter* waiter)
     : thunk_(this),
       filters_(std::move(filters)),
       connector_(std::move(message_pipe),
                  Connector::SINGLE_THREADED_SEND,
                  waiter),
       incoming_receiver_(nullptr),
       next_request_id_(0),
       testing_mode_(false),
+      pending_task_for_messages_(false),
       weak_factory_(this) {
   filters_.SetSink(&thunk_);
+  if (expects_sync_requests)
+    connector_.RegisterSyncHandleWatch();
   connector_.set_incoming_receiver(filters_.GetHead());
 }
 
-Router::~Router() {
-  weak_factory_.InvalidateWeakPtrs();
-
-  for (auto& pair : async_responders_)
-    delete pair.second;
-  for (auto& pair : sync_responders_)
-    delete pair.second;
-}
+Router::~Router() {}
 
 bool Router::Accept(Message* message) {
   DCHECK(thread_checker_.CalledOnValidThread());
   DCHECK(!message->has_flag(kMessageExpectsResponse));
   return connector_.Accept(message);
 }
 
 bool Router::AcceptWithResponder(Message* message, MessageReceiver* responder) {
   DCHECK(thread_checker_.CalledOnValidThread());
   DCHECK(message->has_flag(kMessageExpectsResponse));
 
   // Reserve 0 in case we want it to convey special meaning in the future.
   uint64_t request_id = next_request_id_++;
   if (request_id == 0)
     request_id = next_request_id_++;
 
   message->set_request_id(request_id);
   if (!connector_.Accept(message))
     return false;
 
   if (!message->has_flag(kMessageIsSync)) {
     // We assume ownership of |responder|.
-    async_responders_[request_id] = responder;
+    async_responders_[request_id] = make_scoped_ptr(responder);
     return true;
   }
 
-  sync_responders_[request_id] = responder;
+  if (!connector_.RegisterSyncHandleWatch())
+    return false;
+
+  bool response_received = false;
+  scoped_ptr<MessageReceiver> sync_responder(responder);
+  sync_responses_.insert(std::make_pair(
+      request_id, make_scoped_ptr(new SyncResponseInfo(&response_received))));
+
   base::WeakPtr<Router> weak_self = weak_factory_.GetWeakPtr();
-  for (;;) {
-    // TODO(yzshen): Here we should allow incoming sync requests to re-enter and
-    // block async messages.
-    bool result = WaitForIncomingMessage(MOJO_DEADLINE_INDEFINITE);
-    // The message pipe has disconnected.
+  do {
+    bool result = connector_.RunSyncHandleWatch(&response_received);
     if (!result)
       break;
 
     // This instance has been destroyed.
     if (!weak_self)
       break;
 
     // The corresponding response message has arrived.
-    if (sync_responders_.find(request_id) == sync_responders_.end())
-      break;
-  }
+    DCHECK(response_received);
+    DCHECK(ContainsKey(sync_responses_, request_id));
+    auto iter = sync_responses_.find(request_id);
+    DCHECK_EQ(&response_received, iter->second->response_received);
+    scoped_ptr<Message> response = std::move(iter->second->response);
+    sync_responses_.erase(iter);
+    ignore_result(sync_responder->Accept(response.get()));
+  } while (false);
+
+  if (weak_self)
+    connector_.UnregisterSyncHandleWatch();
 
   // Return true means that we take ownership of |responder|.
   return true;
 }
 
 void Router::EnableTestingMode() {
   DCHECK(thread_checker_.CalledOnValidThread());
   testing_mode_ = true;
   connector_.set_enforce_errors_from_incoming_receiver(false);
 }
 
 bool Router::HandleIncomingMessage(Message* message) {
   DCHECK(thread_checker_.CalledOnValidThread());
+
+  const bool during_sync_call =
+      connector_.during_sync_handle_watcher_callback();
+  if (!message->has_flag(kMessageIsSync) &&
+      (during_sync_call || !pending_messages_.empty())) {
+    scoped_ptr<Message> pending_message(new Message);
+    message->MoveTo(pending_message.get());
+    pending_messages_.push(std::move(pending_message));
+
+    if (!pending_task_for_messages_) {
+      pending_task_for_messages_ = true;
+      base::MessageLoop::current()->PostTask(
+          FROM_HERE, base::Bind(&Router::HandleQueuedMessages,
+                                weak_factory_.GetWeakPtr()));
+    }
+
+    return true;
+  }
+
+  return HandleMessageInternal(message);
+}
+
+void Router::HandleQueuedMessages() {
+  DCHECK(thread_checker_.CalledOnValidThread());
+  DCHECK(pending_task_for_messages_);
+
+  base::WeakPtr<Router> weak_self = weak_factory_.GetWeakPtr();
+  while (!pending_messages_.empty()) {
+    scoped_ptr<Message> message(std::move(pending_messages_.front()));
+    pending_messages_.pop();
+
+    bool result = HandleMessageInternal(message.get());
+    if (!weak_self)
+      return;
+
+    if (!result && !testing_mode_) {
+      connector_.RaiseError();
+      break;
+    }
+  }
+
+  pending_task_for_messages_ = false;
+}
+
+bool Router::HandleMessageInternal(Message* message) {
   if (message->has_flag(kMessageExpectsResponse)) {
     if (!incoming_receiver_)
       return false;
 
     MessageReceiverWithStatus* responder =
         new ResponderThunk(weak_factory_.GetWeakPtr());
     bool ok = incoming_receiver_->AcceptWithResponder(message, responder);
     if (!ok)
       delete responder;
     return ok;
 
   } else if (message->has_flag(kMessageIsResponse)) {
-    ResponderMap& responder_map = message->has_flag(kMessageIsSync)
-                                      ? sync_responders_
-                                      : async_responders_;
     uint64_t request_id = message->request_id();
-    ResponderMap::iterator it = responder_map.find(request_id);
-    if (it == responder_map.end()) {
+
+    if (message->has_flag(kMessageIsSync)) {
+      auto it = sync_responses_.find(request_id);
+      if (it == sync_responses_.end()) {
+        DCHECK(testing_mode_);
+        return false;
+      }
+      it->second->response.reset(new Message());
+      message->MoveTo(it->second->response.get());
+      *it->second->response_received = true;
+      return true;
+    }
+
+    auto it = async_responders_.find(request_id);
+    if (it == async_responders_.end()) {
       DCHECK(testing_mode_);
       return false;
     }
-    MessageReceiver* responder = it->second;
-    responder_map.erase(it);
-    bool ok = responder->Accept(message);
-    delete responder;
-    return ok;
+    scoped_ptr<MessageReceiver> responder = std::move(it->second);
+    async_responders_.erase(it);
+    return responder->Accept(message);
   } else {
     if (!incoming_receiver_)
       return false;
 
     return incoming_receiver_->Accept(message);
   }
 }
 
 // ----------------------------------------------------------------------------
 
 }  // namespace internal
 }  // namespace mojo