Change made by external contributor Ibrar Ahmed (ibrar.ahmad@gmail.com), reviewed by erikkay:

http://codereview.chromium.org/6577/show

ports listen_socket and telnet_server to POSIX

I had to make some changes to get this to work on Mac and to fix a few regressions it caused in Windows, so please take a fresh look at this diff Dan.

Ibrar, please take a look at the changes from your patch to mine (you may need to diff listen_socket_unittest.h with listen_socket_unittest.cc manually since I moved a bunch of code here).  Some were bugs that I should have caught in review, some were bugs that only got tickled on the Mac, others were just cleanup.  Comments welcome.
Review URL: http://codereview.chromium.org/9260

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

8 files changed, 597 insertions, 276 deletions

diff --git a/net/base/listen_socket.cc b/net/base/listen_socket.cc
index 8328389..8428e06 100644
--- a/net/base/listen_socket.cc
+++ b/net/base/listen_socket.cc
@@ -2,40 +2,72 @@
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
+#include "build/build_config.h"
+
+#if defined(OS_WIN)
 // winsock2.h must be included first in order to ensure it is included before
 // windows.h.
 #include <winsock2.h>
+#elif defined(OS_POSIX)
+#include <errno.h>
+#include <sys/socket.h>
+#include <arpa/inet.h>
+#include "base/message_loop.h"
+#include "net/base/net_errors.h"
+#include "third_party/libevent/event.h"
+#endif
 
+#include "net/base/net_util.h"
 #include "net/base/listen_socket.h"
 
-#define READ_BUF_SIZE 200
+#if defined(OS_WIN)
+#define socklen_t int
+#elif defined(OS_POSIX)
+const int INVALID_SOCKET = -1; // Used same name as in Windows to avoid #ifdef
+const int SOCKET_ERROR = -1;
+#endif
+
+const int kReadBufSize = 200;
 
 ListenSocket::ListenSocket(SOCKET s, ListenSocketDelegate *del)
+#if defined(OS_WIN)
     : socket_(s),
