Implement ChromiumSocketFactory.

remoting/jingle_glue/chromium_socket_factory.cc

+// 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

[Wez, 2012/07/19 00:40:20]

Your line-wrap seems to be at ten characters short?

[Sergey Ulanov, 2012/07/19 20:16:39]

Emacs wraps comments at 70 chars (by default and also with the standard settings
file for Google's C++ style). IMHO it's more readable that way.

+// 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:
+  explicit UdpPacketSocket();

[Wez, 2012/07/19 00:40:20]

No need for explicit?

[Sergey Ulanov, 2012/07/19 20:16:39]

Done.

+  virtual ~UdpPacketSocket();
+
+  // |min_port| and |max_port| are set to zero if the port number
+  // |should be assigned by the OS.

[Wez, 2012/07/19 00:40:20]

typo: |should

[Wez, 2012/07/19 00:40:20]

nit: Suggest "|min_port| and |max_port| limit the range of ports the socket may
use. If both are zero then the OS assigns a port number."

[Sergey Ulanov, 2012/07/19 20:16:39]

Done.

[Sergey Ulanov, 2012/07/19 20:16:39]

Just removed this comment. meaning of these values is defined by the
PacketSocketFactory interface.

+  bool Init(const talk_base::SocketAddress& local_address,

[Wez, 2012/07/19 00:40:20]

nit: Does local_address accept the usual INADDR_ANY, etc?

[Sergey Ulanov, 2012/07/19 20:16:39]

Yes, but it's never used that way.

+            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, bool* read_again);
+
+  scoped_ptr<net::UDPServerSocket> socket_;
+
+  State state_;
+  int error_;
+
+  talk_base::SocketAddress local_address_;
+
+  scoped_refptr<net::IOBuffer> receive_buffer_;

[Wez, 2012/07/19 00:40:20]

nit: Add a comment explaining why we need these in the class - because
ReadFrom() completes asynchronously and we have to pass it the buffer &
IPEndPoint to populate?

[Sergey Ulanov, 2012/07/19 20:16:39]

Done.

+  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()));
+    net::IPEndPoint address_with_port(local_endpoint.address(), port);

[Wez, 2012/07/19 00:40:20]

nit: endpoint_with_port

[Sergey Ulanov, 2012/07/19 20:16:39]

removed this variable.

+    int result = socket_->Listen(address_with_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) {
+    // TODO(sergeyu): StunPort may try to send stun request before we
+    // are bound. Fix that problem and change this to DCHECK.

[Wez, 2012/07/19 00:40:20]

nit: OS socket APIs will typically return EINVAL rather than crashing the
caller, in this case, so are you sure you prefer a DCHECK? It'll make us fragile
to changes in libjingle that implicitly assume the EINVAL behaviour.

[Wez, 2012/07/19 00:40:20]

Doesn't that mean before Init() has been called? How can that happen?

[Sergey Ulanov, 2012/07/19 20:16:39]

I copied that TODO from PepperPacketSocketFactory, but it's not relevant here.
With the pepper interface bind() is asynchronous but current StunPort doesn't
handle asynchronous bind properly. DCHECK would be helpful to detect such bugs.
Anyway removed this TODO and added NOTREACHED() here.

+    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();

[Wez, 2012/07/19 00:40:20]

nit: Blank line here, for consistency with Init()

[Sergey Ulanov, 2012/07/19 20:16:39]

Done.

+  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 0;

[Wez, 2012/07/19 00:40:20]

nit: return -1; ?

[Sergey Ulanov, 2012/07/19 20:16:39]

Done.

+}
+
+int UdpPacketSocket::SetOption(talk_base::Socket::Option option, int value) {
+  DCHECK(socket_.get());

[Wez, 2012/07/19 00:40:20]

Shouldn't this be an EINVAL return akin to Send()?

[Sergey Ulanov, 2012/07/19 20:16:39]

Done.

+  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;

[Wez, 2012/07/19 00:40:20]

nit: Should we SetError() in case of failure?

[Sergey Ulanov, 2012/07/19 20:16:39]

PhysicalSocketServer doesn't do it.

+    }
+
+    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;
+
+  int result = socket_->SendTo(
+      send_queue_.front().data, send_queue_.front().data->size(),

[Wez, 2012/07/19 00:40:20]

nit: Ick. Pull out send_queue.front() as a local reference variable to make this
readable.

[Sergey Ulanov, 2012/07/19 20:16:39]

Done.

+      send_queue_.front().address,
+      base::Bind(&UdpPacketSocket::OnSendCompleted,
+                 base::Unretained(this)));

