Mojo UDP API implementation.

mojo/services/network/udp_socket_impl.cc

Index: mojo/services/network/udp_socket_impl.cc
diff --git a/mojo/services/network/udp_socket_impl.cc b/mojo/services/network/udp_socket_impl.cc
new file mode 100644
index 0000000000000000000000000000000000000000..d1457e96888034517159d785267bb87ac558c62f
--- /dev/null
+++ b/mojo/services/network/udp_socket_impl.cc
@@ -0,0 +1,254 @@
+// Copyright 2014 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 "mojo/services/network/udp_socket_impl.h"
+
+#include <string.h>
+
+#include <limits>
+
+#include "base/logging.h"
+#include "base/memory/scoped_ptr.h"
+#include "mojo/services/network/net_address_type_converters.h"
+#include "net/base/io_buffer.h"
+#include "net/base/net_errors.h"
+
+namespace mojo {
+
+namespace {
+
+const int kMaxReadSize = 128 * 1024;
+const size_t kMaxPendingSendRequests = 32;
+
+NetworkErrorPtr MakeNetworkError(int error_code) {
+  NetworkErrorPtr error(NetworkError::New());
+  error->code = error_code;
+  if (error_code <= 0)
+    error->description = net::ErrorToString(error_code);
+  return error.Pass();
+}
+
+}  // namespace
+
+UDPSocketImpl::PendingSendRequest::PendingSendRequest() {}
+
+UDPSocketImpl::PendingSendRequest::~PendingSendRequest() {}
+
+UDPSocketImpl::UDPSocketImpl() : socket_(NULL, net::NetLog::Source()),
+                                 bound_(false),
+                                 remaining_recv_slots_(0) {
+}
+
+UDPSocketImpl::~UDPSocketImpl() {
+  while (!pending_send_requests_.empty()) {
+    delete pending_send_requests_.front();
+    pending_send_requests_.pop();
+  }
+}
+
+void UDPSocketImpl::SetSendBufferSize(
+    uint32_t size,
+    const Callback<void(NetworkErrorPtr)>& callback) {
+  if (!bound_) {
+    callback.Run(MakeNetworkError(net::ERR_FAILED));
+    return;
+  }
+
+  if (size > static_cast<uint32_t>(std::numeric_limits<int32_t>::max()))
+    size = std::numeric_limits<int32_t>::max();
+
+  int net_result = socket_.SetSendBufferSize(static_cast<int32_t>(size));
+  callback.Run(MakeNetworkError(net_result));
+}
+
+void UDPSocketImpl::SetReceiveBufferSize(
+    uint32_t size,
+    const Callback<void(NetworkErrorPtr)>& callback) {
+  if (!bound_) {
+    callback.Run(MakeNetworkError(net::ERR_FAILED));
+    return;
+  }
+
+  if (size > static_cast<uint32_t>(std::numeric_limits<int32_t>::max()))
+    size = std::numeric_limits<int32_t>::max();
+
+  int net_result = socket_.SetReceiveBufferSize(static_cast<int32_t>(size));
+  callback.Run(MakeNetworkError(net_result));
+}
+
+void UDPSocketImpl::AllowAddressReuse(
+    const Callback<void(NetworkErrorPtr)>& callback) {
+  if (bound_) {
+    callback.Run(MakeNetworkError(net::ERR_FAILED));
+    return;
+  }
+
+  socket_.AllowAddressReuse();
+  callback.Run(MakeNetworkError(net::OK));
+}
+
+void UDPSocketImpl::Bind(
+    NetAddressPtr addr,
+    const Callback<void(NetworkErrorPtr, NetAddressPtr)>& callback) {
+  if (bound_) {
+    callback.Run(MakeNetworkError(net::ERR_FAILED), NetAddressPtr());
+    return;
+  }
+
+  net::IPEndPoint ip_end_point = addr.To<net::IPEndPoint>();
+  if (ip_end_point.GetFamily() == net::ADDRESS_FAMILY_UNSPECIFIED) {
+    callback.Run(MakeNetworkError(net::ERR_ADDRESS_INVALID), NetAddressPtr());
+    return;
+  }
+
+  int net_result = socket_.Listen(ip_end_point);
+  if (net_result != net::OK) {
+    callback.Run(MakeNetworkError(net_result), NetAddressPtr());
+    return;
+  }
+
+  net::IPEndPoint bound_ip_end_point;
+  NetAddressPtr bound_addr;
+  net_result = socket_.GetLocalAddress(&bound_ip_end_point);
+  if (net_result == net::OK)
+    bound_addr = NetAddress::From(bound_ip_end_point);
+
+  bound_ = true;
+  callback.Run(MakeNetworkError(net::OK), bound_addr.Pass());
+}
+
+void UDPSocketImpl::ReceiveMorePackets(uint32_t number) {
+  if (!bound_ || number == 0)
+    return;
+  if (std::numeric_limits<size_t>::max() - remaining_recv_slots_ < number)
+    return;
+
+  remaining_recv_slots_ += number;
+
+  if (!recvfrom_buffer_.get()) {
+    DCHECK_EQ(number, remaining_recv_slots_);
+    DoRecvFrom();
+  }
+}
+
+void UDPSocketImpl::SendToAndForget(NetAddressPtr addr, Array<uint8_t> data) {
+  SendTo(addr.Pass(), data.Pass(), Callback<void(NetworkErrorPtr)>());
+}
+
+void UDPSocketImpl::SendTo(NetAddressPtr addr,
+                           Array<uint8_t> data,
+                           const Callback<void(NetworkErrorPtr)>& callback) {
+  if (!bound_) {
+    callback.Run(MakeNetworkError(net::ERR_FAILED));
+    return;
+  }
+
+  if (sendto_buffer_.get()) {
+    if (pending_send_requests_.size() >= kMaxPendingSendRequests) {
+      callback.Run(MakeNetworkError(net::ERR_INSUFFICIENT_RESOURCES));
+      return;
+    }
+
+    PendingSendRequest* request = new PendingSendRequest;
+    request->addr = addr.Pass();
+    request->data = data.Pass();
+    request->callback = callback;
+    pending_send_requests_.push(request);
+    return;
+  }
+
+  DCHECK_EQ(0u, pending_send_requests_.size());
+
+  DoSendTo(addr.Pass(), data.Pass(), callback);
+}
+
+void UDPSocketImpl::DoRecvFrom() {
+  DCHECK(bound_);
+  DCHECK(!recvfrom_buffer_.get());
+  DCHECK_GT(remaining_recv_slots_, 0u);
+
+  recvfrom_buffer_ = new net::IOBuffer(kMaxReadSize);
+
+  // It is safe to use base::Unretained(this) because |socket_| is owned by this
+  // object. If this object gets destroyed (and so does |socket_|), the callback
+  // won't be called.
+  int net_result = socket_.RecvFrom(
+      recvfrom_buffer_.get(),
+      kMaxReadSize,
+      &recvfrom_address_,
+      base::Bind(&UDPSocketImpl::OnRecvFromCompleted, base::Unretained(this)));
+  if (net_result != net::ERR_IO_PENDING)
+    OnRecvFromCompleted(net_result);
+}
+
+void UDPSocketImpl::DoSendTo(NetAddressPtr addr,
+                             Array<uint8_t> data,
+                             const Callback<void(NetworkErrorPtr)>& callback) {
+  DCHECK(bound_);
+  DCHECK(!sendto_buffer_.get());
+
+  net::IPEndPoint ip_end_point = addr.To<net::IPEndPoint>();
+  if (ip_end_point.GetFamily() == net::ADDRESS_FAMILY_UNSPECIFIED) {
+    callback.Run(MakeNetworkError(net::ERR_ADDRESS_INVALID));
+    return;
+  }
+
+  sendto_buffer_ = new net::IOBufferWithSize(data.size());
+  if (data.size() > 0)
+    memcpy(sendto_buffer_->data(), &data.storage()[0], data.size());
+
+  // It is safe to use base::Unretained(this) because |socket_| is owned by this
+  // object. If this object gets destroyed (and so does |socket_|), the callback
+  // won't be called.
+  int net_result = socket_.SendTo(sendto_buffer_.get(), sendto_buffer_->size(),
+                                  ip_end_point,
+                                  base::Bind(&UDPSocketImpl::OnSendToCompleted,
+                                             base::Unretained(this), callback));
+  if (net_result != net::ERR_IO_PENDING)
+    OnSendToCompleted(callback, net_result);
+}
+
+void UDPSocketImpl::OnRecvFromCompleted(int net_result) {
+  DCHECK(recvfrom_buffer_.get());
+
+  NetAddressPtr net_address;
+  Array<uint8_t> array;
+  if (net_result >= 0) {
+    net_address = NetAddress::From(recvfrom_address_);
+    std::vector<uint8_t> data(net_result);
+    if (net_result > 0)
+      memcpy(&data[0], recvfrom_buffer_->data(), net_result);
+
+    array.Swap(&data);
+  }
+  recvfrom_buffer_ = NULL;
+
+  client()->OnReceived(MakeNetworkError(net_result), net_address.Pass(),
+                       array.Pass());
+
+  DCHECK_GT(remaining_recv_slots_, 0u);
+  remaining_recv_slots_--;
+  if (remaining_recv_slots_ > 0)
+    DoRecvFrom();
+}
+
+void UDPSocketImpl::OnSendToCompleted(
+    const Callback<void(NetworkErrorPtr)>& callback,
+    int net_result) {
+  DCHECK(sendto_buffer_.get());
+
+  sendto_buffer_ = NULL;
+
+  callback.Run(MakeNetworkError(net_result));
+
+  if (pending_send_requests_.empty())
+    return;
+
+  scoped_ptr<PendingSendRequest> request(pending_send_requests_.front());
+  pending_send_requests_.pop();
+
+  DoSendTo(request->addr.Pass(), request->data.Pass(), request->callback);
+}
+
+}  // namespace mojo