+#elif defined(OS_POSIX)
+    : event_(new event), 
+      socket_(s),
+#endif
       socket_delegate_(del) {
+#if defined(OS_WIN)
   socket_event_ = WSACreateEvent();
-  WSAEventSelect(socket_, socket_event_, FD_ACCEPT | FD_CLOSE | FD_READ);
-  watcher_.StartWatching(socket_event_, this);
+  // TODO(ibrar): error handling in case of socket_event_ == WSA_INVALID_EVENT
+  WatchSocket(NOT_WAITING);
+#endif
 }
 
 ListenSocket::~ListenSocket() {
+#if defined(OS_WIN)
   if (socket_event_) {
-    watcher_.StopWatching();
     WSACloseEvent(socket_event_);
+    socket_event_ = WSA_INVALID_EVENT;
   }
-  if (socket_)
-    closesocket(socket_);
+#endif
+  CloseSocket(socket_);
 }
 
 SOCKET ListenSocket::Listen(std::string ip, int port) {
   SOCKET s = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
   if (s != INVALID_SOCKET) {
     sockaddr_in addr;
+    memset(&addr, 0, sizeof(addr));
     addr.sin_family = AF_INET;
     addr.sin_addr.s_addr = inet_addr(ip.c_str());
     addr.sin_port = htons(port);
-    if (bind(s, reinterpret_cast<SOCKADDR*>(&addr), sizeof(addr))) {
+    if (bind(s, reinterpret_cast<sockaddr*>(&addr), sizeof(addr))) {
+#if defined(OS_WIN)
       closesocket(s);
+#elif defined(OS_POSIX)
+      close(s);
+#endif
       s = INVALID_SOCKET;
     }
   }
@@ -58,61 +90,139 @@ ListenSocket* ListenSocket::Listen(std::string ip, int port,
 void ListenSocket::Listen() {
   int backlog = 10; // TODO(erikkay): maybe don't allow any backlog?
   listen(socket_, backlog);
-  // TODO(erikkay): handle error
+  // TODO(erikkay): error handling
+#if defined(OS_POSIX)
+  WatchSocket(WAITING_ACCEPT);
+#endif
 }
 
 SOCKET ListenSocket::Accept(SOCKET s) {
   sockaddr_in from;
-  int from_len = sizeof(from);
-  SOCKET conn = accept(s, reinterpret_cast<SOCKADDR*>(&from), &from_len);
+  socklen_t from_len = sizeof(from);
+  SOCKET conn = accept(s, reinterpret_cast<sockaddr*>(&from), &from_len);
   if (conn != INVALID_SOCKET) {
-    // a non-blocking socket
-    unsigned long no_block = 1;
-    ioctlsocket(conn, FIONBIO, &no_block);
+    net::SetNonBlocking(conn);
   }
   return conn;
 }
 
 void ListenSocket::Accept() {
   SOCKET conn = Accept(socket_);
-  if (conn == INVALID_SOCKET) {
-    // TODO
-  } else {
+  if (conn != INVALID_SOCKET) {
     scoped_refptr<ListenSocket> sock =
         new ListenSocket(conn, socket_delegate_);
     // it's up to the delegate to AddRef if it wants to keep it around
+#if defined(OS_POSIX)
+    sock->WatchSocket(WAITING_READ);
+#endif
     socket_delegate_->DidAccept(this, sock);
+  } else {
+    // TODO(ibrar): some error handling required here
   }
 }
 
 void ListenSocket::Read() {
-  char buf[READ_BUF_SIZE+1];
+  char buf[kReadBufSize + 1]; // +1 for null termination
   int len;
   do {
-    len = recv(socket_, buf, READ_BUF_SIZE, 0);
+    len = recv(socket_, buf, kReadBufSize, 0);
     if (len == SOCKET_ERROR) {
+#if defined(OS_WIN)
       int err = WSAGetLastError();
       if (err == WSAEWOULDBLOCK) {
+#elif defined(OS_POSIX)
+      if (errno == EWOULDBLOCK || errno == EAGAIN) {
+#endif
         break;
       } else {
-        // TODO - error
+        // TODO(ibrar): some error handling required here
         break;
       }
     } else if (len == 0) {
-      // socket closed, ignore
+      // In Windows, Close() is called by OnObjectSignaled.  In POSIX, we need
+      // to call it here.
+#if defined(OS_POSIX)
+      Close();
+#endif
     } else {
-      // TODO(erikkay): maybe change DidRead to take a length instead
-      DCHECK(len > 0 && len <= READ_BUF_SIZE);
-      buf[len] = 0;
+      // TODO(ibrar): maybe change DidRead to take a length instead
+      DCHECK(len > 0 && len <= kReadBufSize);
+      buf[len] = 0; // already create a buffer with +1 length
       socket_delegate_->DidRead(this, buf);
     }
-  } while (len == READ_BUF_SIZE);
+  } while (len == kReadBufSize);
+}
+
+void ListenSocket::CloseSocket(SOCKET s) {
+  if (s && s != INVALID_SOCKET) {
+    UnwatchSocket();
+#if defined(OS_WIN)
+    closesocket(s);
+#elif defined(OS_POSIX)
+    close(s);
+#endif
+  }
 }
 
 void ListenSocket::Close() {
+#if defined(OS_POSIX)
+  if (wait_state_ == WAITING_CLOSE)
+    return;
+  wait_state_ = WAITING_CLOSE;
+#endif
   socket_delegate_->DidClose(this);
 }
 
+void ListenSocket::UnwatchSocket() {
+#if defined(OS_WIN)
+  watcher_.StopWatching();
+#elif defined(OS_POSIX)
+  MessageLoopForIO::current()->UnwatchSocket(event_.get());
+  wait_state_ = NOT_WAITING;
+#endif
+}
+
+void ListenSocket::WatchSocket(WaitState state) {
+#if defined(OS_WIN)
+  WSAEventSelect(socket_, socket_event_, FD_ACCEPT | FD_CLOSE | FD_READ);
+  watcher_.StartWatching(socket_event_, this);
+#elif defined(OS_POSIX)
+  MessageLoopForIO::current()->WatchSocket(
+      socket_, EV_READ|EV_PERSIST, event_.get(),this);
+  wait_state_ = state;
+#endif
+}
+
+void ListenSocket::SendInternal(const char* bytes, int len) {
+  int sent = send(socket_, bytes, len, 0);
+  if (sent == SOCKET_ERROR) {
+#if defined(OS_WIN)
+  int err = WSAGetLastError();
+  if (err == WSAEWOULDBLOCK) {
+#elif defined(OS_POSIX)
+    if (errno == EWOULDBLOCK || errno == EAGAIN) {
+#endif
+    // TODO (ibrar): there should be logic here to handle this because
+    // it is not an error
+    }
+  } else if (sent != len) {
+    LOG(ERROR) << "send failed: ";
+  }
+}
+
+void ListenSocket::Send(const char* bytes, int len, bool append_linefeed) {
+  SendInternal(bytes, len);
+  if (append_linefeed) {
+    SendInternal("\r\n", 2);
+  }
+}
+
+void ListenSocket::Send(const std::string& str, bool append_linefeed) {
+  Send(str.data(), static_cast<int>(str.length()), append_linefeed);
+}
+
+// TODO (ibrar): We can add these functions into OS dependent files
+#if defined(OS_WIN) 
 // MessageLoop watcher callback
 void ListenSocket::OnObjectSignaled(HANDLE object) {
   WSANETWORKEVENTS ev;
@@ -139,24 +249,17 @@ void ListenSocket::OnObjectSignaled(HANDLE object) {
     Close();
   }
 }
-
-void ListenSocket::SendInternal(const char* bytes, int len) {
-  int sent = send(socket_, bytes, len, 0);
-  if (sent == SOCKET_ERROR) {
-    // TODO
-  } else if (sent != len) {
-    // TODO
+#elif defined(OS_POSIX)
+void ListenSocket::OnSocketReady(short flags) {
+  if (wait_state_ == WAITING_ACCEPT) {
+    Accept();
   }
-}
-
-void ListenSocket::Send(const char* bytes, int len, bool append_linefeed) {
-  SendInternal(bytes, len);
-  if (append_linefeed) {
-    SendInternal("\r\n", 2);
+  if (wait_state_ == WAITING_READ) {
+    Read();
+  }
+  if (wait_state_ == WAITING_CLOSE) {
+    // Close() is called by Read() in the Linux case.
+    // TODO(erikkay): this seems to get hit multiple times after the close
   }
 }
-
-void ListenSocket::Send(const std::string& str, bool append_linefeed) {
-  Send(str.data(), static_cast<int>(str.length()), append_linefeed);
-}
-
+#endif
diff --git a/net/base/listen_socket.h b/net/base/listen_socket.h
index 0364e4d..2b32b5b 100644
--- a/net/base/listen_socket.h
+++ b/net/base/listen_socket.h
@@ -11,15 +11,33 @@
 #ifndef NET_BASE_SOCKET_H_
 #define NET_BASE_SOCKET_H_
 
-#include "base/basictypes.h"
+#if defined(OS_WIN)
+#include <winsock2.h>
 #include "base/object_watcher.h"
+#elif defined(OS_POSIX)
+#include "base/message_loop.h"
+#include "net/base/net_util.h"
+#include "net/base/net_errors.h"
+#include "third_party/libevent/event.h"
+#include "base/message_pump_libevent.h"
+#endif
+
+#include "base/basictypes.h"
 #include "base/ref_counted.h"
 
-#include <winsock2.h>
+#if defined(OS_POSIX)
+struct event;  // From libevent
+#define SOCKET int
+#endif
 
 // Implements a raw socket interface
 class ListenSocket : public base::RefCountedThreadSafe<ListenSocket>,
-                     public base::ObjectWatcher::Delegate {
+#if defined(OS_WIN)
+                     public base::ObjectWatcher::Delegate
+#elif defined(OS_POSIX)
+                     public base::MessagePumpLibevent::Watcher
+#endif
+{
  public:
   // TODO(erikkay): this delegate should really be split into two parts
   // to split up the listener from the connected socket.  Perhaps this class
@@ -52,19 +70,38 @@ class ListenSocket : public base::RefCountedThreadSafe<ListenSocket>,
 
   virtual void SendInternal(const char* bytes, int len);
 
-  // ObjectWatcher delegate
-  virtual void OnObjectSignaled(HANDLE object);
-
   virtual void Listen();
   virtual void Accept();
   virtual void Read();
   virtual void Close();
+  virtual void CloseSocket(SOCKET s);
 
-  SOCKET socket_;
+  enum WaitState {
+    NOT_WAITING      = 0,
+    WAITING_ACCEPT   = 1,
+    WAITING_READ     = 3,
+    WAITING_CLOSE    = 4	
+  };
+  // Pass any value in case of Windows, because in Windows 
+  // we are not using state. 
+  void WatchSocket(WaitState state); 
+  void UnwatchSocket();
+
+#if defined(OS_WIN)
+  // ObjectWatcher delegate
+  virtual void OnObjectSignaled(HANDLE object);
+  base::ObjectWatcher watcher_;
   HANDLE socket_event_;
+#elif defined(OS_POSIX)
+  WaitState wait_state_;
+  // The socket's libevent wrapper
+  scoped_ptr<event> event_;
+  // Called by MessagePumpLibevent when the socket is ready to do I/O
+  void OnSocketReady(short flags);
+#endif
+
+  SOCKET socket_; 
   ListenSocketDelegate *socket_delegate_;
-  
-  base::ObjectWatcher watcher_;
 
  private:
   DISALLOW_EVIL_CONSTRUCTORS(ListenSocket);
diff --git a/net/base/listen_socket_unittest.cc b/net/base/listen_socket_unittest.cc
index 1600d29..4baeeae 100644
--- a/net/base/listen_socket_unittest.cc
+++ b/net/base/listen_socket_unittest.cc
@@ -2,24 +2,307 @@
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
-// Tests ListenSocket.
-
 #include "net/base/listen_socket_unittest.h"
 
-namespace {
+#include "base/platform_test.h"
+#include "net/base/net_util.h"
+
+const int ListenSocketTester::kTestPort = 9999;
+
+static const int kReadBufSize = 1024;
+static const char* kHelloWorld = "HELLO, WORLD";
+static const int kMaxQueueSize = 20;
+static const char* kLoopback = "127.0.0.1";
+static const int kDefaultTimeoutMs = 5000;
+#if defined(OS_POSIX)
+static const char* kSemaphoreName = "chromium.listen_socket";
+#endif
+
+// millisecond sleep
+static void msleep(unsigned long milisec) {
+#if defined(OS_WIN)
+  Sleep(milisec);
+#elif defined(OS_POSIX)
+  struct timespec req = {0};
+  time_t sec = (int)(milisec / 1000);
+  milisec = milisec - (sec * 1000);
+  req.tv_sec = sec;
+  req.tv_nsec = milisec * 1000000L;
+  while (nanosleep(&req, &req) == -1)
+    continue;
+#endif
+}
+
+ListenSocket* ListenSocketTester::DoListen() {
+  return ListenSocket::Listen(kLoopback, kTestPort, this);
+}
+
+void ListenSocketTester::SetUp() {
+#if defined(OS_WIN)
+  InitializeCriticalSection(&lock_);
+  semaphore_ = CreateSemaphore(NULL, 0, kMaxQueueSize, NULL);
+  server_ = NULL;
+  net::EnsureWinsockInit();
+#elif defined(OS_POSIX)
+  ASSERT_EQ(0, pthread_mutex_init(&lock_, NULL ));
+  sem_unlink(kSemaphoreName);
+  semaphore_ = sem_open(kSemaphoreName, O_CREAT, 0, 0);
+  ASSERT_NE(SEM_FAILED, semaphore_); 
+#endif
+  base::Thread::Options options;
+  options.message_loop_type = MessageLoop::TYPE_IO;
+  thread_.reset(new base::Thread("socketio_test"));
+  thread_->StartWithOptions(options);
+  loop_ = (MessageLoopForIO*)thread_->message_loop();
+
+  loop_->PostTask(FROM_HERE, NewRunnableMethod(
+      this, &ListenSocketTester::Listen));
+
+  // verify Listen succeeded
+  ASSERT_TRUE(NextAction(kDefaultTimeoutMs));
+  ASSERT_FALSE(server_ == NULL);
+  ASSERT_EQ(ACTION_LISTEN, last_action_.type());
+
+  // verify the connect/accept and setup test_socket_
+  test_socket_ = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
+  struct sockaddr_in client;
+  client.sin_family = AF_INET;
+  client.sin_addr.s_addr = inet_addr(kLoopback);
+  client.sin_port = htons(kTestPort);
+  int ret = connect(test_socket_,
+                    reinterpret_cast<sockaddr*>(&client), sizeof(client));
+  ASSERT_NE(ret, SOCKET_ERROR);
+
+  net::SetNonBlocking(test_socket_);
+  ASSERT_TRUE(NextAction(kDefaultTimeoutMs));
+  ASSERT_EQ(ACTION_ACCEPT, last_action_.type());
+}
+
+void ListenSocketTester::TearDown() {
+  // verify close
+#if defined(OS_WIN)
+  closesocket(test_socket_);
+#elif defined(OS_POSIX)
+  close(test_socket_);
+#endif
+  ASSERT_TRUE(NextAction(kDefaultTimeoutMs));
+  ASSERT_EQ(ACTION_CLOSE, last_action_.type());
+
+  loop_->PostTask(FROM_HERE, NewRunnableMethod(
+      this, &ListenSocketTester::Shutdown));
+  ASSERT_TRUE(NextAction(kDefaultTimeoutMs));
+  ASSERT_EQ(ACTION_SHUTDOWN, last_action_.type());
+
+#if defined(OS_WIN)
+  CloseHandle(semaphore_);
+  semaphore_ = 0;
+  DeleteCriticalSection(&lock_);
+#elif defined(OS_POSIX)
+  ASSERT_EQ(0, pthread_mutex_lock(&lock_));
+  semaphore_ = NULL;
+  ASSERT_EQ(0, pthread_mutex_unlock(&lock_));
+  ASSERT_EQ(0, sem_unlink(kSemaphoreName));
+  ASSERT_EQ(0, pthread_mutex_destroy(&lock_));
+#endif
+
+  thread_.reset();
+  loop_ = NULL;
+}
+
+void ListenSocketTester::ReportAction(const ListenSocketTestAction& action) {
+#if defined(OS_WIN)
+  EnterCriticalSection(&lock_);
+  queue_.push_back(action);
+  LeaveCriticalSection(&lock_);
+  ReleaseSemaphore(semaphore_, 1, NULL);
+#elif defined(OS_POSIX)
+  ASSERT_EQ(0, pthread_mutex_lock(&lock_));
+  queue_.push_back(action);
+  ASSERT_EQ(0, pthread_mutex_unlock(&lock_));
+  ASSERT_EQ(0, sem_post(semaphore_));
+#endif
+}
+
+bool ListenSocketTester::NextAction(int timeout) {
+#if defined(OS_WIN)
+  DWORD ret = ::WaitForSingleObject(semaphore_, timeout);
+  if (ret != WAIT_OBJECT_0)
+    return false;
+  EnterCriticalSection(&lock_);
+  if (queue_.size() == 0) {
+    LeaveCriticalSection(&lock_);
+    return false;
+  }
+  last_action_ = queue_.front();
+  queue_.pop_front();
+  LeaveCriticalSection(&lock_);
+  return true;
+#elif defined(OS_POSIX)
+  if (semaphore_ == SEM_FAILED)
+    return false;
+  while (true) {
+    int result = sem_trywait(semaphore_);
+    msleep(1);  // 1ms sleep
+    timeout--;
+    if (timeout <= 0)
+      return false;
+    if (result == 0) 
+      break;
+  }
+  pthread_mutex_lock(&lock_);
+  if (queue_.size() == 0) {
+    pthread_mutex_unlock(&lock_);
+    return false;
+  }
+  last_action_ = queue_.front();
+  queue_.pop_front();
+  pthread_mutex_unlock(&lock_);
+  return true;
+#endif
+}
+
+int ListenSocketTester::ClearTestSocket() {
+  char buf[kReadBufSize];
+  int len_ret = 0;
+  int time_out = 0;
+  do {
+    int len = recv(test_socket_, buf, kReadBufSize, 0);
+#if defined(OS_WIN)
+    if (len == SOCKET_ERROR) {
+      int err = WSAGetLastError();
+      if (err == WSAEWOULDBLOCK) {
+#elif defined(OS_POSIX)
+    if (len == SOCKET_ERROR) {
+      if (errno == EWOULDBLOCK || errno == EAGAIN) {
+#endif
+        msleep(1); 
+        time_out++;
+        if (time_out > 10) 
+          break;
+        continue;  // still trying
+      }
+    } else if (len == 0) {
+      // socket closed
+      break;
+    } else {
+      time_out = 0;
+      len_ret += len;
+    }
+  } while (true);
+  return len_ret;
+}
 
-class ListenSocketTest: public testing::Test {
-public:
+void ListenSocketTester::Shutdown() {
+  connection_->Release();
+  connection_ = NULL;
+  server_->Release();
+  server_ = NULL;
+  ReportAction(ListenSocketTestAction(ACTION_SHUTDOWN));
+}
+
+void ListenSocketTester::Listen() {
+  server_ = DoListen();
+  if (server_) {
+    server_->AddRef();
+    ReportAction(ListenSocketTestAction(ACTION_LISTEN));
+  }
+}
+
+void ListenSocketTester::SendFromTester() {
+  connection_->Send(kHelloWorld);
+  ReportAction(ListenSocketTestAction(ACTION_SEND));
+}
+
+void ListenSocketTester::DidAccept(ListenSocket *server, 
+                                   ListenSocket *connection) {
+  connection_ = connection;
+  connection_->AddRef();
+  ReportAction(ListenSocketTestAction(ACTION_ACCEPT));
+}
+
+void ListenSocketTester::DidRead(ListenSocket *connection, 
+                                 const std::string& data) {
+  ReportAction(ListenSocketTestAction(ACTION_READ, data));
+}
+
+void ListenSocketTester::DidClose(ListenSocket *sock) {
+  ReportAction(ListenSocketTestAction(ACTION_CLOSE));
+}
+
+bool ListenSocketTester::Send(SOCKET sock, const std::string& str) {
+  int len = static_cast<int>(str.length());
+  int send_len = send(sock, str.data(), len, 0);
+  if (send_len == SOCKET_ERROR) {
+    LOG(ERROR) << "send failed: " << errno;
+    return false;
+  } else if (send_len != len) {
+    return false;
+  }
+  return true;
+}
+
+void ListenSocketTester::TestClientSend() {
+  ASSERT_TRUE(Send(test_socket_, kHelloWorld));
+  ASSERT_TRUE(NextAction(kDefaultTimeoutMs));
+  ASSERT_EQ(ACTION_READ, last_action_.type());
+  ASSERT_EQ(last_action_.data(), kHelloWorld);
+}
+
+void ListenSocketTester::TestClientSendLong() {
+  int hello_len = strlen(kHelloWorld);
+  std::string long_string;
+  int long_len = 0;
+  for (int i = 0; i < 200; i++) {
+    long_string += kHelloWorld;
+    long_len += hello_len;
+  }
+  ASSERT_TRUE(Send(test_socket_, long_string));
+  int read_len = 0;
+  while (read_len < long_len) {
+    ASSERT_TRUE(NextAction(kDefaultTimeoutMs));
+    ASSERT_EQ(ACTION_READ, last_action_.type());
+    std::string last_data = last_action_.data();
+    size_t len = last_data.length();
+    if (long_string.compare(read_len, len, last_data)) {
+      ASSERT_EQ(long_string.compare(read_len, len, last_data), 0);
+    }
+    read_len += static_cast<int>(last_data.length());
+  }
+  ASSERT_EQ(read_len, long_len);
+}
+
+void ListenSocketTester::TestServerSend() {
+  loop_->PostTask(FROM_HERE, NewRunnableMethod(
+      this, &ListenSocketTester::SendFromTester));
+  ASSERT_TRUE(NextAction(kDefaultTimeoutMs));
+  ASSERT_EQ(ACTION_SEND, last_action_.type());
+  // TODO(erikkay): Without this sleep, the recv seems to fail a small amount
+  // of the time.  I could fix this by making the socket blocking, but then
+  // this test might hang in the case of errors.  It would be nice to do
+  // something that felt more reliable here.
+  msleep(10);  // sleep for 10ms
+  const int buf_len = 200;
+  char buf[buf_len+1];
+  int recv_len = recv(test_socket_, buf, buf_len, 0);
+  buf[recv_len] = 0;
+  ASSERT_STREQ(buf, kHelloWorld);
+}
+
+
+class ListenSocketTest: public PlatformTest {
+ public:
   ListenSocketTest() {
     tester_ = NULL;
   }
 
   virtual void SetUp() {
+    PlatformTest::SetUp();
     tester_ = new ListenSocketTester();
     tester_->SetUp();
   }
 
   virtual void TearDown() {
+    PlatformTest::TearDown();
     tester_->TearDown();
     tester_ = NULL;
   }
@@ -27,8 +310,6 @@ public:
   scoped_refptr<ListenSocketTester> tester_;
 };
 
-}  // namespace
-
 TEST_F(ListenSocketTest, ClientSend) {
   tester_->TestClientSend();
 }
diff --git a/net/base/listen_socket_unittest.h b/net/base/listen_socket_unittest.h
index 97c592e..5a04099 100644
--- a/net/base/listen_socket_unittest.h
+++ b/net/base/listen_socket_unittest.h
@@ -5,25 +5,34 @@
 #ifndef NET_BASE_LISTEN_SOCKET_UNITTEST_H_
 #define NET_BASE_LISTEN_SOCKET_UNITTEST_H_
 
-#include <winsock2.h>
+#include "build/build_config.h"
 
-#include <deque>
-#include <string>
+#if defined(OS_WIN)
+#include <winsock2.h>
+#elif defined(OS_POSIX)
+#include <sys/socket.h>
+#include <errno.h>
+#include <semaphore.h>
+#include <arpa/inet.h>
+#endif
 
+#include "base/thread.h"
 #include "base/basictypes.h"
 #include "base/logging.h"
 #include "base/message_loop.h"
 #include "base/string_util.h"
 #include "base/thread.h"
+#include "net/base/net_util.h"
 #include "net/base/listen_socket.h"
 #include "net/base/winsock_init.h"
 #include "testing/gtest/include/gtest/gtest.h"
 
-namespace {
-
-const int TEST_PORT = 9999;
-const std::string HELLO_WORLD("HELLO, WORLD");
-const int MAX_QUEUE_SIZE = 20;
+#if defined(OS_POSIX)
+// Used same name as in Windows to avoid #ifdef where refrenced
+#define SOCKET int
+const int INVALID_SOCKET = -1; 
+const int SOCKET_ERROR = -1;
+#endif
 
 enum ActionType {
   ACTION_NONE = 0,
@@ -32,6 +41,7 @@ enum ActionType {
   ACTION_READ = 3,
   ACTION_SEND = 4,
   ACTION_CLOSE = 5,
+  ACTION_SHUTDOWN = 6
 };
 
 class ListenSocketTestAction {
@@ -46,222 +56,71 @@ class ListenSocketTestAction {
   const ActionType type() const { return action_; }
 
  private:
-  std::string data_;
   ActionType action_;
+  std::string data_;
 };
 
+
 // This had to be split out into a separate class because I couldn't
 // make a the testing::Test class refcounted.
 class ListenSocketTester :
     public ListenSocket::ListenSocketDelegate,
     public base::RefCountedThreadSafe<ListenSocketTester> {
+
  protected:
-  virtual ListenSocket* DoListen() {
-    return ListenSocket::Listen("127.0.0.1", TEST_PORT, this);
-  }
+  virtual ListenSocket* DoListen();
 
  public:
   ListenSocketTester()
-      : server_(NULL),
-        connection_(NULL),
-        thread_(NULL),
-        loop_(NULL) {
+      : thread_(NULL),
+        loop_(NULL), 
+        server_(NULL),
+        connection_(NULL){
   }
 
   virtual ~ListenSocketTester() {
   }
 
-  virtual void SetUp() {
-    InitializeCriticalSection(&lock_);
-    semaphore_ = CreateSemaphore(NULL, 0, MAX_QUEUE_SIZE, NULL);
-    server_ = NULL;
-    net::EnsureWinsockInit();
+  virtual void SetUp();
+  virtual void TearDown();
 
-    thread_.reset(new base::Thread("socketio_test"));
-    thread_->Start();
-    loop_ = thread_->message_loop();
-
-    loop_->PostTask(FROM_HERE, NewRunnableMethod(
-        this, &ListenSocketTester::Listen));
-
-    // verify Listen succeeded
-    ASSERT_TRUE(NextAction());
-    ASSERT_FALSE(server_ == NULL);
-    ASSERT_EQ(ACTION_LISTEN, last_action_.type());
-
-    // verify the connect/accept and setup test_socket_
-    test_socket_ = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
-    struct sockaddr_in client;
-    client.sin_family = AF_INET;
-    client.sin_addr.s_addr = inet_addr("127.0.0.1");
-    client.sin_port = htons(TEST_PORT);
-    int ret = connect(test_socket_,
-                      reinterpret_cast<SOCKADDR*>(&client), sizeof(client));
-    ASSERT_NE(ret, SOCKET_ERROR);
-    // non-blocking socket
-    unsigned long no_block = 1;
-    ioctlsocket(test_socket_, FIONBIO, &no_block);
-    ASSERT_TRUE(NextAction());
-    ASSERT_EQ(ACTION_ACCEPT, last_action_.type());
-  }
-
-  virtual void TearDown() {
-    // verify close
-    closesocket(test_socket_);
-    ASSERT_TRUE(NextAction(5000));
-    ASSERT_EQ(ACTION_CLOSE, last_action_.type());
-
-    CloseHandle(semaphore_);
-    semaphore_ = 0;
-    DeleteCriticalSection(&lock_);
-    if (connection_) {
-      loop_->ReleaseSoon(FROM_HERE, connection_);
-      connection_ = NULL;
-    }
-    if (server_) {
-      loop_->ReleaseSoon(FROM_HERE, server_);
-      server_ = NULL;
-    }
-    thread_.reset();
-    loop_ = NULL;
-  }
-
-  void ReportAction(const ListenSocketTestAction& action) {
-    EnterCriticalSection(&lock_);
-    queue_.push_back(action);
-    LeaveCriticalSection(&lock_);
-    ReleaseSemaphore(semaphore_, 1, NULL);
-  }
-
-  bool NextAction(int timeout = 5000) {
-    DWORD ret = ::WaitForSingleObject(semaphore_, timeout);
-    if (ret != WAIT_OBJECT_0)
-      return false;
-    EnterCriticalSection(&lock_);
-    if (queue_.size() == 0)
-      return false;
-    last_action_ = queue_.front();
-    queue_.pop_front();
-    LeaveCriticalSection(&lock_);
-    return true;
-  }
+  void ReportAction(const ListenSocketTestAction& action);
+  bool NextAction(int timeout);
 
   // read all pending data from the test socket
-  int ClearTestSocket() {
-    char buf[1024];
-    int len = 0;
-    do {
-      int ret = recv(test_socket_, buf, 1024, 0);
-      if (ret < 0) {
-        int err = WSAGetLastError();
-        if (err == WSAEWOULDBLOCK) {
-          break;
-        }
-      } else {
-        len += ret;
-      }
-    } while (true);
-    return len;
-  }
-
-  void Listen() {
-    server_ = DoListen();
-    if (server_) {
-      server_->AddRef();
-      ReportAction(ListenSocketTestAction(ACTION_LISTEN));
-    }
-  }
-
-  void SendFromTester() {
-    connection_->Send(HELLO_WORLD);
-    ReportAction(ListenSocketTestAction(ACTION_SEND));
-  }
-
-  virtual void DidAccept(ListenSocket *server, ListenSocket *connection) {
-    connection_ = connection;
-    connection_->AddRef();
-    ReportAction(ListenSocketTestAction(ACTION_ACCEPT));
-  }
-
-  virtual void DidRead(ListenSocket *connection, const std::string& data) {
-    ReportAction(ListenSocketTestAction(ACTION_READ, data));
-  }
-
-  virtual void DidClose(ListenSocket *sock) {
-    ReportAction(ListenSocketTestAction(ACTION_CLOSE));
-  }
-
-  virtual bool Send(SOCKET sock, const std::string& str) {
-    int len = static_cast<int>(str.length());
-    int send_len = send(sock, str.data(), len, 0);
-    if (send_len != len) {
-      return false;
-    }
-    return true;
-  }
-
+  int ClearTestSocket();
+  // Release the connection and server sockets
+  void Shutdown();
+  void Listen();
+  void SendFromTester();
+  virtual void DidAccept(ListenSocket *server, ListenSocket *connection);
+  virtual void DidRead(ListenSocket *connection, const std::string& data);
+  virtual void DidClose(ListenSocket *sock);
+  virtual bool Send(SOCKET sock, const std::string& str);
   // verify the send/read from client to server
-  void TestClientSend() {
-    ASSERT_TRUE(Send(test_socket_, HELLO_WORLD));
-    ASSERT_TRUE(NextAction());
-    ASSERT_EQ(ACTION_READ, last_action_.type());
-    ASSERT_EQ(last_action_.data(), HELLO_WORLD);
-  }
-
+  void TestClientSend();
   // verify send/read of a longer string
-  void TestClientSendLong() {
-    int hello_len = static_cast<int>(HELLO_WORLD.length());
-    std::string long_string;
-    int long_len = 0;
-    for (int i = 0; i < 200; i++) {
-      long_string += HELLO_WORLD;
-      long_len += hello_len;
-    }
-    ASSERT_TRUE(Send(test_socket_, long_string));
-    int read_len = 0;
-    while (read_len < long_len) {
-      ASSERT_TRUE(NextAction());
-      ASSERT_EQ(ACTION_READ, last_action_.type());
-      std::string last_data = last_action_.data();
-      size_t len = last_data.length();
-      if (long_string.compare(read_len, len, last_data)) {
-        ASSERT_EQ(long_string.compare(read_len, len, last_data), 0);
-      }
-      read_len += static_cast<int>(last_data.length());
-    }
-    ASSERT_EQ(read_len, long_len);
-  }
-
+  void TestClientSendLong();
   // verify a send/read from server to client
-  void TestServerSend() {
-    loop_->PostTask(FROM_HERE, NewRunnableMethod(
-        this, &ListenSocketTester::SendFromTester));
-    ASSERT_TRUE(NextAction());
-    ASSERT_EQ(ACTION_SEND, last_action_.type());
-    // TODO(erikkay): Without this sleep, the recv seems to fail a small amount
-    // of the time.  I could fix this by making the socket blocking, but then
-    // this test might hang in the case of errors.  It would be nice to do
-    // something that felt more reliable here.
-    Sleep(10);
-    const int buf_len = 200;
-    char buf[buf_len+1];
-    int recv_len = recv(test_socket_, buf, buf_len, 0);
-    buf[recv_len] = 0;
-    ASSERT_EQ(buf, HELLO_WORLD);
-  }
+  void TestServerSend();
+
+#if defined(OS_WIN)
+  CRITICAL_SECTION lock_;
+  HANDLE semaphore_;
+#elif defined(OS_POSIX)
+  pthread_mutex_t lock_;
+  sem_t* semaphore_;
+#endif
 
   scoped_ptr<base::Thread> thread_;
-  MessageLoop* loop_;
+  MessageLoopForIO* loop_;
   ListenSocket* server_;
   ListenSocket* connection_;
-  CRITICAL_SECTION lock_;
-  HANDLE semaphore_;
   ListenSocketTestAction last_action_;
   std::deque<ListenSocketTestAction> queue_;
   SOCKET test_socket_;
+  static const int kTestPort;
 };
 
-}  // namespace
-
 #endif  // NET_BASE_LISTEN_SOCKET_UNITTEST_H_
 
diff --git a/net/base/net_util.cc b/net/base/net_util.cc
index 507b7b4..05419fb 100644
--- a/net/base/net_util.cc
+++ b/net/base/net_util.cc
@@ -10,8 +10,14 @@
 #include <unicode/uscript.h>
 #include <unicode/uset.h>
 
-#ifdef OS_WIN
+#include "build/build_config.h"
+
+#if defined(OS_WIN)
 #include <windows.h>
+#include <winsock2.h>
+#elif defined(OS_POSIX)
+#include <sys/socket.h>
+#include <fcntl.h>
 #endif
 
 #include "net/base/net_util.h"
@@ -915,4 +921,16 @@ bool IsPortAllowedByFtp(int port) {
   return IsPortAllowedByDefault(port);
 }
 
+int SetNonBlocking(int fd) {
+#if defined(OS_WIN)
+  unsigned long no_block = 1;
+  return ioctlsocket(fd, FIONBIO, &no_block);
+#elif defined(OS_POSIX)
+  int flags = fcntl(fd, F_GETFL, 0);
+  if (-1 == flags)
+    flags = 0;
+  return fcntl(fd, F_SETFL, flags | O_NONBLOCK);
+#endif
+}
+
 }  // namespace net
diff --git a/net/base/net_util.h b/net/base/net_util.h
index db28a76..7adfc57 100644
--- a/net/base/net_util.h
+++ b/net/base/net_util.h
@@ -131,6 +131,9 @@ bool IsPortAllowedByDefault(int port);
 // restricted.
 bool IsPortAllowedByFtp(int port);
 
+// Set socket to non-blocking mode
+int SetNonBlocking(int fd);
+
 }  // namespace net
 
 #endif  // NET_BASE_NET_UTIL_H__
diff --git a/net/base/telnet_server.cc b/net/base/telnet_server.cc
index 025d2fc..3260951 100644
--- a/net/base/telnet_server.cc
+++ b/net/base/telnet_server.cc
@@ -2,13 +2,32 @@
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
+#include "build/build_config.h"
+
+#if defined(OS_WIN)
 // winsock2.h must be included first in order to ensure it is included before
 // windows.h.
 #include <winsock2.h>
+#elif defined(OS_POSIX)
+#include <errno.h>
+#include <sys/socket.h>
+#include "base/message_loop.h"
+#include "net/base/net_errors.h"
+#include "third_party/libevent/event.h"
+#include "base/message_pump_libevent.h"
+#endif
 
 #include "net/base/telnet_server.h"
 
-#define READ_BUF_SIZE 200
+#if defined(OS_POSIX)
+// Used same name as in Windows to avoid #ifdef where refrenced
+#define SOCKET int
+const int INVALID_SOCKET = -1; 
+const int SOCKET_ERROR = -1;
+struct event;  // From libevent
+#endif
+
+const int kReadBufSize = 200;
 
 // Telnet protocol constants.
 class TelnetProtocol {
@@ -109,7 +128,9 @@ void TelnetServer::Accept() {
   } else {
     scoped_refptr<TelnetServer> sock =
         new TelnetServer(conn, socket_delegate_);
-
+#if defined(OS_POSIX)
+    sock->WatchSocket(WAITING_READ);
+#endif
     // Setup the way we want to communicate
     sock->SendIAC(TelnetProtocol::DO, TelnetProtocol::ECHO);
     sock->SendIAC(TelnetProtocol::DO, TelnetProtocol::NAWS);
@@ -126,7 +147,7 @@ TelnetServer* TelnetServer::Listen(std::string ip, int port,
                                    ListenSocketDelegate *del) {
   SOCKET s = ListenSocket::Listen(ip, port);
   if (s == INVALID_SOCKET) {
-    // TODO
+    // TODO (ibrar): error handling
   } else {
     TelnetServer *serv = new TelnetServer(s, del);
     serv->Listen();
@@ -226,18 +247,23 @@ void TelnetServer::StateMachineStep(unsigned char c) {
 }
 
 void TelnetServer::Read() {
-  char buf[READ_BUF_SIZE];
+  char buf[kReadBufSize + 1];
   int len;
   do {
-    len = recv(socket_, buf, READ_BUF_SIZE, 0);
+    len = recv(socket_, buf, kReadBufSize, 0);
+
+#if defined(OS_WIN)
     if (len == SOCKET_ERROR) {
       int err = WSAGetLastError();
-      if (err == WSAEWOULDBLOCK) {
+      if (err == WSAEWOULDBLOCK)
         break;
-      } else {
-        // TODO - error
+#else
+    if (len == SOCKET_ERROR) {
+      if (errno == EWOULDBLOCK || errno == EAGAIN) 
         break;
-      }
+#endif
+    } else if (len == 0) {
+      Close();
     } else {
       const char *data = buf;
       for (int i = 0; i < len; ++i) {
@@ -246,6 +272,5 @@ void TelnetServer::Read() {
         data++;
       }
     }
-  } while (len == READ_BUF_SIZE);
+  } while (len == kReadBufSize);
 }
-
diff --git a/net/base/telnet_server_unittest.cc b/net/base/telnet_server_unittest.cc
index 5825c05..08a1607 100644
--- a/net/base/telnet_server_unittest.cc
+++ b/net/base/telnet_server_unittest.cc
@@ -4,35 +4,30 @@
 
 // Tests TelnetServer.
 
+#include "base/platform_test.h"
 #include "net/base/listen_socket_unittest.h"
 #include "net/base/telnet_server.h"
 
-namespace {
-
-const std::string CRLF("\r\n");
+static const char* kCRLF = "\r\n";
 
 class TelnetServerTester : public ListenSocketTester {
 public:
   virtual ListenSocket* DoListen() {
-    return TelnetServer::Listen("127.0.0.1", TEST_PORT, this);
+    return TelnetServer::Listen("127.0.0.1", kTestPort, this);
   }
 
   virtual void SetUp() {
     ListenSocketTester::SetUp();
     // With TelnetServer, there's some control codes sent at connect time,
     // so we need to eat those to avoid affecting the subsequent tests.
-    // TODO(erikkay): Unfortunately, without the sleep, we don't seem to
-    // reliably get the 15 bytes without an EWOULDBLOCK.  It would be nice if
-    // there were a more reliable mechanism here.
-    Sleep(10);
-    ASSERT_EQ(ClearTestSocket(), 15);
+    EXPECT_EQ(ClearTestSocket(), 15);
   }
 
   virtual bool Send(SOCKET sock, const std::string& str) {
     if (ListenSocketTester::Send(sock, str)) {
       // TelnetServer currently calls DidRead after a CRLF, so we need to
       // append one to the end of the data that we send.
-      if (ListenSocketTester::Send(sock, CRLF)) {
+      if (ListenSocketTester::Send(sock, kCRLF)) {
         return true;
       }
     }
@@ -40,18 +35,20 @@ public:
   }
 };
 
-class TelnetServerTest: public testing::Test {
+class TelnetServerTest: public PlatformTest {
 protected:
   TelnetServerTest() {
     tester_ = NULL;
   }
 
   virtual void SetUp() {
+    PlatformTest::SetUp();
     tester_ = new TelnetServerTester();
     tester_->SetUp();
   }
 
   virtual void TearDown() {
+    PlatformTest::TearDown();
     tester_->TearDown();
     tester_ = NULL;
   }
@@ -59,8 +56,6 @@ protected:
   scoped_refptr<TelnetServerTester> tester_;
 };
 
-}  // namespace
-
 TEST_F(TelnetServerTest, ServerClientSend) {
   tester_->TestClientSend();
 }