Add a base class for objects that want to filter messages on the UI thread. ...

ipc/ipc_channel_proxy.cc

 // Copyright (c) 2006-2008 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 "base/message_loop.h"
 #include "base/ref_counted.h"
 #include "base/scoped_ptr.h"
 #include "base/thread.h"
 #include "ipc/ipc_channel_proxy.h"
 #include "ipc/ipc_logging.h"
@@ skipping 43 matching lines @@
   return false;
 }
 
 void ChannelProxy::MessageFilter::OnDestruct() const {
   delete this;
 }
 
 //------------------------------------------------------------------------------
 
 ChannelProxy::Context::Context(Channel::Listener* listener,
-                               MessageFilter* filter,
                                MessageLoop* ipc_message_loop)
     : listener_message_loop_(MessageLoop::current()),
       listener_(listener),
       ipc_message_loop_(ipc_message_loop),
       channel_(NULL),
       peer_pid_(0),
       channel_connected_called_(false) {
-  if (filter)
-    filters_.push_back(make_scoped_refptr(filter));
 }
 
 void ChannelProxy::Context::CreateChannel(const std::string& id,
                                           const Channel::Mode& mode) {
   DCHECK(channel_ == NULL);
   channel_id_ = id;
   channel_ = new Channel(id, mode, this);
 }
 
 bool ChannelProxy::Context::TryFilters(const Message& message) {
@@ skipping 27 matching lines @@
   // NOTE: This code relies on the listener's message loop not going away while
   // this thread is active.  That should be a reasonable assumption, but it
   // feels risky.  We may want to invent some more indirect way of referring to
   // a MessageLoop if this becomes a problem.
   listener_message_loop_->PostTask(FROM_HERE, NewRunnableMethod(
       this, &Context::OnDispatchMessage, message));
 }
 
 // Called on the IPC::Channel thread
 void ChannelProxy::Context::OnChannelConnected(int32 peer_pid) {
+  // Add any pending filters.  This avoids a race condition where someone
+  // creates a ChannelProxy, calls AddFilter, and then right after starts the
+  // peer process.  The IO thread could receive a message before the task to add
+  // the filter is run on the IO thread.
+  OnAddFilter();
+
   peer_pid_ = peer_pid;
   for (size_t i = 0; i < filters_.size(); ++i)
     filters_[i]->OnChannelConnected(peer_pid);
 
   // See above comment about using listener_message_loop_ here.
   listener_message_loop_->PostTask(FROM_HERE, NewRunnableMethod(
       this, &Context::OnDispatchConnected));
 }
 
 // Called on the IPC::Channel thread
@@ skipping 51 matching lines @@
   if (!channel_) {
     delete message;
     OnChannelClosed();
     return;
   }
   if (!channel_->Send(message))
     OnChannelError();
 }
 
 // Called on the IPC::Channel thread
