message_pump_libevent refactor:

base/message_pump_libevent.cc

 // Copyright (c) 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_pump_libevent.h"
 
 #include <fcntl.h>
 
 #include "base/logging.h"
 #include "base/scoped_nsautorelease_pool.h"
 #include "base/time.h"
 #include "third_party/libevent/event.h"
 
 namespace base {
 
 // Return 0 on success
 // Too small a function to bother putting in a library?
-static int SetNonBlocking(int fd)
-{
-    int flags = fcntl(fd, F_GETFL, 0);
-    if (-1 == flags)
-      flags = 0;
-    return fcntl(fd, F_SETFL, flags | O_NONBLOCK);
+static int SetNonBlocking(int fd) {
+  int flags = fcntl(fd, F_GETFL, 0);
+  if (flags == -1)
+    flags = 0;
+  return fcntl(fd, F_SETFL, flags | O_NONBLOCK);
+}
+
+MessagePumpLibevent::FileDescriptorWatcher::FileDescriptorWatcher()
+    : is_persistent_(false),
+      event_(NULL) {
+}
+
+MessagePumpLibevent::FileDescriptorWatcher::~FileDescriptorWatcher() {
+  if (event_.get()) {
+    StopWatchingFileDescriptor();
+  }
+}
+
+void MessagePumpLibevent::FileDescriptorWatcher::Init(event *e,
+                                                      bool is_persistent) {
+  DCHECK(e);
+  DCHECK(event_.get() == NULL);
+
+  is_persistent_ = is_persistent;
+  event_.reset(e);
+}
+
+event *MessagePumpLibevent::FileDescriptorWatcher::ReleaseEvent() {
+  return event_.release();
+}
+
+bool MessagePumpLibevent::FileDescriptorWatcher::StopWatchingFileDescriptor() {
+  if (event_.get() == NULL) {
+    return true;
+  }
+
+  // event_del() is a no-op of the event isn't active.
+  return (event_del(event_.get()) == 0);
 }
 
 // Called if a byte is received on the wakeup pipe.
 void MessagePumpLibevent::OnWakeup(int socket, short flags, void* context) {
-
   base::MessagePumpLibevent* that =
               static_cast<base::MessagePumpLibevent*>(context);
   DCHECK(that->wakeup_pipe_out_ == socket);
 
   // Remove and discard the wakeup byte.
   char buf;
   int nread = read(socket, &buf, 1);
   DCHECK(nread == 1);
   // Tell libevent to break out of inner loop.
   event_base_loopbreak(that->event_base_);
@@ skipping 15 matching lines @@
     return false;
   if (SetNonBlocking(fds[0]))
     return false;
   if (SetNonBlocking(fds[1]))
     return false;
   wakeup_pipe_out_ = fds[0];
   wakeup_pipe_in_ = fds[1];
 
   wakeup_event_ = new event;
   event_set(wakeup_event_, wakeup_pipe_out_, EV_READ | EV_PERSIST,
-            OnWakeup, this);
+            OnWakeup, this);
   event_base_set(event_base_, wakeup_event_);
 
   if (event_add(wakeup_event_, 0))
     return false;
   return true;
 }
 
 MessagePumpLibevent::~MessagePumpLibevent() {
   DCHECK(wakeup_event_);
   DCHECK(event_base_);
   event_del(wakeup_event_);
   delete wakeup_event_;
   event_base_free(event_base_);
 }
 
-void MessagePumpLibevent::WatchSocket(int socket, short interest_mask,
-                                      event* e, Watcher* watcher) {
+bool MessagePumpLibevent::WatchFileDescriptor(int fd,
+                                              bool persistent,
+                                              Mode mode,
+                                              FileDescriptorWatcher *controller,
+                                              Watcher *delegate) {
+  DCHECK(fd > 0);
+  DCHECK(controller);
+  DCHECK(delegate);
+  DCHECK(mode == WATCH_READ || mode == WATCH_WRITE || mode == WATCH_READ_WRITE);
 
-  // Set current interest mask and message pump for this event
-  event_set(e, socket, interest_mask, OnReadinessNotification, watcher);
+  int event_mask = persistent ? EV_PERSIST : 0;
+  if ((mode & WATCH_READ) != 0) {
+    event_mask |= EV_READ;
+  }
+  if ((mode & WATCH_WRITE) != 0) {
+    event_mask |= EV_WRITE;
+  }
+
+  // |should_delete_event| is true if we're modifying an event that's currently
+  // active in |controller|.
+  // If we're modifying an existing event and there's an error then we need to
+  // tell libevent to clean it up via event_delete() before returning.
+  bool should_delete_event = true;
+  scoped_ptr<event> evt(controller->ReleaseEvent());
+  if (evt.get() == NULL) {
+    should_delete_event = false;
+    // Ownership is transferred to the controller.
+    evt.reset(new event);
+  }
+
+  // Set current interest mask and message pump for this event.
+  event_set(evt.get(), fd, event_mask, OnLibeventNotification,
+            delegate);
 
   // Tell libevent which message pump this socket will belong to when we add it.
-  event_base_set(event_base_, e);
+  if (event_base_set(event_base_, evt.get()) != 0) {
+    if (should_delete_event) {
+      event_del(evt.get());
+    }
+    return false;
+  }
 
   // Add this socket to the list of monitored sockets.
-  if (event_add(e, NULL))
-    NOTREACHED();
+  if (event_add(evt.get(), NULL) != 0) {
+    if (should_delete_event) {
+      event_del(evt.get());
+    }
+    return false;
+  }
+
+  // Transfer ownership of e to controller.
+  controller->Init(evt.release(), persistent);
+  return true;
 }
 
-void MessagePumpLibevent::WatchFileHandle(int fd, short interest_mask,
-                                          event* e, FileWatcher* watcher) {
-  // Set current interest mask and message pump for this event
-  if ((interest_mask & EV_READ) != 0) {
-    event_set(e, fd, interest_mask, OnFileReadReadinessNotification, watcher);
-  } else {
-    event_set(e, fd, interest_mask, OnFileWriteReadinessNotification, watcher);
+
+void MessagePumpLibevent::OnLibeventNotification(int fd, short flags,
+                                                 void* context) {
+  Watcher* watcher = static_cast<Watcher*>(context);
+
+  if (flags & EV_WRITE) {
+    watcher->OnFileCanWriteWithoutBlocking(fd);
   }
-
-  // Tell libevent which message pump this fd will belong to when we add it.
-  event_base_set(event_base_, e);
-
-  // Add this fd to the list of monitored sockets.
-  if (event_add(e, NULL))
-    NOTREACHED();
-}
-
-void MessagePumpLibevent::UnwatchSocket(event* e) {
-  // Remove this socket from the list of monitored sockets.
-  if (event_del(e))
-    NOTREACHED();
-}
-
-void MessagePumpLibevent::UnwatchFileHandle(event* e) {
-  // Remove this fd from the list of monitored fds.
-  if (event_del(e))
-    NOTREACHED();
-}
-
-void MessagePumpLibevent::OnReadinessNotification(int socket, short flags,
-                                                  void* context) {
-  // The given socket is ready for I/O.
-  // Tell the owner what kind of I/O the socket is ready for.
-  Watcher* watcher = static_cast<Watcher*>(context);
-  watcher->OnSocketReady(flags);
-}
-
-void MessagePumpLibevent::OnFileReadReadinessNotification(int fd, short flags,
-                                                          void* context) {
-  FileWatcher* watcher = static_cast<FileWatcher*>(context);
-  watcher->OnFileReadReady(fd);
-}
-
-void MessagePumpLibevent::OnFileWriteReadinessNotification(int fd, short flags,
-                                                           void* context) {
-  FileWatcher* watcher = static_cast<FileWatcher*>(context);
-  watcher->OnFileWriteReady(fd);
+  if (flags & EV_READ) {
+    watcher->OnFileCanReadWithoutBlocking(fd);
+  }
 }
 
 // Reentrant!
 void MessagePumpLibevent::Run(Delegate* delegate) {
   DCHECK(keep_running_) << "Quit must have been called outside of Run!";
 
   bool old_in_run = in_run_;
   in_run_ = true;
 
   for (;;) {
@@ skipping 56 matching lines @@
 }
 
 void MessagePumpLibevent::ScheduleDelayedWork(const Time& delayed_work_time) {
   // We know that we can't be blocked on Wait right now since this method can
   // only be called on the same thread as Run, so we only need to update our
   // record of how long to sleep when we do sleep.
   delayed_work_time_ = delayed_work_time;
 }
 
 }  // namespace base