Adds Channel-associated interface support on ChannelProxy's thread

ipc/ipc_mojo_bootstrap.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 "ipc/ipc_mojo_bootstrap.h"
 
 #include <stdint.h>
 
 #include <map>
 #include <memory>
@@ skipping 103 matching lines @@
   }
 
   void CloseEndpointHandle(mojo::InterfaceId id, bool is_local) override {
     if (!mojo::IsValidInterfaceId(id))
       return;
 
     base::AutoLock locker(lock_);
     if (!is_local) {
       DCHECK(ContainsKey(endpoints_, id));
       DCHECK(!mojo::IsMasterInterfaceId(id));
-      control_message_proxy_.NotifyEndpointClosedBeforeSent(id);
+      NotifyEndpointClosedBeforeSent(id);
       return;
     }
 
     DCHECK(ContainsKey(endpoints_, id));
     Endpoint* endpoint = endpoints_[id].get();
     DCHECK(!endpoint->client());
     DCHECK(!endpoint->closed());
     MarkClosedAndMaybeRemove(endpoint);
 
     if (!mojo::IsMasterInterfaceId(id))
-      control_message_proxy_.NotifyPeerEndpointClosed(id);
+      NotifyPeerEndpointClosed(id);
   }
 
   mojo::InterfaceEndpointController* AttachEndpointClient(
       const mojo::ScopedInterfaceEndpointHandle& handle,
       mojo::InterfaceEndpointClient* client,
       scoped_refptr<base::SingleThreadTaskRunner> runner) override {
     const mojo::InterfaceId id = handle.id();
 
     DCHECK(mojo::IsValidInterfaceId(id));
     DCHECK(client);
@@ skipping 239 matching lines @@
       endpoints_.erase(endpoint->id());
   }
 
   void MarkPeerClosedAndMaybeRemove(Endpoint* endpoint) {
     lock_.AssertAcquired();
     endpoint->set_peer_closed();
     if (endpoint->closed() && endpoint->peer_closed())
       endpoints_.erase(endpoint->id());
   }
 