[Wez, 2012/07/19 00:40:20]

off-topic: Whenever I see this it frightens me because I forget that
net::UdpSocket drops all callbacks when deleted - I wish there were a way to
document that a use of base::Unretained is safe, as opposed to all the broken
usages we have elsewhere...

[Sergey Ulanov, 2012/07/19 20:16:39]

I agree.

+  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
+    // errors.

[Wez, 2012/07/19 00:40:20]

nit: No need for the second use of "errors".

[Sergey Ulanov, 2012/07/19 20:16:39]

Done.

+    if (result != net::ERR_ADDRESS_UNREACHABLE) {
+      LOG(ERROR) << "Send failed on a UDP socket: " << result;
+      error_ = EINVAL;
+      return;
+    }
+  }
+
+  send_queue_size_ -= send_queue_.front().data->size();

[Wez, 2012/07/19 00:40:20]

nit: Clarify that we don't need to worry about partial sends since this is a
datagram socket.

[Sergey Ulanov, 2012/07/19 20:16:39]

Done.

+  send_queue_.pop_front();
+  DoSend();
+}
+
+void UdpPacketSocket::DoRead() {
+  bool read_again = true;
+  while (read_again) {
+    receive_buffer_ = new net::IOBuffer(kReceiveBufferSize);
+    int result = socket_->RecvFrom(
+        receive_buffer_, kReceiveBufferSize, &receive_address_,
+        base::Bind(&UdpPacketSocket::OnReadCompleted, base::Unretained(this)));
+    HandleReadResult(result, &read_again);

[Wez, 2012/07/19 00:40:20]

It seems cleaner to check whether result > 0 after HandleReadResult and break if
not, rather than keep a read_again flag.

[Sergey Ulanov, 2012/07/19 20:16:39]

Done.

+  }
+}
+
+void UdpPacketSocket::OnReadCompleted(int result) {
+  bool read_again;
+  HandleReadResult(result, &read_again);
+  if (read_again) {
+    DoRead();
+  }
+}
+
+void UdpPacketSocket::HandleReadResult(int result, bool* read_again) {

[Wez, 2012/07/19 00:40:20]

Why not have read_again be the return value of HandleReadResult()?

[Sergey Ulanov, 2012/07/19 20:16:39]

It's more readable with the parameter instead of return value. Anyway I removed
it now.

+  if (result == net::ERR_IO_PENDING) {
+    *read_again = false;
+    return;
+  }
+
+  if (result > 0) {
+    talk_base::SocketAddress address;
+    if (!jingle_glue::IPEndPointToSocketAddress(receive_address_, &address)) {
+      LOG(ERROR) << "Failed to covert address received from RecvFrom().";

[Wez, 2012/07/19 00:40:20]

I think you want to read again even if the peer address didn't convert for some
reason (although it always will...)?

[Wez, 2012/07/19 00:40:20]

typo: covert

[Sergey Ulanov, 2012/07/19 20:16:39]

Done.

[Sergey Ulanov, 2012/07/19 20:16:39]

Done.

+      return;
+    }
+    SignalReadPacket(this, receive_buffer_->data(), result, address);
+    *read_again = true;
+  } else {
+    LOG(ERROR) << "Received error when reading from UDP socket: " << result;
+    *read_again = false;
+  }
+}
+
+}  // 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