Add a mechanism to extend a libevent based message pump.

message_loop/message_pump_libevent.cc

 // Copyright (c) 2012 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/message_pump_libevent.h"
 
 #include <errno.h>
 #include <fcntl.h>
 #include <unistd.h>
 
@@ skipping 101 matching lines @@
   watcher_->OnFileCanWriteWithoutBlocking(fd);
   pump->DidProcessIOEvent();
 }
 
 MessagePumpLibevent::MessagePumpLibevent()
     : keep_running_(true),
       in_run_(false),
       processed_io_events_(false),
       event_base_(event_base_new()),
       wakeup_pipe_in_(-1),
-      wakeup_pipe_out_(-1) {
+      wakeup_pipe_out_(-1),
+      event_source_(nullptr) {
   if (!Init())
      NOTREACHED();
 }
 
 MessagePumpLibevent::~MessagePumpLibevent() {
   DCHECK(wakeup_event_);
   DCHECK(event_base_);
   event_del(wakeup_event_);
   delete wakeup_event_;
   if (wakeup_pipe_in_ >= 0) {
@@ skipping 74 matching lines @@
 }
 
 void MessagePumpLibevent::AddIOObserver(IOObserver *obs) {
   io_observers_.AddObserver(obs);
 }
 
 void MessagePumpLibevent::RemoveIOObserver(IOObserver *obs) {
   io_observers_.RemoveObserver(obs);
 }
 
+void MessagePumpLibevent::SetEventSource(EventSource* event_source) {
+  DCHECK(event_source);
+  DCHECK(!event_source_ || event_source_ == event_source);
+  event_source_ = event_source;
+}
+
+void MessagePumpLibevent::ClearEventSource() {
+  event_source_ = nullptr;
+}
+
 // Tell libevent to break out of inner loop.
 static void timer_callback(int fd, short events, void *context)
 {
   event_base_loopbreak((struct event_base *)context);
 }
 
 // Reentrant!
 void MessagePumpLibevent::Run(Delegate* delegate) {
   AutoReset<bool> auto_reset_keep_running(&keep_running_, true);
   AutoReset<bool> auto_reset_in_run(&in_run_, true);
 
   // event_base_loopexit() + EVLOOP_ONCE is leaky, see http://crbug.com/25641.
   // Instead, make our own timer and reuse it on each call to event_base_loop().
   scoped_ptr<event> timer_event(new event);
 
   for (;;) {
 #if defined(OS_MACOSX)
     mac::ScopedNSAutoreleasePool autorelease_pool;
 #endif
 
     bool did_work = delegate->DoWork();
     if (!keep_running_)
       break;
 
+    did_work |= PollEventSource();
+    if (!keep_running_)
+      break;
+
     event_base_loop(event_base_, EVLOOP_NONBLOCK);
     did_work |= processed_io_events_;
     processed_io_events_ = false;
     if (!keep_running_)
       break;
 
     did_work |= delegate->DoDelayedWork(&delayed_work_time_);
     if (!keep_running_)
       break;
 
@@ skipping 58 matching lines @@
 }
 
 void MessagePumpLibevent::WillProcessIOEvent() {
   FOR_EACH_OBSERVER(IOObserver, io_observers_, WillProcessIOEvent());
 }
 
 void MessagePumpLibevent::DidProcessIOEvent() {
   FOR_EACH_OBSERVER(IOObserver, io_observers_, DidProcessIOEvent());
 }
 
+bool MessagePumpLibevent::PollEventSource() {
+  return event_source_ && event_source_->Poll();
+}
+
 bool MessagePumpLibevent::Init() {
   int fds[2];
   if (pipe(fds)) {
     DLOG(ERROR) << "pipe() failed, errno: " << errno;
     return false;
   }
   if (SetNonBlocking(fds[0])) {
     DLOG(ERROR) << "SetNonBlocking for pipe fd[0] failed, errno: " << errno;
     return false;
   }
@@ skipping 53 matching lines @@
   // Remove and discard the wakeup byte.
   char buf;
   int nread = HANDLE_EINTR(read(socket, &buf, 1));
   DCHECK_EQ(nread, 1);
   that->processed_io_events_ = true;
   // Tell libevent to break out of inner loop.
   event_base_loopbreak(that->event_base_);
 }
 
 }  // namespace base