Porting of listen_socket, telnet_server to linux

net/base/listen_socket_unittest.h

Index: net/base/listen_socket_unittest.h
===================================================================
--- net/base/listen_socket_unittest.h	(revision 3911)
+++ net/base/listen_socket_unittest.h	(working copy)
@@ -5,20 +5,52 @@
 #ifndef NET_BASE_LISTEN_SOCKET_UNITTEST_H_
 #define NET_BASE_LISTEN_SOCKET_UNITTEST_H_
 
+#include "build/build_config.h"
+
+#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 <deque>
-#include <string>
-
+#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"
 
+#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
+
+const int kReadBufSize = 1024;
+
+// 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
+}
+
 namespace {
 
 const int TEST_PORT = 9999;
@@ -46,10 +78,11 @@
   const ActionType type() const { return action_; }
 
  private:
+  ActionType action_;
   std::string data_;
-  ActionType action_;
 };
 
+
 // This had to be split out into a separate class because I couldn't
 // make a the testing::Test class refcounted.
 class ListenSocketTester :
@@ -62,24 +95,30 @@
 
  public:
   ListenSocketTester()
-      : server_(NULL),
-        connection_(NULL),
-        thread_(NULL),
-        loop_(NULL) {
+      : thread_(NULL),
+        loop_(NULL), 
+        server_(NULL),
+        connection_(NULL){
   }
 
   virtual ~ListenSocketTester() {
   }
 
   virtual void SetUp() {
+#if defined(OS_WIN)
     InitializeCriticalSection(&lock_);
     semaphore_ = CreateSemaphore(NULL, 0, MAX_QUEUE_SIZE, NULL);
     server_ = NULL;
     net::EnsureWinsockInit();
-
+#elif defined(OS_POSIX)
+    pthread_mutex_init(&lock_, NULL );
+    sem_init(&semaphore_, 0, 0);  
+#endif
+    base::Thread::Options options;
+    options.message_loop_type = MessageLoop::TYPE_IO;
     thread_.reset(new base::Thread("socketio_test"));
-    thread_->Start();
-    loop_ = thread_->message_loop();
+    thread_->StartWithOptions(options);
+    loop_ = (MessageLoopForIO*)thread_->message_loop();
 
     loop_->PostTask(FROM_HERE, NewRunnableMethod(
         this, &ListenSocketTester::Listen));
@@ -96,24 +135,32 @@
     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));
+                      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);
+    net::SetNonBlocking(test_socket_);
     ASSERT_TRUE(NextAction());
     ASSERT_EQ(ACTION_ACCEPT, last_action_.type());
   }
 
   virtual void TearDown() {
     // verify close
+#if defined(OS_WIN)
     closesocket(test_socket_);
+#elif defined(OS_POSIX)
+    close(test_socket_);
+#endif
     ASSERT_TRUE(NextAction(5000));
     ASSERT_EQ(ACTION_CLOSE, last_action_.type());
-
+#if defined(OS_WIN)
     CloseHandle(semaphore_);
     semaphore_ = 0;
     DeleteCriticalSection(&lock_);
+#elif defined(OS_POSIX)
+    sem_destroy(&semaphore_);
+    pthread_mutex_destroy(&lock_);
+#endif
     if (connection_) {
       loop_->ReleaseSoon(FROM_HERE, connection_);
       connection_ = NULL;
@@ -127,41 +174,84 @@
   }
 
   void ReportAction(const ListenSocketTestAction& action) {
+#if defined(OS_WIN)
     EnterCriticalSection(&lock_);
     queue_.push_back(action);
     LeaveCriticalSection(&lock_);
     ReleaseSemaphore(semaphore_, 1, NULL);
+#elif defined(OS_POSIX)
+    pthread_mutex_lock(&lock_);
+    queue_.push_back(action);
+    pthread_mutex_unlock(&lock_);
+    sem_post(&semaphore_);
+#endif
   }
 
   bool NextAction(int timeout = 5000) {
+#if defined(OS_WIN)
     DWORD ret = ::WaitForSingleObject(semaphore_, timeout);
     if (ret != WAIT_OBJECT_0)
       return false;
     EnterCriticalSection(&lock_);
-    if (queue_.size() == 0)
+    if (queue_.size() == 0) {
+      LeaveCriticalSection(&lock_);
       return false;
+    }
     last_action_ = queue_.front();
     queue_.pop_front();
     LeaveCriticalSection(&lock_);
     return true;
+#elif defined(OS_POSIX)
+    while (true) {
+      int result = sem_trywait(&semaphore_);
+      msleep(1); //1ms sleep
+      timeout--;
+      if(timeout <= 0) break;
+      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
   }
 
   // read all pending data from the test socket
   int ClearTestSocket() {
-    char buf[1024];
-    int len = 0;
+    char buf[kReadBufSize];
+    int len_ret = 0;
+    int len;
+    int time_out = 0;
     do {
-      int ret = recv(test_socket_, buf, 1024, 0);
-      if (ret < 0) {
+      len = recv(test_socket_, buf, kReadBufSize, 0);
+
+#if defined(OS_WIN)
+      if (len == SOCKET_ERROR) {
         int err = WSAGetLastError();
         if (err == WSAEWOULDBLOCK) {
-          break;
+#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) {
+        return len_ret;
       } else {
-        len += ret;
+        time_out = 0;
+        len_ret += len;
       }
-    } while (true);
-    return len;
+    } while (len == kReadBufSize);
+    return len_ret;
   }
 
   void Listen() {
@@ -194,8 +284,11 @@
   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) {
+    if (send_len == SOCKET_ERROR) {
+      LOG(ERROR) << "send failed: " << strerror(errno);
       return false;
+    } else if (send_len != len) {
+      return false;
     }
     return true;
   }
@@ -242,20 +335,25 @@
     // 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);
+    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_EQ(buf, HELLO_WORLD);
   }
+#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_;