ipc/ipc_channel_proxy.cc

   1 // Copyright (c) 2006-2008 The Chromium Authors. All rights reserved.
   2 // Use of this source code is governed by a BSD-style license that can be
   3 // found in the LICENSE file.
   4
   5 #include "base/message_loop.h"
   6 #include "base/thread.h"
   7 #include "ipc/ipc_channel_proxy.h"
   8 #include "ipc/ipc_logging.h"
   9 #include "ipc/ipc_message_utils.h"
  10
  11 namespace IPC {
  12
  13 //------------------------------------------------------------------------------
  14
  15 // This task ensures the message is deleted if the task is deleted without
  16 // having been run.
  17 class SendTask : public Task {
  18  public:
  19   SendTask(ChannelProxy::Context* context, Message* message)
  20       : context_(context),
  21         message_(message) {
  22   }
  23
  24   virtual void Run() {
  25     context_->OnSendMessage(message_.release());
  26   }
  27
  28  private:
  29   scoped_refptr<ChannelProxy::Context> context_;
  30   scoped_ptr<Message> message_;
  31
  32   DISALLOW_COPY_AND_ASSIGN(SendTask);
  33 };
  34
  35 //------------------------------------------------------------------------------
  36
  37 ChannelProxy::Context::Context(Channel::Listener* listener,
  38                                MessageFilter* filter,
  39                                MessageLoop* ipc_message_loop)
  40     : listener_message_loop_(MessageLoop::current()),
  41       listener_(listener),
  42       ipc_message_loop_(ipc_message_loop),
  43       channel_(NULL),
  44       peer_pid_(0),
  45       channel_connected_called_(false) {
  46   if (filter)
  47     filters_.push_back(filter);
  48 }
  49
  50 void ChannelProxy::Context::CreateChannel(const std::string& id,
  51                                           const Channel::Mode& mode) {
  52   DCHECK(channel_ == NULL);
  53   channel_id_ = id;
  54   channel_ = new Channel(id, mode, this);
  55 }
  56
  57 bool ChannelProxy::Context::TryFilters(const Message& message) {
  58 #ifdef IPC_MESSAGE_LOG_ENABLED
  59   Logging* logger = Logging::current();
  60   if (logger->Enabled())
  61     logger->OnPreDispatchMessage(message);
  62 #endif
  63
  64   for (size_t i = 0; i < filters_.size(); ++i) {
  65     if (filters_[i]->OnMessageReceived(message)) {
  66 #ifdef IPC_MESSAGE_LOG_ENABLED
  67       if (logger->Enabled())
  68         logger->OnPostDispatchMessage(message, channel_id_);
  69 #endif
  70       return true;
  71     }
  72   }
  73   return false;
  74 }
  75
  76 // Called on the IPC::Channel thread
  77 void ChannelProxy::Context::OnMessageReceived(const Message& message) {
  78   // First give a chance to the filters to process this message.
  79   if (!TryFilters(message))
  80     OnMessageReceivedNoFilter(message);
  81 }
  82
  83 // Called on the IPC::Channel thread
  84 void ChannelProxy::Context::OnMessageReceivedNoFilter(const Message& message) {
  85   // NOTE: This code relies on the listener's message loop not going away while
  86   // this thread is active.  That should be a reasonable assumption, but it
  87   // feels risky.  We may want to invent some more indirect way of referring to
  88   // a MessageLoop if this becomes a problem.
  89   listener_message_loop_->PostTask(FROM_HERE, NewRunnableMethod(
  90       this, &Context::OnDispatchMessage, message));
  91 }
  92
  93 // Called on the IPC::Channel thread
  94 void ChannelProxy::Context::OnChannelConnected(int32 peer_pid) {
  95   peer_pid_ = peer_pid;
  96   for (size_t i = 0; i < filters_.size(); ++i)
  97     filters_[i]->OnChannelConnected(peer_pid);
  98
  99   // See above comment about using listener_message_loop_ here.
 100   listener_message_loop_->PostTask(FROM_HERE, NewRunnableMethod(
 101       this, &Context::OnDispatchConnected));
 102 }
 103
 104 // Called on the IPC::Channel thread
 105 void ChannelProxy::Context::OnChannelError() {
 106   for (size_t i = 0; i < filters_.size(); ++i)
 107     filters_[i]->OnChannelError();
 108
 109   // See above comment about using listener_message_loop_ here.
 110   listener_message_loop_->PostTask(FROM_HERE, NewRunnableMethod(
 111       this, &Context::OnDispatchError));
 112 }
 113
 114 // Called on the IPC::Channel thread
 115 void ChannelProxy::Context::OnChannelOpened() {
 116   DCHECK(channel_ != NULL);
 117
 118   // Assume a reference to ourselves on behalf of this thread.  This reference
 119   // will be released when we are closed.
 120   AddRef();
 121
 122   if (!channel_->Connect()) {
 123     OnChannelError();
 124     return;
 125   }
 126
 127   for (size_t i = 0; i < filters_.size(); ++i)
 128     filters_[i]->OnFilterAdded(channel_);
 129 }
 130
 131 // Called on the IPC::Channel thread
 132 void ChannelProxy::Context::OnChannelClosed() {
 133   // It's okay for IPC::ChannelProxy::Close to be called more than once, which
 134   // would result in this branch being taken.
 135   if (!channel_)
 136     return;
 137
 138   for (size_t i = 0; i < filters_.size(); ++i) {
 139     filters_[i]->OnChannelClosing();
 140     filters_[i]->OnFilterRemoved();
 141   }
 142
 143   // We don't need the filters anymore.
 144   filters_.clear();
 145
 146   delete channel_;
 147   channel_ = NULL;
 148
 149   // Balance with the reference taken during startup.  This may result in
 150   // self-destruction.
 151   Release();
 152 }
 153
 154 // Called on the IPC::Channel thread
 155 void ChannelProxy::Context::OnSendMessage(Message* message) {
 156   if (!channel_->Send(message))
 157     OnChannelError();
 158 }
 159
 160 // Called on the IPC::Channel thread
 161 void ChannelProxy::Context::OnAddFilter(MessageFilter* filter) {
 162   filters_.push_back(filter);
 163
 164   // If the channel has already been created, then we need to send this message
 165   // so that the filter gets access to the Channel.
 166   if (channel_)
 167     filter->OnFilterAdded(channel_);
 168
 169   // Balances the AddRef in ChannelProxy::AddFilter.
 170   filter->Release();
 171 }
 172
 173 // Called on the IPC::Channel thread
 174 void ChannelProxy::Context::OnRemoveFilter(MessageFilter* filter) {
 175   for (size_t i = 0; i < filters_.size(); ++i) {
 176     if (filters_[i].get() == filter) {
 177       filter->OnFilterRemoved();
 178       filters_.erase(filters_.begin() + i);
 179       return;
 180     }
 181   }
 182
 183   NOTREACHED() << "filter to be removed not found";
 184 }
 185
 186 // Called on the listener's thread
 187 void ChannelProxy::Context::OnDispatchMessage(const Message& message) {
 188   if (!listener_)
 189     return;
 190
 191   OnDispatchConnected();
 192
 193 #ifdef IPC_MESSAGE_LOG_ENABLED
 194   Logging* logger = Logging::current();
 195   if (message.type() == IPC_LOGGING_ID) {
 196     logger->OnReceivedLoggingMessage(message);
 197     return;
 198   }
 199
 200   if (logger->Enabled())
 201     logger->OnPreDispatchMessage(message);
 202 #endif
 203
 204   listener_->OnMessageReceived(message);
 205
 206 #ifdef IPC_MESSAGE_LOG_ENABLED
 207   if (logger->Enabled())
 208     logger->OnPostDispatchMessage(message, channel_id_);
 209 #endif
 210 }
 211
 212 // Called on the listener's thread
 213 void ChannelProxy::Context::OnDispatchConnected() {
 214   if (channel_connected_called_)
 215     return;
 216
 217   channel_connected_called_ = true;
 218   if (listener_)
 219     listener_->OnChannelConnected(peer_pid_);
 220 }
 221
 222 // Called on the listener's thread
 223 void ChannelProxy::Context::OnDispatchError() {
 224   if (listener_)
 225     listener_->OnChannelError();
 226 }
 227
 228 //-----------------------------------------------------------------------------
 229
 230 ChannelProxy::ChannelProxy(const std::string& channel_id, Channel::Mode mode,
 231                            Channel::Listener* listener, MessageFilter* filter,
 232                            MessageLoop* ipc_thread)
 233     : context_(new Context(listener, filter, ipc_thread)) {
 234   Init(channel_id, mode, ipc_thread, true);
 235 }
 236
 237 ChannelProxy::ChannelProxy(const std::string& channel_id, Channel::Mode mode,
 238                            MessageLoop* ipc_thread, Context* context,
 239                            bool create_pipe_now)
 240     : context_(context) {
 241   Init(channel_id, mode, ipc_thread, create_pipe_now);
 242 }
 243
 244 void ChannelProxy::Init(const std::string& channel_id, Channel::Mode mode,
 245                         MessageLoop* ipc_thread_loop, bool create_pipe_now) {
 246   if (create_pipe_now) {
 247     // Create the channel immediately.  This effectively sets up the
 248     // low-level pipe so that the client can connect.  Without creating
 249     // the pipe immediately, it is possible for a listener to attempt
 250     // to connect and get an error since the pipe doesn't exist yet.
 251     context_->CreateChannel(channel_id, mode);
 252   } else {
 253     context_->ipc_message_loop()->PostTask(FROM_HERE, NewRunnableMethod(
 254         context_.get(), &Context::CreateChannel, channel_id, mode));
 255   }
 256
 257   // complete initialization on the background thread
 258   context_->ipc_message_loop()->PostTask(FROM_HERE, NewRunnableMethod(
 259       context_.get(), &Context::OnChannelOpened));
 260 }
 261
 262 void ChannelProxy::Close() {
 263   // Clear the backpointer to the listener so that any pending calls to
 264   // Context::OnDispatchMessage or OnDispatchError will be ignored.  It is
 265   // possible that the channel could be closed while it is receiving messages!
 266   context_->Clear();
 267
 268   if (context_->ipc_message_loop()) {
 269     context_->ipc_message_loop()->PostTask(FROM_HERE, NewRunnableMethod(
 270         context_.get(), &Context::OnChannelClosed));
 271   }
 272 }
 273
 274 bool ChannelProxy::Send(Message* message) {
 275 #ifdef IPC_MESSAGE_LOG_ENABLED
 276   Logging::current()->OnSendMessage(message, context_->channel_id());
 277 #endif
 278
 279   context_->ipc_message_loop()->PostTask(FROM_HERE,
 280                                          new SendTask(context_.get(), message));
 281   return true;
 282 }
 283
 284 void ChannelProxy::AddFilter(MessageFilter* filter) {
 285   // We want to addref the filter to prevent it from
 286   // being destroyed before the OnAddFilter call is invoked.
 287   filter->AddRef();
 288   context_->ipc_message_loop()->PostTask(FROM_HERE, NewRunnableMethod(
 289       context_.get(), &Context::OnAddFilter, filter));
 290 }
 291
 292 void ChannelProxy::RemoveFilter(MessageFilter* filter) {
 293   context_->ipc_message_loop()->PostTask(FROM_HERE, NewRunnableMethod(
 294       context_.get(), &Context::OnRemoveFilter, filter));
 295 }
 296
 297 void ChannelProxy::ClearIPCMessageLoop() {
 298   context()->ClearIPCMessageLoop();
 299 }
 300
 301 #if defined(OS_POSIX)
 302 // See the TODO regarding lazy initialization of the channel in
 303 // ChannelProxy::Init().
 304 // We assume that IPC::Channel::GetClientFileDescriptorMapping() is thread-safe.
 305 int ChannelProxy::GetClientFileDescriptor() const {
 306   Channel *channel = context_.get()->channel_;
 307   DCHECK(channel); // Channel must have been created first.
 308   return channel->GetClientFileDescriptor();
 309 }
 310 #endif
 311
 312 //-----------------------------------------------------------------------------
 313
 314 }  // namespace IPC