-void ChannelProxy::Context::OnAddFilter(MessageFilter* filter) {
-  filters_.push_back(make_scoped_refptr(filter));
+void ChannelProxy::Context::OnAddFilter() {
+  std::vector<scoped_refptr<MessageFilter> > filters;
+  {
+    AutoLock auto_lock(pending_filters_lock_);
+    filters.swap(pending_filters_);
+  }
 
-  // If the channel has already been created, then we need to send this message
-  // so that the filter gets access to the Channel.
-  if (channel_)
-    filter->OnFilterAdded(channel_);
+  for (size_t i = 0; i < filters.size(); ++i) {
+    filters_.push_back(filters[i]);
+
+    // If the channel has already been created, then we need to send this
+    // message so that the filter gets access to the Channel.
+    if (channel_)
+      filters[i]->OnFilterAdded(channel_);
+    // Ditto for the peer process id.
+    if (peer_pid_)
+      filters[i]->OnChannelConnected(peer_pid_);
+  }
 }
 
 // Called on the IPC::Channel thread
 void ChannelProxy::Context::OnRemoveFilter(MessageFilter* filter) {
   for (size_t i = 0; i < filters_.size(); ++i) {
     if (filters_[i].get() == filter) {
       filter->OnFilterRemoved();
       filters_.erase(filters_.begin() + i);
       return;
     }
   }
 
   NOTREACHED() << "filter to be removed not found";
 }
 
 // Called on the listener's thread
+void ChannelProxy::Context::AddFilter(MessageFilter* filter) {
+  AutoLock auto_lock(pending_filters_lock_);
+  pending_filters_.push_back(make_scoped_refptr(filter));
+  ipc_message_loop_->PostTask(
+      FROM_HERE,
+      NewRunnableMethod(this, &Context::OnAddFilter));
+}
+
+// Called on the listener's thread
 void ChannelProxy::Context::OnDispatchMessage(const Message& message) {
   if (!listener_)
     return;
 
   OnDispatchConnected();
 
 #ifdef IPC_MESSAGE_LOG_ENABLED
   Logging* logger = Logging::current();
   if (message.type() == IPC_LOGGING_ID) {
     logger->OnReceivedLoggingMessage(message);
@@ skipping 23 matching lines @@
 }
 
 // Called on the listener's thread
 void ChannelProxy::Context::OnDispatchError() {
   if (listener_)
     listener_->OnChannelError();
 }
 
 //-----------------------------------------------------------------------------
 
-ChannelProxy::ChannelProxy(const std::string& channel_id, Channel::Mode mode,
-                           Channel::Listener* listener, MessageFilter* filter,
+ChannelProxy::ChannelProxy(const std::string& channel_id,
+                           Channel::Mode mode,
+                           Channel::Listener* listener,
                            MessageLoop* ipc_thread)
-    : context_(new Context(listener, filter, ipc_thread)) {
+    : context_(new Context(listener, ipc_thread)) {
   Init(channel_id, mode, ipc_thread, true);
 }
 
-ChannelProxy::ChannelProxy(const std::string& channel_id, Channel::Mode mode,
-                           MessageLoop* ipc_thread, Context* context,
+ChannelProxy::ChannelProxy(const std::string& channel_id,
+                           Channel::Mode mode,
+                           MessageLoop* ipc_thread,
+                           Context* context,
                            bool create_pipe_now)
     : context_(context) {
   Init(channel_id, mode, ipc_thread, create_pipe_now);
 }
 
 ChannelProxy::~ChannelProxy() {
   Close();
 }
 
 void ChannelProxy::Init(const std::string& channel_id, Channel::Mode mode,
                         MessageLoop* ipc_thread_loop, bool create_pipe_now) {
   if (create_pipe_now) {
     // Create the channel immediately.  This effectively sets up the
     // low-level pipe so that the client can connect.  Without creating
     // the pipe immediately, it is possible for a listener to attempt
     // to connect and get an error since the pipe doesn't exist yet.
     context_->CreateChannel(channel_id, mode);
   } else {
     context_->ipc_message_loop()->PostTask(FROM_HERE, NewRunnableMethod(
         context_.get(), &Context::CreateChannel, channel_id, mode));
   }
 
   // complete initialization on the background thread
   context_->ipc_message_loop()->PostTask(FROM_HERE, NewRunnableMethod(
       context_.get(), &Context::OnChannelOpened));

[Matt Perry, 2010/12/02 19:00:39]

Couldn't this cause OnChannelOpened to run, then OnChannelConnected, all before
any subsequent calls to AddFiler? I think you still need to call this from a
PostTask to the listener thread.

[jam, 2010/12/02 19:08:53]

That's fine, since OnAddFilter will call OnChannelConnected if a filter is added
after the channel is connected.
this can result in the channel not working (i had tried this before).  the
problem is that if you create a channel and call send() on it, you don't expect
that you have to run the current message loop in between.

 }
 
 void ChannelProxy::Close() {
   // Clear the backpointer to the listener so that any pending calls to
   // Context::OnDispatchMessage or OnDispatchError will be ignored.  It is
   // possible that the channel could be closed while it is receiving messages!
   context_->Clear();
 
   if (context_->ipc_message_loop()) {
     context_->ipc_message_loop()->PostTask(FROM_HERE, NewRunnableMethod(
         context_.get(), &Context::OnChannelClosed));
   }
 }
 
 bool ChannelProxy::Send(Message* message) {
 #ifdef IPC_MESSAGE_LOG_ENABLED
   Logging::current()->OnSendMessage(message, context_->channel_id());
 #endif
 
   context_->ipc_message_loop()->PostTask(FROM_HERE,
                                          new SendTask(context_.get(), message));
   return true;
 }
 
 void ChannelProxy::AddFilter(MessageFilter* filter) {
-  context_->ipc_message_loop()->PostTask(
-      FROM_HERE,
-      NewRunnableMethod(
-          context_.get(),
-          &Context::OnAddFilter,
-          make_scoped_refptr(filter)));
+  context_->AddFilter(filter);
 }
 
 void ChannelProxy::RemoveFilter(MessageFilter* filter) {
   context_->ipc_message_loop()->PostTask(
       FROM_HERE, NewRunnableMethod(
           context_.get(),
           &Context::OnRemoveFilter,
           make_scoped_refptr(filter)));
 }
 
 void ChannelProxy::ClearIPCMessageLoop() {
   context()->ClearIPCMessageLoop();
 }
 
 #if defined(OS_POSIX) && !defined(OS_NACL)
 // See the TODO regarding lazy initialization of the channel in
 // ChannelProxy::Init().
 // We assume that IPC::Channel::GetClientFileDescriptorMapping() is thread-safe.
 int ChannelProxy::GetClientFileDescriptor() const {
   Channel *channel = context_.get()->channel_;
   DCHECK(channel); // Channel must have been created first.
   return channel->GetClientFileDescriptor();
 }
 #endif
 
 //-----------------------------------------------------------------------------
 
 }  // namespace IPC