Add the ability to observe MessagePumpMojo

mojo/common/message_pump_mojo.cc

 // Copyright 2013 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/common/message_pump_mojo.h"
 
 #include <algorithm>
 #include <vector>
 
 #include "base/debug/alias.h"
@@ skipping 79 matching lines @@
   handler_data.wait_signals = wait_signals;
   handler_data.deadline = deadline;
   handler_data.id = next_handler_id_++;
   handlers_[handle] = handler_data;
 }
 
 void MessagePumpMojo::RemoveHandler(const Handle& handle) {
   handlers_.erase(handle);
 }
 
+void MessagePumpMojo::AddObserver(Observer* observer) {
+  observers_.AddObserver(observer);
+}
+
+void MessagePumpMojo::RemoveObserver(Observer* observer) {
+  observers_.RemoveObserver(observer);
+}
+
 void MessagePumpMojo::Run(Delegate* delegate) {
   RunState run_state;
   // TODO: better deal with error handling.
   CHECK(run_state.read_handle.is_valid());
   CHECK(run_state.write_handle.is_valid());
   RunState* old_state = NULL;
   {
     base::AutoLock auto_lock(run_state_lock_);
     old_state = run_state_;
     run_state_ = &run_state;
@@ skipping 58 matching lines @@
   const MojoResult result =
       WaitMany(wait_state.handles, wait_state.wait_signals, deadline);
   bool did_work = true;
   if (result == 0) {
     // Control pipe was written to.
     ReadMessageRaw(run_state.read_handle.get(), NULL, NULL, NULL, NULL,
                    MOJO_READ_MESSAGE_FLAG_MAY_DISCARD);
   } else if (result > 0) {
     const size_t index = static_cast<size_t>(result);
     DCHECK(handlers_.find(wait_state.handles[index]) != handlers_.end());
+    WillSignalHandler();
     handlers_[wait_state.handles[index]].handler->OnHandleReady(
         wait_state.handles[index]);
+    DidSignalHandler();
   } else {
     switch (result) {
       case MOJO_RESULT_CANCELLED:
       case MOJO_RESULT_FAILED_PRECONDITION:
         RemoveFirstInvalidHandle(wait_state);
         break;
       case MOJO_RESULT_DEADLINE_EXCEEDED:
         did_work = false;
         break;
       default:
         base::debug::Alias(&result);
         // Unexpected result is likely fatal, crash so we can determine cause.
         CHECK(false);
     }
   }
 
   // Notify and remove any handlers whose time has expired. Make a copy in case
   // someone tries to add/remove new handlers from notification.
   const HandleToHandler cloned_handlers(handlers_);
   const base::TimeTicks now(internal::NowTicks());
   for (HandleToHandler::const_iterator i = cloned_handlers.begin();
        i != cloned_handlers.end(); ++i) {
     // Since we're iterating over a clone of the handlers, verify the handler is
     // still valid before notifying.
     if (!i->second.deadline.is_null() && i->second.deadline < now &&
         handlers_.find(i->first) != handlers_.end() &&
         handlers_[i->first].id == i->second.id) {
+      WillSignalHandler();
       i->second.handler->OnHandleError(i->first, MOJO_RESULT_DEADLINE_EXCEEDED);
+      DidSignalHandler();
       handlers_.erase(i->first);
       did_work = true;
     }
   }
   return did_work;
 }
 
 void MessagePumpMojo::RemoveFirstInvalidHandle(const WaitState& wait_state) {
   // TODO(sky): deal with control pipe going bad.
   for (size_t i = 0; i < wait_state.handles.size(); ++i) {
     const MojoResult result =
         Wait(wait_state.handles[i], wait_state.wait_signals[i], 0);
     if (result == MOJO_RESULT_INVALID_ARGUMENT) {
       // We should never have an invalid argument. If we do it indicates
       // RemoveHandler() was not invoked and is likely to cause problems else
       // where in the stack if we ignore it.
       CHECK(false);
     } else if (result == MOJO_RESULT_FAILED_PRECONDITION ||
                result == MOJO_RESULT_CANCELLED) {
       CHECK_NE(i, 0u);  // Indicates the control pipe went bad.
 
       // Remove the handle first, this way if OnHandleError() tries to remove
       // the handle our iterator isn't invalidated.
       CHECK(handlers_.find(wait_state.handles[i]) != handlers_.end());
       MessagePumpMojoHandler* handler =
           handlers_[wait_state.handles[i]].handler;
       handlers_.erase(wait_state.handles[i]);
+      WillSignalHandler();
       handler->OnHandleError(wait_state.handles[i], result);
+      DidSignalHandler();
       return;
     }
   }
 }
 
 void MessagePumpMojo::SignalControlPipe(const RunState& run_state) {
   const MojoResult result =
       WriteMessageRaw(run_state.write_handle.get(), NULL, 0, NULL, 0,
                       MOJO_WRITE_MESSAGE_FLAG_NONE);
   // If we can't write we likely won't wake up the thread and there is a strong
@@ skipping 21 matching lines @@
   MojoDeadline deadline = TimeTicksToMojoDeadline(run_state.delayed_work_time,
                                                   now);
   for (HandleToHandler::const_iterator i = handlers_.begin();
        i != handlers_.end(); ++i) {
     deadline = std::min(
         TimeTicksToMojoDeadline(i->second.deadline, now), deadline);
   }
   return deadline;
 }
 
+void MessagePumpMojo::WillSignalHandler() {
+  FOR_EACH_OBSERVER(Observer, observers_, WillSignalHandler());
+}
+
+void MessagePumpMojo::DidSignalHandler() {
+  FOR_EACH_OBSERVER(Observer, observers_, DidSignalHandler());
+}
+
 }  // namespace common
 }  // namespace mojo