Mojo C++ bindings: sync call support for associated interfaces and master interfaces

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

 // 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.
 
 #include "mojo/public/cpp/bindings/lib/interface_endpoint_client.h"
 
 #include <stdint.h>
 #include <utility>
 
 #include "base/bind.h"
 #include "base/location.h"
 #include "base/macros.h"
 #include "base/message_loop/message_loop.h"
 #include "base/single_thread_task_runner.h"
 #include "base/stl_util.h"
 #include "base/thread_task_runner_handle.h"
 #include "mojo/public/cpp/bindings/associated_group.h"
+#include "mojo/public/cpp/bindings/lib/interface_endpoint_controller.h"
 #include "mojo/public/cpp/bindings/lib/multiplex_router.h"
 
 namespace mojo {
 namespace internal {
 
 // ----------------------------------------------------------------------------
 
 namespace {
 
 void DCheckIfInvalid(const base::WeakPtr<InterfaceEndpointClient>& client,
@@ skipping 65 matching lines @@
   bool accept_was_invoked_;
   scoped_refptr<base::SingleThreadTaskRunner> task_runner_;
 
   DISALLOW_COPY_AND_ASSIGN(ResponderThunk);
 };
 
 }  // namespace
 
 // ----------------------------------------------------------------------------
 
+InterfaceEndpointClient::SyncResponseInfo::SyncResponseInfo(
+    bool* in_response_received)
+    : response_received(in_response_received) {}
+
+InterfaceEndpointClient::SyncResponseInfo::~SyncResponseInfo() {}
+
+// ----------------------------------------------------------------------------
+
 InterfaceEndpointClient::HandleIncomingMessageThunk::HandleIncomingMessageThunk(
     InterfaceEndpointClient* owner)
     : owner_(owner) {}
 
 InterfaceEndpointClient::HandleIncomingMessageThunk::
     ~HandleIncomingMessageThunk() {}
 
 bool InterfaceEndpointClient::HandleIncomingMessageThunk::Accept(
     Message* message) {
   return owner_->HandleValidatedMessage(message);
 }
 
 // ----------------------------------------------------------------------------
 
 InterfaceEndpointClient::InterfaceEndpointClient(
     ScopedInterfaceEndpointHandle handle,
     MessageReceiverWithResponderStatus* receiver,
-    scoped_ptr<MessageFilter> payload_validator)
+    scoped_ptr<MessageFilter> payload_validator,
+    bool expect_sync_requests)
     : handle_(std::move(handle)),
       incoming_receiver_(receiver),
       payload_validator_(std::move(payload_validator)),
       thunk_(this),
       next_request_id_(1),
       encountered_error_(false),
       weak_ptr_factory_(this) {
   DCHECK(handle_.is_valid());
   DCHECK(handle_.is_local());
 
   // TODO(yzshen): the way to use validator (or message filter in general)
   // directly is a little awkward.
   payload_validator_->set_sink(&thunk_);
 
-  handle_.router()->AttachEndpointClient(handle_, this);
+  controller_ = handle_.router()->AttachEndpointClient(handle_, this);
+  if (expect_sync_requests)
+    controller_->AllowWokenUpBySyncWatchOnSameThread();
 }
 
 InterfaceEndpointClient::~InterfaceEndpointClient() {
   DCHECK(thread_checker_.CalledOnValidThread());
 
-  STLDeleteValues(&responders_);
-
   handle_.router()->DetachEndpointClient(handle_);
 }
 
 AssociatedGroup* InterfaceEndpointClient::associated_group() {
   if (!associated_group_)
     associated_group_ = handle_.router()->CreateAssociatedGroup();
   return associated_group_.get();
 }
 
 uint32_t InterfaceEndpointClient::interface_id() const {
   DCHECK(thread_checker_.CalledOnValidThread());
   return handle_.id();
 }
 
 ScopedInterfaceEndpointHandle InterfaceEndpointClient::PassHandle() {
   DCHECK(thread_checker_.CalledOnValidThread());
   DCHECK(!has_pending_responders());
 
   if (!handle_.is_valid())
     return ScopedInterfaceEndpointHandle();
 
+  controller_ = nullptr;
   handle_.router()->DetachEndpointClient(handle_);
 
   return std::move(handle_);
 }
 
 void InterfaceEndpointClient::RaiseError() {
   DCHECK(thread_checker_.CalledOnValidThread());
 
   handle_.router()->RaiseError();
 }
 
 bool InterfaceEndpointClient::Accept(Message* message) {
   DCHECK(thread_checker_.CalledOnValidThread());
+  DCHECK(controller_);
   DCHECK(!message->has_flag(kMessageExpectsResponse));
 
   if (encountered_error_)
     return false;
 
-  return handle_.router()->SendMessage(handle_, message);
+  return controller_->SendMessage(message);
 }
 
 bool InterfaceEndpointClient::AcceptWithResponder(Message* message,
                                                   MessageReceiver* responder) {
   DCHECK(thread_checker_.CalledOnValidThread());
+  DCHECK(controller_);
   DCHECK(message->has_flag(kMessageExpectsResponse));
 
-  // TODO(yzshen): Sync method call using assoicated interfaces or master
-  // interfaces that serve associated interfaces hasn't been supported yet.
-  if (message->has_flag(kMessageIsSync)) {
-    NOTIMPLEMENTED();
-    return false;
-  }
-
   if (encountered_error_)
     return false;
 
   // 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 (!handle_.router()->SendMessage(handle_, message))
+  if (!controller_->SendMessage(message))
     return false;
 
-  // We assume ownership of |responder|.
-  responders_[request_id] = responder;
+  if (!message->has_flag(kMessageIsSync)) {
+    // We assume ownership of |responder|.
+    async_responders_[request_id] = make_scoped_ptr(responder);
+    return true;
+  }
+
+  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<InterfaceEndpointClient> weak_self =
+      weak_ptr_factory_.GetWeakPtr();
+  controller_->SyncWatch(&response_received);
+  // Make sure that this instance hasn't been destroyed.
+  if (weak_self) {
+    DCHECK(ContainsKey(sync_responses_, request_id));
+    auto iter = sync_responses_.find(request_id);
+    DCHECK_EQ(&response_received, iter->second->response_received);
+    if (response_received) {
+      scoped_ptr<Message> response = std::move(iter->second->response);
+      ignore_result(sync_responder->Accept(response.get()));
+    }
+    sync_responses_.erase(iter);
+  }
+
+  // Return true means that we take ownership of |responder|.
   return true;
 }
 
 bool InterfaceEndpointClient::HandleIncomingMessage(Message* message) {
   DCHECK(thread_checker_.CalledOnValidThread());
 
   return payload_validator_->Accept(message);
 }
 
 void InterfaceEndpointClient::NotifyError() {
@@ skipping 13 matching lines @@
       return false;
 
     MessageReceiverWithStatus* responder =
         new ResponderThunk(weak_ptr_factory_.GetWeakPtr());
     bool ok = incoming_receiver_->AcceptWithResponder(message, responder);
     if (!ok)
       delete responder;
     return ok;
   } else if (message->has_flag(kMessageIsResponse)) {
     uint64_t request_id = message->request_id();
-    ResponderMap::iterator it = responders_.find(request_id);
-    if (it == responders_.end())
+
+    if (message->has_flag(kMessageIsSync)) {
+      auto it = sync_responses_.find(request_id);
+      if (it == sync_responses_.end())
+        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())
       return false;
-    MessageReceiver* responder = it->second;
-    responders_.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