+  void NotifyPeerEndpointClosed(mojo::InterfaceId id) {
+    if (task_runner_->BelongsToCurrentThread()) {
+      if (connector_.is_valid())
+        control_message_proxy_.NotifyPeerEndpointClosed(id);
+    } else {
+      task_runner_->PostTask(
+          FROM_HERE,
+          base::Bind(&ChannelAssociatedGroupController
+              ::NotifyPeerEndpointClosed, this, id));
+    }
+  }
+
+  void NotifyEndpointClosedBeforeSent(mojo::InterfaceId id) {
+    if (task_runner_->BelongsToCurrentThread()) {
+      if (connector_.is_valid())
+        control_message_proxy_.NotifyEndpointClosedBeforeSent(id);
+    } else {
+      task_runner_->PostTask(
+          FROM_HERE,
+          base::Bind(&ChannelAssociatedGroupController
+              ::NotifyEndpointClosedBeforeSent, this, id));
+    }
+  }
+
   Endpoint* FindOrInsertEndpoint(mojo::InterfaceId id, bool* inserted) {
     lock_.AssertAcquired();
     DCHECK(!inserted || !*inserted);
 
     auto iter = endpoints_.find(id);
     if (iter != endpoints_.end())
       return iter->second.get();
 
     Endpoint* endpoint = new Endpoint(this, id);
     endpoints_.insert({ id, endpoint });
     if (inserted)
       *inserted = true;
     return endpoint;
   }
 
   // mojo::MessageReceiver:
   bool Accept(mojo::Message* message) override {
     DCHECK(thread_checker_.CalledOnValidThread());
 
     if (mojo::PipeControlMessageHandler::IsPipeControlMessage(message)) {
       if (!control_message_handler_.Accept(message))

[yzshen1, 2016/07/14 15:04:04]

can we directly return false instead of raising error? The boolean's meaning is
different from that of MultiplexRouter::ProcessIncomingMessage

[Ken Rockot(use gerrit already), 2016/07/14 15:27:37]

Good point! Done

         RaiseError();
       return true;
     }
 
     mojo::InterfaceId id = message->interface_id();
     DCHECK(mojo::IsValidInterfaceId(id));
 
     base::AutoLock locker(lock_);
-    bool inserted = false;
-    Endpoint* endpoint = FindOrInsertEndpoint(id, &inserted);
-    if (inserted) {
-      MarkClosedAndMaybeRemove(endpoint);
-      if (!mojo::IsMasterInterfaceId(id))
-        control_message_proxy_.NotifyPeerEndpointClosed(id);
-      return true;
-    }
-
-    if (endpoint->closed())
+    Endpoint* endpoint = GetEndpointForDispatch(id);
+    if (!endpoint)
       return true;
 
     mojo::InterfaceEndpointClient* client = endpoint->client();
     if (!client || !endpoint->task_runner()->BelongsToCurrentThread()) {
       // No client has been bound yet or the client runs tasks on another
       // thread. We assume the other thread must always be the one on which
       // |proxy_task_runner_| runs tasks, since that's the only valid scenario.
       //
       // If the client is not yet bound, it must be bound by the time this task
       // runs or else it's programmer error.
       DCHECK(proxy_task_runner_);
-      CHECK(false);
       std::unique_ptr<mojo::Message> passed_message(new mojo::Message);
       message->MoveTo(passed_message.get());
       proxy_task_runner_->PostTask(
           FROM_HERE,
           base::Bind(&ChannelAssociatedGroupController::AcceptOnProxyThread,
                      this, base::Passed(&passed_message)));
       return true;
     }
 
     // We do not expect to receive sync responses on the master endpoint thread.
     // If it's happening, it's a bug.
     DCHECK(!message->has_flag(mojo::Message::kFlagIsSync));
 
     bool result = false;
     {
       base::AutoUnlock unlocker(lock_);
       result = client->HandleIncomingMessage(message);
     }
 
     if (!result)
       RaiseError();

[yzshen1, 2016/07/14 15:04:04]

I think we could return false directly here?

[Ken Rockot(use gerrit already), 2016/07/14 15:27:37]

Yep, done

 
     return true;
   }
 
   void AcceptOnProxyThread(std::unique_ptr<mojo::Message> message) {
     DCHECK(proxy_task_runner_->BelongsToCurrentThread());
 
-    // TODO(rockot): Implement this.
-    NOTREACHED();
+    mojo::InterfaceId id = message->interface_id();
+    DCHECK(mojo::IsValidInterfaceId(id) && !mojo::IsMasterInterfaceId(id));
+
+    base::AutoLock locker(lock_);
+    Endpoint* endpoint = GetEndpointForDispatch(id);
+    if (!endpoint)
+      return;
+
+    mojo::InterfaceEndpointClient* client = endpoint->client();
+    if (!client)
+      return;
+
+    DCHECK(endpoint->task_runner()->BelongsToCurrentThread());
+
+    // TODO(rockot): Implement sync dispatch. For now, sync messages are
+    // unsupported here.
+    DCHECK(!message->has_flag(mojo::Message::kFlagIsSync));
+
+    bool result = false;
+    {
+      base::AutoUnlock unlocker(lock_);
+      result = client->HandleIncomingMessage(message.get());
+    }
+
+    if (!result)
+      RaiseError();
+  }
+
+  Endpoint* GetEndpointForDispatch(mojo::InterfaceId id) {
+    lock_.AssertAcquired();
+    bool inserted = false;
+    Endpoint* endpoint = FindOrInsertEndpoint(id, &inserted);
+    if (inserted) {
+      MarkClosedAndMaybeRemove(endpoint);
+      if (!mojo::IsMasterInterfaceId(id))
+        NotifyPeerEndpointClosed(id);
+      return nullptr;
+    }
+
+    if (endpoint->closed())
+      return nullptr;
+
+    return endpoint;
   }
 
   // mojo::PipeControlMessageHandlerDelegate:
   bool OnPeerAssociatedEndpointClosed(mojo::InterfaceId id) override {
     DCHECK(thread_checker_.CalledOnValidThread());
 
     if (mojo::IsMasterInterfaceId(id))
       return false;
 
     base::AutoLock locker(lock_);
@@ skipping 68 matching lines @@
   void set_connection_error_handler(const base::Closure& handler) {
     DCHECK(endpoint_client_);
     endpoint_client_->set_connection_error_handler(handler);
   }
 
   mojo::AssociatedGroup* associated_group() {
     DCHECK(controller_);
     return controller_->associated_group();
   }
 
+  ChannelAssociatedGroupController* controller() {
+    DCHECK(controller_);
+    return controller_.get();
+  }
+
   mojom::Bootstrap* operator->() {
     DCHECK(proxy_);
     return proxy_.get();
   }
 
  private:
   std::unique_ptr<mojom::BootstrapProxy> proxy_;
   scoped_refptr<ChannelAssociatedGroupController> controller_;
   std::unique_ptr<mojo::InterfaceEndpointClient> endpoint_client_;
 
@@ skipping 15 matching lines @@
   void set_connection_error_handler(const base::Closure& handler) {
     DCHECK(endpoint_client_);
     endpoint_client_->set_connection_error_handler(handler);
   }
 
   mojo::AssociatedGroup* associated_group() {
     DCHECK(controller_);
     return controller_->associated_group();
   }
 
+  ChannelAssociatedGroupController* controller() {
+    DCHECK(controller_);
+    return controller_.get();
+  }
+
   void Bind(mojo::ScopedMessagePipeHandle handle) {
     DCHECK(!controller_);
     controller_ =
         new ChannelAssociatedGroupController(false, std::move(handle));
     stub_.serialization_context()->group_controller = controller_;
 
     endpoint_client_.reset(new mojo::InterfaceEndpointClient(
         controller_->CreateLocalEndpointHandle(mojo::kMasterInterfaceId),
         &stub_,
         base::MakeUnique<typename mojom::Bootstrap::RequestValidator_>(),
@@ skipping 10 matching lines @@
 
 // MojoBootstrap for the server process. You should create the instance
 // using MojoBootstrap::Create().
 class MojoServerBootstrap : public MojoBootstrap {
  public:
   MojoServerBootstrap();
 
  private:
   // MojoBootstrap implementation.
   void Connect() override;
+
   mojo::AssociatedGroup* GetAssociatedGroup() override {
     return bootstrap_.associated_group();
   }
 
+  void SetProxyTaskRunner(
+      scoped_refptr<base::SingleThreadTaskRunner> task_runner) override {
+    bootstrap_.controller()->SetProxyTaskRunner(task_runner);
+  }
+
   void OnInitDone(int32_t peer_pid);
 
   BootstrapMasterProxy bootstrap_;
   IPC::mojom::ChannelAssociatedPtrInfo send_channel_;
   IPC::mojom::ChannelAssociatedRequest receive_channel_request_;
 
   DISALLOW_COPY_AND_ASSIGN(MojoServerBootstrap);
 };
 
 MojoServerBootstrap::MojoServerBootstrap() = default;
@@ skipping 38 matching lines @@
 
 // MojoBootstrap for client processes. You should create the instance
 // using MojoBootstrap::Create().
 class MojoClientBootstrap : public MojoBootstrap, public mojom::Bootstrap {
  public:
   MojoClientBootstrap();
 
  private:
   // MojoBootstrap implementation.
   void Connect() override;
+
   mojo::AssociatedGroup* GetAssociatedGroup() override {
     return binding_.associated_group();
   }
 
+  void SetProxyTaskRunner(
+      scoped_refptr<base::SingleThreadTaskRunner> task_runner) override {
+    binding_.controller()->SetProxyTaskRunner(task_runner);
+  }
+
   // mojom::Bootstrap implementation.
   void Init(mojom::ChannelAssociatedRequest receive_channel,
             mojom::ChannelAssociatedPtrInfo send_channel,
             int32_t peer_pid,
             const InitCallback& callback) override;
 
   BootstrapMasterBinding binding_;
 
   DISALLOW_COPY_AND_ASSIGN(MojoClientBootstrap);
 };
@@ skipping 66 matching lines @@
 
 bool MojoBootstrap::HasFailed() const {
   return state() == STATE_ERROR;
 }
 
 mojo::ScopedMessagePipeHandle MojoBootstrap::TakeHandle() {
   return std::move(handle_);
 }
 
 }  // namespace IPC