base::Bind: Convert Socket::Read.

BUG=none
TEST=none
R=csilv

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

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

6 files changed, 128 insertions, 21 deletions

diff --git a/remoting/jingle_glue/ssl_socket_adapter.cc b/remoting/jingle_glue/ssl_socket_adapter.cc
index 1948c68..7414c4b 100644
--- a/remoting/jingle_glue/ssl_socket_adapter.cc
+++ b/remoting/jingle_glue/ssl_socket_adapter.cc
@@ -187,7 +187,7 @@ void SSLSocketAdapter::OnConnectEvent(talk_base::AsyncSocket* socket) {
 
 TransportSocket::TransportSocket(talk_base::AsyncSocket* socket,
                                  SSLSocketAdapter *ssl_adapter)
-    : read_callback_(NULL),
+    : old_read_callback_(NULL),
       write_callback_(NULL),
       read_buffer_len_(0),
       write_buffer_len_(0),
@@ -290,7 +290,25 @@ base::TimeDelta TransportSocket::GetConnectTimeMicros() const {
 int TransportSocket::Read(net::IOBuffer* buf, int buf_len,
                           net::OldCompletionCallback* callback) {
   DCHECK(buf);
-  DCHECK(!read_callback_);
+  DCHECK(!old_read_callback_ && read_callback_.is_null());
+  DCHECK(!read_buffer_.get());
+  int result = socket_->Recv(buf->data(), buf_len);
+  if (result < 0) {
+    result = net::MapSystemError(socket_->GetError());
+    if (result == net::ERR_IO_PENDING) {
+      old_read_callback_ = callback;
+      read_buffer_ = buf;
+      read_buffer_len_ = buf_len;
+    }
+  }
+  if (result != net::ERR_IO_PENDING)
+    was_used_to_convey_data_ = true;
+  return result;
+}
+int TransportSocket::Read(net::IOBuffer* buf, int buf_len,
+                          const net::CompletionCallback& callback) {
+  DCHECK(buf);
+  DCHECK(!old_read_callback_ && read_callback_.is_null());
   DCHECK(!read_buffer_.get());
   int result = socket_->Recv(buf->data(), buf_len);
   if (result < 0) {
@@ -336,13 +354,15 @@ bool TransportSocket::SetSendBufferSize(int32 size) {
 }
 
 void TransportSocket::OnReadEvent(talk_base::AsyncSocket* socket) {
-  if (read_callback_) {
+  if (old_read_callback_ || !read_callback_.is_null()) {
     DCHECK(read_buffer_.get());
-    net::OldCompletionCallback* callback = read_callback_;
+    net::OldCompletionCallback* old_callback = old_read_callback_;
+    net::CompletionCallback callback = read_callback_;
     scoped_refptr<net::IOBuffer> buffer = read_buffer_;
     int buffer_len = read_buffer_len_;
 
-    read_callback_ = NULL;
+    old_read_callback_ = NULL;
+    read_callback_.Reset();
     read_buffer_ = NULL;
     read_buffer_len_ = 0;
 
@@ -350,6 +370,7 @@ void TransportSocket::OnReadEvent(talk_base::AsyncSocket* socket) {
     if (result < 0) {
       result = net::MapSystemError(socket_->GetError());
       if (result == net::ERR_IO_PENDING) {
+        old_read_callback_ = old_callback;
         read_callback_ = callback;
         read_buffer_ = buffer;
         read_buffer_len_ = buffer_len;
@@ -357,7 +378,10 @@ void TransportSocket::OnReadEvent(talk_base::AsyncSocket* socket) {
       }
     }
     was_used_to_convey_data_ = true;
-    callback->RunWithParams(Tuple1<int>(result));
+    if (old_callback)
+      old_callback->RunWithParams(Tuple1<int>(result));
+    else
+      callback.Run(result);
   }
 }
 
diff --git a/remoting/jingle_glue/ssl_socket_adapter.h b/remoting/jingle_glue/ssl_socket_adapter.h
index ffa593d..f929bf8 100644
--- a/remoting/jingle_glue/ssl_socket_adapter.h
+++ b/remoting/jingle_glue/ssl_socket_adapter.h
@@ -55,10 +55,11 @@ class TransportSocket : public net::StreamSocket, public sigslot::has_slots<> {
   virtual int64 NumBytesRead() const OVERRIDE;
   virtual base::TimeDelta GetConnectTimeMicros() const OVERRIDE;
 
-  // net::Socket implementation
-
+  // net::Socket implementation.
   virtual int Read(net::IOBuffer* buf, int buf_len,
                    net::OldCompletionCallback* callback) OVERRIDE;
+  virtual int Read(net::IOBuffer* buf, int buf_len,
+                   const net::CompletionCallback& callback) OVERRIDE;
   virtual int Write(net::IOBuffer* buf, int buf_len,
                     net::OldCompletionCallback* callback) OVERRIDE;
   virtual bool SetReceiveBufferSize(int32 size) OVERRIDE;
@@ -70,7 +71,8 @@ class TransportSocket : public net::StreamSocket, public sigslot::has_slots<> {
   void OnReadEvent(talk_base::AsyncSocket* socket);
   void OnWriteEvent(talk_base::AsyncSocket* socket);
 
-  net::OldCompletionCallback* read_callback_;
+  net::OldCompletionCallback* old_read_callback_;
+  net::CompletionCallback read_callback_;
   net::OldCompletionCallback* write_callback_;
 
   scoped_refptr<net::IOBuffer> read_buffer_;
diff --git a/remoting/protocol/fake_session.cc b/remoting/protocol/fake_session.cc
index 3b5499a..f1a890a 100644
--- a/remoting/protocol/fake_session.cc
+++ b/remoting/protocol/fake_session.cc
@@ -36,7 +36,10 @@ void FakeSocket::AppendInputData(const char* data, int data_size) {
     memcpy(read_buffer_->data(),
            &(*input_data_.begin()) + input_pos_, result);
     input_pos_ += result;
-    read_callback_->Run(result);
+    if (old_read_callback_)
+      old_read_callback_->Run(result);
+    else
+      read_callback_.Run(result);
     read_buffer_ = NULL;
   }
 }
@@ -54,6 +57,23 @@ int FakeSocket::Read(net::IOBuffer* buf, int buf_len,
     read_pending_ = true;
     read_buffer_ = buf;
     read_buffer_size_ = buf_len;
+    old_read_callback_ = callback;
+    return net::ERR_IO_PENDING;
+  }
+}
+int FakeSocket::Read(net::IOBuffer* buf, int buf_len,
+                     const net::CompletionCallback& callback) {
+  EXPECT_EQ(message_loop_, MessageLoop::current());
+  if (input_pos_ < static_cast<int>(input_data_.size())) {
+    int result = std::min(buf_len,
+                          static_cast<int>(input_data_.size()) - input_pos_);
+    memcpy(buf->data(), &(*input_data_.begin()) + input_pos_, result);
+    input_pos_ += result;
+    return result;
+  } else {
+    read_pending_ = true;
+    read_buffer_ = buf;
+    read_buffer_size_ = buf_len;
     read_callback_ = callback;
     return net::ERR_IO_PENDING;
   }
@@ -146,6 +166,7 @@ base::TimeDelta FakeSocket::GetConnectTimeMicros() const {
 
 FakeUdpSocket::FakeUdpSocket()
     : read_pending_(false),
+      old_read_callback_(NULL),
       input_pos_(0),
       message_loop_(MessageLoop::current()) {
 }
@@ -165,7 +186,10 @@ void FakeUdpSocket::AppendInputPacket(const char* data, int data_size) {
     int result = std::min(data_size, read_buffer_size_);
     memcpy(read_buffer_->data(), data, result);
     input_pos_ = input_packets_.size();
-    read_callback_->Run(result);
+    if (old_read_callback_)
+      old_read_callback_->Run(result);
+    else
+      old_read_callback_->Run(result);
     read_buffer_ = NULL;
   }
 }
@@ -183,6 +207,23 @@ int FakeUdpSocket::Read(net::IOBuffer* buf, int buf_len,
     read_pending_ = true;
     read_buffer_ = buf;
     read_buffer_size_ = buf_len;
+    old_read_callback_ = callback;
+    return net::ERR_IO_PENDING;
+  }
+}
+int FakeUdpSocket::Read(net::IOBuffer* buf, int buf_len,
+                        const net::CompletionCallback& callback) {
+  EXPECT_EQ(message_loop_, MessageLoop::current());
+  if (input_pos_ < static_cast<int>(input_packets_.size())) {
+    int result = std::min(
+        buf_len, static_cast<int>(input_packets_[input_pos_].size()));
+    memcpy(buf->data(), &(*input_packets_[input_pos_].begin()), result);
+    ++input_pos_;
+    return result;
+  } else {
+    read_pending_ = true;
+    read_buffer_ = buf;
+    read_buffer_size_ = buf_len;
     read_callback_ = callback;
     return net::ERR_IO_PENDING;
   }
diff --git a/remoting/protocol/fake_session.h b/remoting/protocol/fake_session.h
index 2faa554..a8239f5 100644
--- a/remoting/protocol/fake_session.h
+++ b/remoting/protocol/fake_session.h
@@ -10,6 +10,7 @@
 #include <vector>
 
 #include "base/memory/scoped_ptr.h"
+#include "net/base/completion_callback.h"
 #include "net/socket/socket.h"
 #include "net/socket/stream_socket.h"
 #include "remoting/protocol/session.h"
@@ -40,6 +41,8 @@ class FakeSocket : public net::StreamSocket {
   // net::Socket implementation.
   virtual int Read(net::IOBuffer* buf, int buf_len,
                    net::OldCompletionCallback* callback) OVERRIDE;
+  virtual int Read(net::IOBuffer* buf, int buf_len,
+                   const net::CompletionCallback& callback) OVERRIDE;
   virtual int Write(net::IOBuffer* buf, int buf_len,
                     net::OldCompletionCallback* callback) OVERRIDE;
 
@@ -66,7 +69,8 @@ class FakeSocket : public net::StreamSocket {
   bool read_pending_;
   scoped_refptr<net::IOBuffer> read_buffer_;
   int read_buffer_size_;
-  net::OldCompletionCallback* read_callback_;
+  net::OldCompletionCallback* old_read_callback_;
+  net::CompletionCallback read_callback_;
 
   std::string written_data_;
   std::string input_data_;
@@ -97,6 +101,8 @@ class FakeUdpSocket : public net::Socket {
   // net::Socket implementation.
   virtual int Read(net::IOBuffer* buf, int buf_len,
                    net::OldCompletionCallback* callback) OVERRIDE;
+  virtual int Read(net::IOBuffer* buf, int buf_len,
+                   const net::CompletionCallback& callback) OVERRIDE;
   virtual int Write(net::IOBuffer* buf, int buf_len,
                     net::OldCompletionCallback* callback) OVERRIDE;
 
@@ -107,7 +113,8 @@ class FakeUdpSocket : public net::Socket {
   bool read_pending_;
   scoped_refptr<net::IOBuffer> read_buffer_;
   int read_buffer_size_;
-  net::OldCompletionCallback* read_callback_;
+  net::OldCompletionCallback* old_read_callback_;
+  net::CompletionCallback read_callback_;
 
   std::vector<std::string> written_packets_;
   std::vector<std::string> input_packets_;
diff --git a/remoting/protocol/pepper_transport_socket_adapter.cc b/remoting/protocol/pepper_transport_socket_adapter.cc
index a48afce..0385b9b 100644
--- a/remoting/protocol/pepper_transport_socket_adapter.cc
+++ b/remoting/protocol/pepper_transport_socket_adapter.cc
@@ -49,7 +49,7 @@ PepperTransportSocketAdapter::PepperTransportSocketAdapter(
       transport_(transport),
       connected_(false),
       get_address_pending_(false),
-      read_callback_(NULL),
+      old_read_callback_(NULL),
       write_callback_(NULL) {
   callback_factory_.Initialize(this);
 }
@@ -68,7 +68,29 @@ void PepperTransportSocketAdapter::AddRemoteCandidate(
 int PepperTransportSocketAdapter::Read(net::IOBuffer* buf, int buf_len,
                                        net::OldCompletionCallback* callback) {
   DCHECK(CalledOnValidThread());
-  DCHECK(!read_callback_);
+  DCHECK(!old_read_callback_ && read_callback_.is_null());
+  DCHECK(!read_buffer_);
+
+  if (!transport_.get())
+    return net::ERR_SOCKET_NOT_CONNECTED;
+
+  int result = PPErrorToNetError(transport_->Recv(
+      buf->data(), buf_len,
+      callback_factory_.NewOptionalCallback(
+          &PepperTransportSocketAdapter::OnRead)));
+
+  if (result == net::ERR_IO_PENDING) {
+    old_read_callback_ = callback;
+    read_buffer_ = buf;
+  }
+
+  return result;
+}
+int PepperTransportSocketAdapter::Read(
+    net::IOBuffer* buf, int buf_len,
+    const net::CompletionCallback& callback) {
+  DCHECK(CalledOnValidThread());
+  DCHECK(!old_read_callback_ && read_callback_.is_null());
   DCHECK(!read_buffer_);
 
   if (!transport_.get())
@@ -296,13 +318,20 @@ void PepperTransportSocketAdapter::OnConnect(int result) {
 
 void PepperTransportSocketAdapter::OnRead(int32_t result) {
   DCHECK(CalledOnValidThread());
-  DCHECK(read_callback_);
+  DCHECK(old_read_callback_ || read_callback_.is_null());
   DCHECK(read_buffer_);
 
-  net::OldCompletionCallback* callback = read_callback_;
-  read_callback_ = NULL;
-  read_buffer_ = NULL;
-  callback->Run(PPErrorToNetError(result));
+  if (old_read_callback_) {
+    net::OldCompletionCallback* callback = old_read_callback_;
+    old_read_callback_ = NULL;
+    read_buffer_ = NULL;
+    callback->Run(PPErrorToNetError(result));
+  } else {
+    net::CompletionCallback callback = read_callback_;
+    read_callback_.Reset();
+    read_buffer_ = NULL;
+    callback.Run(PPErrorToNetError(result));
+  }
 }
 
 void PepperTransportSocketAdapter::OnWrite(int32_t result) {
diff --git a/remoting/protocol/pepper_transport_socket_adapter.h b/remoting/protocol/pepper_transport_socket_adapter.h
index c25506d..40b75ce 100644
--- a/remoting/protocol/pepper_transport_socket_adapter.h
+++ b/remoting/protocol/pepper_transport_socket_adapter.h
@@ -10,6 +10,7 @@
 #include "base/callback.h"
 #include "base/memory/scoped_ptr.h"
 #include "base/threading/non_thread_safe.h"
+#include "net/base/completion_callback.h"
 #include "net/base/net_log.h"
 #include "net/socket/stream_socket.h"
 #include "ppapi/c/pp_stdint.h"
@@ -50,6 +51,8 @@ class PepperTransportSocketAdapter : public base::NonThreadSafe,
   // net::Socket implementation.
   virtual int Read(net::IOBuffer* buf, int buf_len,
                    net::OldCompletionCallback* callback) OVERRIDE;
+  virtual int Read(net::IOBuffer* buf, int buf_len,
+                   const net::CompletionCallback& callback) OVERRIDE;
   virtual int Write(net::IOBuffer* buf, int buf_len,
                     net::OldCompletionCallback* callback) OVERRIDE;
   virtual bool SetReceiveBufferSize(int32 size) OVERRIDE;
@@ -91,7 +94,8 @@ class PepperTransportSocketAdapter : public base::NonThreadSafe,
 
   bool get_address_pending_;
 
-  net::OldCompletionCallback* read_callback_;
+  net::OldCompletionCallback* old_read_callback_;
+  net::CompletionCallback read_callback_;
   scoped_refptr<net::IOBuffer> read_buffer_;
 
   net::OldCompletionCallback* write_callback_;