Implement ChromiumSocketFactory.

The new PacketSocketFactory will be used by chromoting host in order to be able to
use chromium UDP sockets instead of libjingle sockets

BUG=137140


Review URL: https://chromiumcodereview.appspot.com/10783028

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

4 files changed, 492 insertions, 0 deletions

diff --git a/remoting/jingle_glue/chromium_socket_factory.cc b/remoting/jingle_glue/chromium_socket_factory.cc
new file mode 100644
index 0000000..7de727c
--- /dev/null
+++ b/remoting/jingle_glue/chromium_socket_factory.cc
@@ -0,0 +1,357 @@
+// Copyright (c) 2012 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 "remoting/jingle_glue/chromium_socket_factory.h"
+
+#include "base/bind.h"
+#include "base/logging.h"
+#include "base/memory/scoped_ptr.h"
+#include "jingle/glue/utils.h"
+#include "net/base/io_buffer.h"
+#include "net/base/ip_endpoint.h"
+#include "net/base/net_errors.h"
+#include "net/udp/udp_server_socket.h"
+#include "third_party/libjingle/source/talk/base/asyncpacketsocket.h"
+
+namespace remoting {
+
+namespace {
+
+// Size of the buffer to allocate for RecvFrom().
+const int kReceiveBufferSize = 65536;
+
+// Maximum amount of data in the send buffers. This is necessary to
+// prevent out-of-memory crashes if the caller sends data faster than
+// Pepper's UDP API can handle it. This maximum should never be
+// reached under normal conditions.
+const int kMaxSendBufferSize = 256 * 1024;
+
+class UdpPacketSocket : public talk_base::AsyncPacketSocket {
+ public:
+  UdpPacketSocket();
+  virtual ~UdpPacketSocket();
+
+  bool Init(const talk_base::SocketAddress& local_address,
+            int min_port, int max_port);
+
+  // talk_base::AsyncPacketSocket interface.
+  virtual talk_base::SocketAddress GetLocalAddress() const;
+  virtual talk_base::SocketAddress GetRemoteAddress() const;
+  virtual int Send(const void* data, size_t data_size);
+  virtual int SendTo(const void* data, size_t data_size,
+                     const talk_base::SocketAddress& address);
+  virtual int Close();
+  virtual State GetState() const;
+  virtual int GetOption(talk_base::Socket::Option option, int* value);
+  virtual int SetOption(talk_base::Socket::Option option, int value);
+  virtual int GetError() const;
+  virtual void SetError(int error);
+
+ private:
+  struct PendingPacket {
+    PendingPacket(const void* buffer,
+                  int buffer_size,
+                  const net::IPEndPoint& address);
+
+    scoped_refptr<net::IOBufferWithSize> data;
+    net::IPEndPoint address;
+  };
+
+  void OnBindCompleted(int error);
+
+  void DoSend();
+  void OnSendCompleted(int result);
+
+  void DoRead();
+  void OnReadCompleted(int result);
+  void HandleReadResult(int result);
+
+  scoped_ptr<net::UDPServerSocket> socket_;
+
+  State state_;
+  int error_;
+
+  talk_base::SocketAddress local_address_;
+
+  // Receive buffer and address are populated by asynchronous reads.
+  scoped_refptr<net::IOBuffer> receive_buffer_;
+  net::IPEndPoint receive_address_;
+
+  bool send_pending_;
+  std::list<PendingPacket> send_queue_;
+  int send_queue_size_;
+
+  DISALLOW_COPY_AND_ASSIGN(UdpPacketSocket);
+};
+
+UdpPacketSocket::PendingPacket::PendingPacket(
+    const void* buffer,
+    int buffer_size,
+    const net::IPEndPoint& address)
+    : data(new net::IOBufferWithSize(buffer_size)),
+      address(address) {
+  memcpy(data->data(), buffer, buffer_size);
+}
+
+UdpPacketSocket::UdpPacketSocket()
+    : state_(STATE_CLOSED),
+      error_(0),
+      send_pending_(false),
+      send_queue_size_(0) {
+}
+
+UdpPacketSocket::~UdpPacketSocket() {
+  Close();
+}
+
+bool UdpPacketSocket::Init(const talk_base::SocketAddress& local_address,
+                           int min_port, int max_port) {
+  net::IPEndPoint local_endpoint;
+  if (!jingle_glue::SocketAddressToIPEndPoint(
+          local_address, &local_endpoint)) {
+    return false;
+  }
+
+  for (int port = min_port; port <= max_port; ++port) {
+    socket_.reset(new net::UDPServerSocket(NULL, net::NetLog::Source()));
+    int result = socket_->Listen(
+        net::IPEndPoint(local_endpoint.address(), port));
+    if (result == net::OK) {
+      break;
+    } else {
+      socket_.reset();
+    }
+  }
+
+  if (!socket_.get()) {
+    // Failed to bind the socket.
+    return false;
+  }
+
+  if (socket_->GetLocalAddress(&local_endpoint) != net::OK ||
+      !jingle_glue::IPEndPointToSocketAddress(local_endpoint,
+                                              &local_address_)) {
+    return false;
+  }
+
+  state_ = STATE_BOUND;
+  DoRead();
+
+  return true;
+}
+
+talk_base::SocketAddress UdpPacketSocket::GetLocalAddress() const {
+  DCHECK_EQ(state_, STATE_BOUND);
+  return local_address_;
+}
+
+talk_base::SocketAddress UdpPacketSocket::GetRemoteAddress() const {
+  // UDP sockets are not connected - this method should never be called.
+  NOTREACHED();
+  return talk_base::SocketAddress();
+}
+
+int UdpPacketSocket::Send(const void* data, size_t data_size) {
+  // UDP sockets are not connected - this method should never be called.
+  NOTREACHED();
+  return EWOULDBLOCK;
+}
+
+int UdpPacketSocket::SendTo(const void* data, size_t data_size,
+                            const talk_base::SocketAddress& address) {
+  if (state_ != STATE_BOUND) {
+    NOTREACHED();
+    return EINVAL;
+  }
+
+  if (error_ != 0) {
+    return error_;
+  }
+
+  net::IPEndPoint endpoint;
+  if (!jingle_glue::SocketAddressToIPEndPoint(address, &endpoint)) {
+    return EINVAL;
+  }
+
+  if (send_queue_size_ >= kMaxSendBufferSize) {
+    return EWOULDBLOCK;
+  }
+
+  send_queue_.push_back(PendingPacket(data, data_size, endpoint));
+  send_queue_size_ += data_size;
+
+  DoSend();
+  return data_size;
+}
+
+int UdpPacketSocket::Close() {
+  state_ = STATE_CLOSED;
+  socket_.reset();
+  return 0;
+}
+
+talk_base::AsyncPacketSocket::State UdpPacketSocket::GetState() const {
+  return state_;
+}
+
+int UdpPacketSocket::GetOption(talk_base::Socket::Option option, int* value) {
+  // This method is never called by libjingle.
+  NOTIMPLEMENTED();
+  return -1;
+}
+
+int UdpPacketSocket::SetOption(talk_base::Socket::Option option, int value) {
+  if (state_ != STATE_BOUND) {
+    NOTREACHED();
+    return EINVAL;
+  }
+
+  switch (option) {
+    case talk_base::Socket::OPT_DONTFRAGMENT:
+      NOTIMPLEMENTED();
+      return -1;
+
+    case talk_base::Socket::OPT_RCVBUF: {
+      bool success = socket_->SetReceiveBufferSize(value);
+      return success ? 0 : -1;
+    }
+
+    case talk_base::Socket::OPT_SNDBUF: {
+      bool success = socket_->SetSendBufferSize(value);
+      return success ? 0 : -1;
+    }
+
+    case talk_base::Socket::OPT_NODELAY:
+      // OPT_NODELAY is only for TCP sockets.
+      NOTREACHED();
+      return -1;
+
+    case talk_base::Socket::OPT_IPV6_V6ONLY:
+      NOTIMPLEMENTED();
+      return -1;
+  }
+
+  NOTREACHED();
+  return -1;
+}
+
+int UdpPacketSocket::GetError() const {
+  return error_;
+}
+
+void UdpPacketSocket::SetError(int error) {
+  error_ = error;
+}
+
+void UdpPacketSocket::DoSend() {
+  if (send_pending_ || send_queue_.empty())
+    return;
+
+  PendingPacket& packet = send_queue_.front();
+  int result = socket_->SendTo(
+      packet.data, packet.data->size(), packet.address,
+      base::Bind(&UdpPacketSocket::OnSendCompleted,
+                 base::Unretained(this)));
+  if (result == net::ERR_IO_PENDING) {
+    send_pending_ = true;
+  } else {
+    OnSendCompleted(result);
+  }
+}
+
+void UdpPacketSocket::OnSendCompleted(int result) {
+  send_pending_ = false;
+
+  if (result < 0) {
+    // Treat all errors except ERR_ADDRESS_UNREACHABLE as fatal.
+    if (result != net::ERR_ADDRESS_UNREACHABLE) {
+      LOG(ERROR) << "Send failed on a UDP socket: " << result;
+      error_ = EINVAL;
+      return;
+    }
+  }
+
+  // Don't need to worry about partial sends because this is a datagram
+  // socket.
+  send_queue_size_ -= send_queue_.front().data->size();
+  send_queue_.pop_front();
+  DoSend();
+}
+
+void UdpPacketSocket::DoRead() {
+  int result = 0;
+  while (result >= 0) {
+    receive_buffer_ = new net::IOBuffer(kReceiveBufferSize);
+    result = socket_->RecvFrom(
+        receive_buffer_, kReceiveBufferSize, &receive_address_,
+        base::Bind(&UdpPacketSocket::OnReadCompleted, base::Unretained(this)));
+    HandleReadResult(result);
+  }
+}
+
+void UdpPacketSocket::OnReadCompleted(int result) {
+  HandleReadResult(result);
+  if (result >= 0) {
+    DoRead();
+  }
+}
+
+void UdpPacketSocket::HandleReadResult(int result) {
+  if (result == net::ERR_IO_PENDING) {
+    return;
+  }
+
+  if (result > 0) {
+    talk_base::SocketAddress address;
+    if (!jingle_glue::IPEndPointToSocketAddress(receive_address_, &address)) {
+      NOTREACHED();
+      LOG(ERROR) << "Failed to convert address received from RecvFrom().";
+      return;
+    }
+    SignalReadPacket(this, receive_buffer_->data(), result, address);
+  } else {
+    LOG(ERROR) << "Received error when reading from UDP socket: " << result;
+  }
+}
+
+}  // namespace
+
+ChromiumPacketSocketFactory::ChromiumPacketSocketFactory() {
+}
+
+ChromiumPacketSocketFactory::~ChromiumPacketSocketFactory() {
+}
+
+talk_base::AsyncPacketSocket* ChromiumPacketSocketFactory::CreateUdpSocket(
+      const talk_base::SocketAddress& local_address,
+      int min_port, int max_port) {
+  scoped_ptr<UdpPacketSocket> result(new UdpPacketSocket());
+  if (!result->Init(local_address, min_port, max_port))
+    return NULL;
+  return result.release();
+}
+
+talk_base::AsyncPacketSocket*
+ChromiumPacketSocketFactory::CreateServerTcpSocket(
+    const talk_base::SocketAddress& local_address,
+    int min_port, int max_port,
+    bool ssl) {
+  // We don't use TCP sockets for remoting connections.
+  NOTREACHED();
+  return NULL;
+}
+
+talk_base::AsyncPacketSocket*
+ChromiumPacketSocketFactory::CreateClientTcpSocket(
+      const talk_base::SocketAddress& local_address,
+      const talk_base::SocketAddress& remote_address,
+      const talk_base::ProxyInfo& proxy_info,
+      const std::string& user_agent,
+      bool ssl) {
+  // We don't use TCP sockets for remoting connections.
+  NOTREACHED();
+  return NULL;
+}
+
+}  // namespace remoting
diff --git a/remoting/jingle_glue/chromium_socket_factory.h b/remoting/jingle_glue/chromium_socket_factory.h
new file mode 100644
index 0000000..f142ad2
--- /dev/null
+++ b/remoting/jingle_glue/chromium_socket_factory.h
@@ -0,0 +1,38 @@
+// Copyright (c) 2012 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.
+
+#ifndef REMOTING_JINGLE_GLUE_CHROMIUM_SOCKET_FACTORY_H_
+#define REMOTING_JINGLE_GLUE_CHROMIUM_SOCKET_FACTORY_H_
+
+#include "base/compiler_specific.h"
+#include "third_party/libjingle/source/talk/base/packetsocketfactory.h"
+
+namespace remoting {
+
+class ChromiumPacketSocketFactory : public talk_base::PacketSocketFactory {
+ public:
+  explicit ChromiumPacketSocketFactory();
+  virtual ~ChromiumPacketSocketFactory();
+
+  virtual talk_base::AsyncPacketSocket* CreateUdpSocket(
+      const talk_base::SocketAddress& local_address,
+      int min_port, int max_port) OVERRIDE;
+  virtual talk_base::AsyncPacketSocket* CreateServerTcpSocket(
+      const talk_base::SocketAddress& local_address,
+      int min_port, int max_port,
+      bool ssl) OVERRIDE;
+  virtual talk_base::AsyncPacketSocket* CreateClientTcpSocket(
+      const talk_base::SocketAddress& local_address,
+      const talk_base::SocketAddress& remote_address,
+      const talk_base::ProxyInfo& proxy_info,
+      const std::string& user_agent,
+      bool ssl) OVERRIDE;
+
+ private:
+  DISALLOW_COPY_AND_ASSIGN(ChromiumPacketSocketFactory);
+};
+
+}  // namespace remoting
+
+#endif  // REMOTING_JINGLE_GLUE_CHROMIUM_SOCKET_FACTORY_H_
diff --git a/remoting/jingle_glue/chromium_socket_factory_unittest.cc b/remoting/jingle_glue/chromium_socket_factory_unittest.cc
new file mode 100644
index 0000000..fbd9fce
--- /dev/null
+++ b/remoting/jingle_glue/chromium_socket_factory_unittest.cc
@@ -0,0 +1,94 @@
+// Copyright (c) 2012 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 "remoting/jingle_glue/chromium_socket_factory.h"
+
+#include "base/message_loop.h"
+#include "base/run_loop.h"
+#include "testing/gmock/include/gmock/gmock.h"
+#include "testing/gtest/include/gtest/gtest.h"
+#include "third_party/libjingle/source/talk/base/asyncpacketsocket.h"
+#include "third_party/libjingle/source/talk/base/socketaddress.h"
+
+namespace remoting {
+
+class ChromiumSocketFactoryTest : public testing::Test,
+                                  public sigslot::has_slots<> {
+ public:
+  virtual void SetUp() OVERRIDE {
+    socket_factory_.reset(new ChromiumPacketSocketFactory());
+
+    socket_.reset(socket_factory_->CreateUdpSocket(
+        talk_base::SocketAddress("127.0.0.1", 0), 0, 0));
+    ASSERT_TRUE(socket_.get() != NULL);
+    EXPECT_EQ(socket_->GetState(), talk_base::AsyncPacketSocket::STATE_BOUND);
+    socket_->SignalReadPacket.connect(
+        this, &ChromiumSocketFactoryTest::OnPacket);
+  }
+
+  void OnPacket(talk_base::AsyncPacketSocket* socket,
+                const char* data, size_t size,
+                const talk_base::SocketAddress& address) {
+    EXPECT_EQ(socket, socket_.get());
+    last_packet_.assign(data, data + size);
+    last_address_ = address;
+    run_loop_.Quit();
+  }
+
+ protected:
+  MessageLoopForIO message_loop_;
+  base::RunLoop run_loop_;
+
+  scoped_ptr<talk_base::PacketSocketFactory> socket_factory_;
+  scoped_ptr<talk_base::AsyncPacketSocket> socket_;
+
+  std::string last_packet_;
+  talk_base::SocketAddress last_address_;
+};
+
+TEST_F(ChromiumSocketFactoryTest, SendAndReceive) {
+  // UDP packets may be lost, so we have to retry sending it more than once.
+  const int kMaxAttempts = 3;
+  const base::TimeDelta kAttemptPeriod = base::TimeDelta::FromSeconds(1);
+
+  scoped_ptr<talk_base::AsyncPacketSocket> sending_socket;
+  talk_base::SocketAddress address;
+
+  sending_socket.reset(socket_factory_->CreateUdpSocket(
+      talk_base::SocketAddress("127.0.0.1", 0), 0, 0));
+  ASSERT_TRUE(sending_socket.get() != NULL);
+  EXPECT_EQ(sending_socket->GetState(),
+            talk_base::AsyncPacketSocket::STATE_BOUND);
+  address = sending_socket->GetLocalAddress();
+
+  std::string test_packet("TEST PACKET");
+  int attempts = 0;
+  while (last_packet_.empty() && attempts++ < kMaxAttempts) {
+    sending_socket->SendTo(test_packet.data(), test_packet.size(),
+                           socket_->GetLocalAddress());
+    message_loop_.PostDelayedTask(FROM_HERE, run_loop_.QuitClosure(),
+                                  kAttemptPeriod);
+    run_loop_.Run();
+  }
+  EXPECT_EQ(test_packet, last_packet_);
+  EXPECT_EQ(address, last_address_);
+}
+
+TEST_F(ChromiumSocketFactoryTest, SetOptions) {
+  EXPECT_EQ(0, socket_->SetOption(talk_base::Socket::OPT_SNDBUF, 4096));
+  EXPECT_EQ(0, socket_->SetOption(talk_base::Socket::OPT_RCVBUF, 4096));
+}
+
+TEST_F(ChromiumSocketFactoryTest, PortRange) {
+  const int kMinPort = 12400;
+  const int kMaxPort = 12410;
+  socket_.reset(socket_factory_->CreateUdpSocket(
+      talk_base::SocketAddress("127.0.0.1", 0), kMaxPort, kMaxPort));
+  ASSERT_TRUE(socket_.get() != NULL);
+  EXPECT_EQ(socket_->GetState(), talk_base::AsyncPacketSocket::STATE_BOUND);
+  EXPECT_GE(socket_->GetLocalAddress().port(), kMinPort);
+  EXPECT_LE(socket_->GetLocalAddress().port(), kMaxPort);
+}
+
+}  // namespace remoting
diff --git a/remoting/remoting.gyp b/remoting/remoting.gyp
index f3ca639..feaebad 100644
--- a/remoting/remoting.gyp
+++ b/remoting/remoting.gyp
@@ -1480,6 +1480,8 @@
         '../third_party/libjingle/libjingle.gyp:libjingle_p2p',
       ],
       'sources': [
+        'jingle_glue/chromium_socket_factory.cc',
+        'jingle_glue/chromium_socket_factory.h',
         'jingle_glue/iq_sender.cc',
         'jingle_glue/iq_sender.h',
         'jingle_glue/javascript_signal_strategy.cc',
@@ -1706,6 +1708,7 @@
         'host/video_frame_capturer_helper_unittest.cc',
         'host/video_frame_capturer_mac_unittest.cc',
         'host/video_frame_capturer_unittest.cc',
+        'jingle_glue/chromium_socket_factory_unittest.cc',
         'jingle_glue/fake_signal_strategy.cc',
         'jingle_glue/fake_signal_strategy.h',
         'jingle_glue/iq_sender_unittest.cc',