Add a ChannelEndpointClient abstraction.

mojo/edk/system/channel_endpoint.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/edk/system/channel_endpoint.h"
 
 #include "base/logging.h"
 #include "mojo/edk/system/channel.h"
-#include "mojo/edk/system/message_pipe.h"
+#include "mojo/edk/system/channel_endpoint_client.h"
 #include "mojo/edk/system/transport_data.h"
 
 namespace mojo {
 namespace system {
 
-ChannelEndpoint::ChannelEndpoint(MessagePipe* message_pipe,
-                                 unsigned port,
+ChannelEndpoint::ChannelEndpoint(ChannelEndpointClient* client,
+                                 unsigned client_port,
                                  MessageInTransitQueue* message_queue)
-    : message_pipe_(message_pipe), port_(port), channel_(nullptr) {
-  DCHECK(message_pipe_.get() || message_queue);
-  DCHECK(port_ == 0 || port_ == 1);
+    : client_(client), client_port_(client_port), channel_(nullptr) {
+  DCHECK(client_.get() || message_queue);
 
   if (message_queue)
     paused_message_queue_.Swap(message_queue);
 }
 
 bool ChannelEndpoint::EnqueueMessage(scoped_ptr<MessageInTransit> message) {
   DCHECK(message);
 
   base::AutoLock locker(lock_);
 
   if (!channel_ || !remote_id_.is_valid()) {
     // We may reach here if we haven't been attached or run yet.
     // TODO(vtl): We may also reach here if the channel is shut down early for
     // some reason (with live message pipes on it). We can't check |state_| yet,
     // until it's protected under lock, but in this case we should return false
     // (and not enqueue any messages).
     paused_message_queue_.AddMessage(message.Pass());
     return true;
   }
 
   // TODO(vtl): Currently, this only works in the "running" case.
   DCHECK(remote_id_.is_valid());
 
   return WriteMessageNoLock(message.Pass());
 }
 
-void ChannelEndpoint::DetachFromMessagePipe() {
+void ChannelEndpoint::DetachFromClient() {
   {
     base::AutoLock locker(lock_);
-    DCHECK(message_pipe_.get());
-    message_pipe_ = nullptr;
+    DCHECK(client_.get());
+    client_ = nullptr;
 
     if (!channel_)
       return;
     DCHECK(local_id_.is_valid());
     DCHECK(remote_id_.is_valid());
     channel_->DetachEndpoint(this, local_id_, remote_id_);
     channel_ = nullptr;
     local_id_ = ChannelEndpointId();
     remote_id_ = ChannelEndpointId();
   }
@@ skipping 12 matching lines @@
   DCHECK(!remote_id_.is_valid());
   channel_ = channel;
   local_id_ = local_id;
   remote_id_ = remote_id;
 
   while (!paused_message_queue_.IsEmpty()) {
     LOG_IF(WARNING, !WriteMessageNoLock(paused_message_queue_.GetMessage()))
         << "Failed to write enqueue message to channel";
   }
 
-  if (!message_pipe_.get()) {
+  if (!client_.get()) {
     channel_->DetachEndpoint(this, local_id_, remote_id_);
     channel_ = nullptr;
     local_id_ = ChannelEndpointId();
     remote_id_ = ChannelEndpointId();
   }
 }
 
 bool ChannelEndpoint::OnReadMessage(
     const MessageInTransit::View& message_view,
     embedder::ScopedPlatformHandleVectorPtr platform_handles) {
   scoped_ptr<MessageInTransit> message(new MessageInTransit(message_view));
-  scoped_refptr<MessagePipe> message_pipe;
-  unsigned port;
+  scoped_refptr<ChannelEndpointClient> client;
+  unsigned client_port;
   {
     base::AutoLock locker(lock_);
     DCHECK(channel_);
-    if (!message_pipe_.get()) {
+    if (!client_.get()) {
       // This isn't a failure per se. (It just means that, e.g., the other end
       // of the message point closed first.)
       return true;
     }
 
     if (message_view.transport_data_buffer_size() > 0) {
       DCHECK(message_view.transport_data_buffer());
       message->SetDispatchers(TransportData::DeserializeDispatchers(
           message_view.transport_data_buffer(),
           message_view.transport_data_buffer_size(), platform_handles.Pass(),
           channel_));
     }
 
     // Take a ref, and call |EnqueueMessage()| outside the lock.
-    message_pipe = message_pipe_;
-    port = port_;
+    client = client_;
+    client_port = client_port_;
   }
 
-  MojoResult result = message_pipe->EnqueueMessage(
-      MessagePipe::GetPeerPort(port), message.Pass());
-  return (result == MOJO_RESULT_OK);
+  return client->OnReadMessage(client_port, message.Pass());
 }
 
 void ChannelEndpoint::DetachFromChannel() {
-  scoped_refptr<MessagePipe> message_pipe;
-  unsigned port;
+  scoped_refptr<ChannelEndpointClient> client;
+  unsigned client_port;
   {
     base::AutoLock locker(lock_);
 
-    if (message_pipe_.get()) {
-      // Take a ref, and call |Close()| outside the lock.
-      message_pipe = message_pipe_;
-      port = port_;
+    if (client_.get()) {
+      // Take a ref, and call |OnDetachFromChannel()| outside the lock.
+      client = client_;
+      client_port = client_port_;
     }
 
     // |channel_| may already be null if we already detached from the channel in
-    // |DetachFromMessagePipe()| by calling |Channel::DetachEndpoint()| (and
-    // there are racing detaches).
+    // |DetachFromClient()| by calling |Channel::DetachEndpoint()| (and there
+    // are racing detaches).
     if (channel_) {
       DCHECK(local_id_.is_valid());
       DCHECK(remote_id_.is_valid());
       channel_ = nullptr;
       local_id_ = ChannelEndpointId();
       remote_id_ = ChannelEndpointId();
     }
   }
 
-  if (message_pipe.get())
-    message_pipe->Close(port);
+  if (client.get())
+    client->OnDetachFromChannel(client_port);
 }
 
 ChannelEndpoint::~ChannelEndpoint() {
-  DCHECK(!message_pipe_.get());
+  DCHECK(!client_.get());
   DCHECK(!channel_);
   DCHECK(!local_id_.is_valid());
   DCHECK(!remote_id_.is_valid());
 }
 
 bool ChannelEndpoint::WriteMessageNoLock(scoped_ptr<MessageInTransit> message) {
   DCHECK(message);
 
   lock_.AssertAcquired();
 
   DCHECK(channel_);
   DCHECK(local_id_.is_valid());
   DCHECK(remote_id_.is_valid());
 
   message->SerializeAndCloseDispatchers(channel_);
   message->set_source_id(local_id_);
   message->set_destination_id(remote_id_);
   return channel_->WriteMessage(message.Pass());
 }
 
 }  // namespace system
 }  // namespace mojo