message_pump_libevent refactor:

* Unify WatchSocket & WatchFileHandle.
* Better encapsulate libevent.
* Fix a bug with blocking writes in ipc_posix.cc

Review URL: http://codereview.chromium.org/13757

git-svn-id: svn://svn.chromium.org/chrome/trunk/src@7010 0039d316-1c4b-4281-b951-d872f2087c98

1 files changed, 39 insertions, 22 deletions

diff --git a/net/base/tcp_client_socket_libevent.cc b/net/base/tcp_client_socket_libevent.cc
index 79909c20..7a9084b 100644
--- a/net/base/tcp_client_socket_libevent.cc
+++ b/net/base/tcp_client_socket_libevent.cc
@@ -68,7 +68,6 @@ TCPClientSocket::TCPClientSocket(const AddressList& addresses)
     addresses_(addresses),
     current_ai_(addresses_.head()),
     waiting_connect_(false),
-    event_(new event),
     write_callback_(NULL),
     callback_(NULL) {
 }
@@ -106,12 +105,17 @@ int TCPClientSocket::Connect(CompletionCallback* callback) {
     return MapPosixError(errno);
   }
 
-  // Initialize event_ and link it to our MessagePump.
+  // Initialize socket_watcher_ and link it to our MessagePump.
   // POLLOUT is set if the connection is established.
-  // POLLIN is set if the connection fails,
-  // so select for both read and write.
-  MessageLoopForIO::current()->WatchSocket(
-     socket_, EV_READ|EV_WRITE|EV_PERSIST, event_.get(), this);
+  // POLLIN is set if the connection fails.
+  if (!MessageLoopForIO::current()->WatchFileDescriptor(
+          socket_, true, MessageLoopForIO::WATCH_WRITE, &socket_watcher_,
+          this)) {
+    DLOG(INFO) << "WatchFileDescriptor failed: " << errno;
+    close(socket_);
+    socket_ = kInvalidSocket;
+    return MapPosixError(errno);
+  }
 
   waiting_connect_ = true;
   callback_ = callback;
@@ -127,7 +131,7 @@ void TCPClientSocket::Disconnect() {
   if (socket_ == kInvalidSocket)
     return;
 
-  MessageLoopForIO::current()->UnwatchSocket(event_.get());
+  socket_watcher_.StopWatchingFileDescriptor();
   close(socket_);
   socket_ = kInvalidSocket;
   waiting_connect_ = false;
@@ -170,8 +174,12 @@ int TCPClientSocket::Read(char* buf,
     return MapPosixError(errno);
   }
 
-  MessageLoopForIO::current()->WatchSocket(
-     socket_, EV_READ|EV_PERSIST, event_.get(), this);
+  if (!MessageLoopForIO::current()->WatchFileDescriptor(
+          socket_, true, MessageLoopForIO::WATCH_READ, &socket_watcher_, this))
+      {
+    DLOG(INFO) << "WatchFileDescriptor failed on read, errno " << errno;
+    return MapPosixError(errno);
+  }
 
   buf_ = buf;
   buf_len_ = buf_len;
@@ -196,8 +204,13 @@ int TCPClientSocket::Write(const char* buf,
   if (errno != EAGAIN && errno != EWOULDBLOCK)
     return MapPosixError(errno);
 
-  MessageLoopForIO::current()->WatchSocket(
-     socket_, EV_WRITE|EV_PERSIST, event_.get(), this);
+  if (!MessageLoopForIO::current()->WatchFileDescriptor(
+          socket_, true, MessageLoopForIO::WATCH_WRITE, &socket_watcher_, this))
+      {
+    DLOG(INFO) << "WatchFileDescriptor failed on write, errno " << errno;
+    return MapPosixError(errno);
+  }
+
 
   write_buf_ = buf;
   write_buf_len_ = buf_len;
@@ -263,12 +276,13 @@ void TCPClientSocket::DidCompleteConnect() {
     result = Connect(callback_);
   } else {
     result = MapPosixError(error_code);
-    MessageLoopForIO::current()->UnwatchSocket(event_.get());
+    socket_watcher_.StopWatchingFileDescriptor();
     waiting_connect_ = false;
   }
 
-  if (result != ERR_IO_PENDING)
+  if (result != ERR_IO_PENDING) {
     DoCallback(result);
+  }
 }
 
 void TCPClientSocket::DidCompleteRead() {
@@ -285,7 +299,7 @@ void TCPClientSocket::DidCompleteRead() {
   if (result != ERR_IO_PENDING) {
     buf_ = NULL;
     buf_len_ = 0;
-    MessageLoopForIO::current()->UnwatchSocket(event_.get());
+    socket_watcher_.StopWatchingFileDescriptor();
     DoCallback(result);
   }
 }
@@ -304,21 +318,24 @@ void TCPClientSocket::DidCompleteWrite() {
   if (result != ERR_IO_PENDING) {
     write_buf_ = NULL;
     write_buf_len_ = 0;
-    MessageLoopForIO::current()->UnwatchSocket(event_.get());
+    socket_watcher_.StopWatchingFileDescriptor();
     DoWriteCallback(result);
   }
 }
 
-void TCPClientSocket::OnSocketReady(short flags) {
-  // the only used bits of flags are EV_READ and EV_WRITE
+void TCPClientSocket::OnFileCanReadWithoutBlocking(int fd) {
+  // When a socket connects it signals both Read and Write, we handle
+  // DidCompleteConnect() in the write handler.
+  if (!waiting_connect_ && callback_) {
+    DidCompleteRead();
+  }
+}
 
+void TCPClientSocket::OnFileCanWriteWithoutBlocking(int fd) {
   if (waiting_connect_) {
     DidCompleteConnect();
-  } else {
-    if ((flags & EV_WRITE) && write_callback_)
-      DidCompleteWrite();
-    if ((flags & EV_READ) && callback_)
-      DidCompleteRead();
+  } else if (write_callback_) {
+    DidCompleteWrite();
   }
 }