mojo/services/network/public/cpp/udp_socket_wrapper.cc

   1 // Copyright 2014 The Chromium Authors. All rights reserved.
   2 // Use of this source code is governed by a BSD-style license that can be
   3 // found in the LICENSE file.
   4
   5 #include "network/public/cpp/udp_socket_wrapper.h"
   6
   7 #include <assert.h>
   8
   9 #include "third_party/mojo/src/mojo/public/cpp/environment/logging.h"
  10
  11 namespace mojo {
  12 namespace {
  13
  14 const uint32_t kDefaultReceiveQueueSlots = 32;
  15
  16 }  // namespace
  17
  18 UDPSocketWrapper::NegotiateCallbackHandler::NegotiateCallbackHandler(
  19     UDPSocketWrapper* delegate)
  20     : delegate_(delegate) {
  21 }
  22
  23 UDPSocketWrapper::NegotiateCallbackHandler::~NegotiateCallbackHandler() {}
  24
  25 void UDPSocketWrapper::NegotiateCallbackHandler::Run(
  26     uint32_t actual_size) const {
  27   delegate_->OnNegotiateMaxPendingSendRequestsCompleted(actual_size);
  28 }
  29
  30 UDPSocketWrapper::SendCallbackHandler::SendCallbackHandler(
  31     UDPSocketWrapper* delegate,
  32     const ErrorCallback& forward_callback)
  33     : delegate_(delegate),
  34       forward_callback_(forward_callback) {
  35 }
  36
  37 UDPSocketWrapper::SendCallbackHandler::~SendCallbackHandler() {}
  38
  39 void UDPSocketWrapper::SendCallbackHandler::Run(NetworkErrorPtr result) const {
  40   delegate_->OnSendToCompleted(result.Pass(), forward_callback_);
  41 }
  42
  43 UDPSocketWrapper::ReceiverBindingCallback::ReceiverBindingCallback(
  44     UDPSocketWrapper* delegate,
  45     const Callback<void(NetworkErrorPtr, NetAddressPtr)>& wrapper_callback)
  46     : delegate_(delegate), wrapper_callback_(wrapper_callback) {
  47 }
  48
  49 UDPSocketWrapper::ReceiverBindingCallback::~ReceiverBindingCallback() {
  50 }
  51
  52 void UDPSocketWrapper::ReceiverBindingCallback::Run(
  53     NetworkErrorPtr result,
  54     NetAddressPtr addr,
  55     InterfaceRequest<UDPSocketReceiver> request) const {
  56   delegate_->StartReceivingData(request.Pass());
  57   wrapper_callback_.Run(result.Pass(), addr.Pass());
  58 }
  59
  60 UDPSocketWrapper::ReceivedData::ReceivedData() {}
  61 UDPSocketWrapper::ReceivedData::~ReceivedData() {}
  62
  63 UDPSocketWrapper::SendRequest::SendRequest() {}
  64 UDPSocketWrapper::SendRequest::~SendRequest() {}
  65
  66 UDPSocketWrapper::UDPSocketWrapper(UDPSocketPtr socket)
  67     : binding_(this),
  68       socket_(socket.Pass()),
  69       max_receive_queue_size_(kDefaultReceiveQueueSlots),
  70       max_pending_sends_(1),
  71       current_pending_sends_(0) {
  72   Initialize(0);
  73 }
  74
  75 UDPSocketWrapper::UDPSocketWrapper(UDPSocketPtr socket,
  76                                    uint32_t receive_queue_slots,
  77                                    uint32_t requested_max_pending_sends)
  78     : binding_(this),
  79       socket_(socket.Pass()),
  80       max_receive_queue_size_(receive_queue_slots),
  81       max_pending_sends_(1),
  82       current_pending_sends_(0) {
  83   Initialize(requested_max_pending_sends);
  84 }
  85
  86 UDPSocketWrapper::~UDPSocketWrapper() {
  87   while (!receive_queue_.empty()) {
  88     delete receive_queue_.front();
  89     receive_queue_.pop();
  90   }
  91   while (!send_requests_.empty()) {
  92     delete send_requests_.front();
  93     send_requests_.pop();
  94   }
  95 }
  96
  97 void UDPSocketWrapper::AllowAddressReuse(const ErrorCallback& callback) {
  98   socket_->AllowAddressReuse(callback);
  99 }
 100
 101 void UDPSocketWrapper::Bind(
 102     NetAddressPtr addr,
 103     const Callback<void(NetworkErrorPtr, NetAddressPtr)>& callback) {
 104   socket_->Bind(
 105       addr.Pass(),
 106       BindOrConnectCallback(static_cast<BindOrConnectCallback::Runnable*>(
 107           new ReceiverBindingCallback(this, callback))));
 108 }
 109
 110 void UDPSocketWrapper::Connect(
 111     NetAddressPtr remote_addr,
 112     const Callback<void(NetworkErrorPtr, NetAddressPtr)>& callback) {
 113   socket_->Connect(
 114       remote_addr.Pass(),
 115       BindOrConnectCallback(static_cast<BindOrConnectCallback::Runnable*>(
 116           new ReceiverBindingCallback(this, callback))));
 117 }
 118
 119 void UDPSocketWrapper::SetSendBufferSize(uint32_t size,
 120                                          const ErrorCallback& callback) {
 121   socket_->SetSendBufferSize(size, callback);
 122 }
 123
 124 void UDPSocketWrapper::SetReceiveBufferSize(uint32_t size,
 125                                             const ErrorCallback& callback) {
 126   socket_->SetReceiveBufferSize(size, callback);
 127 }
 128
 129 bool UDPSocketWrapper::ReceiveFrom(const ReceiveCallback& callback) {
 130   if (receive_queue_.empty()) {
 131     receive_requests_.push(callback);
 132     return false;
 133   }
 134
 135   ReceivedData* data = receive_queue_.front();
 136   receive_queue_.pop();
 137   socket_->ReceiveMore(1);
 138   callback.Run(data->result.Pass(), data->src_addr.Pass(), data->data.Pass());
 139   delete data;
 140   return true;
 141 }
 142
 143 void UDPSocketWrapper::SendTo(NetAddressPtr dest_addr,
 144                               Array<uint8_t> data,
 145                               const ErrorCallback& callback) {
 146   if (current_pending_sends_ >= max_pending_sends_) {
 147     SendRequest* request = new SendRequest();
 148     request->dest_addr = dest_addr.Pass();
 149     request->data = data.Pass();
 150     request->callback = callback;
 151     send_requests_.push(request);
 152     return;
 153   }
 154
 155   MOJO_DCHECK(send_requests_.empty());
 156   current_pending_sends_++;
 157   socket_->SendTo(dest_addr.Pass(), data.Pass(),
 158                   ErrorCallback(static_cast<ErrorCallback::Runnable*>(
 159                       new SendCallbackHandler(this, callback))));
 160 }
 161
 162 void UDPSocketWrapper::OnReceived(NetworkErrorPtr result,
 163                                   NetAddressPtr src_addr,
 164                                   Array<uint8_t> data) {
 165   if (!receive_requests_.empty()) {
 166     // The cache should be empty if there are user requests waiting for data.
 167     MOJO_DCHECK(receive_queue_.empty());
 168
 169     socket_->ReceiveMore(1);
 170
 171     ReceiveCallback callback = receive_requests_.front();
 172     receive_requests_.pop();
 173
 174     callback.Run(result.Pass(), src_addr.Pass(), data.Pass());
 175     return;
 176   }
 177
 178   MOJO_DCHECK(receive_queue_.size() < max_receive_queue_size_);
 179   ReceivedData* received_data = new ReceivedData();
 180   received_data->result = result.Pass();
 181   received_data->src_addr = src_addr.Pass();
 182   received_data->data = data.Pass();
 183   receive_queue_.push(received_data);
 184 }
 185
 186 void UDPSocketWrapper::Initialize(uint32_t requested_max_pending_sends) {
 187   socket_->NegotiateMaxPendingSendRequests(
 188       requested_max_pending_sends,
 189       Callback<void(uint32_t)>(
 190           static_cast< Callback<void(uint32_t)>::Runnable*>(
 191               new NegotiateCallbackHandler(this))));
 192 }
 193
 194 void UDPSocketWrapper::OnNegotiateMaxPendingSendRequestsCompleted(
 195     uint32_t actual_size) {
 196   MOJO_DCHECK(max_pending_sends_ == 1);
 197
 198   if (actual_size == 0) {
 199     assert(false);
 200     return;
 201   }
 202
 203   max_pending_sends_ = actual_size;
 204
 205   while (ProcessNextSendRequest());
 206 }
 207
 208 void UDPSocketWrapper::OnSendToCompleted(
 209     NetworkErrorPtr result,
 210     const ErrorCallback& forward_callback) {
 211   current_pending_sends_--;
 212   ProcessNextSendRequest();
 213
 214   forward_callback.Run(result.Pass());
 215 }
 216
 217 bool UDPSocketWrapper::ProcessNextSendRequest() {
 218   if (current_pending_sends_ >= max_pending_sends_ || send_requests_.empty())
 219     return false;
 220
 221   SendRequest* request = send_requests_.front();
 222   send_requests_.pop();
 223
 224   current_pending_sends_++;
 225
 226   socket_->SendTo(
 227       request->dest_addr.Pass(), request->data.Pass(),
 228       ErrorCallback(static_cast<ErrorCallback::Runnable*>(
 229           new SendCallbackHandler(this, request->callback))));
 230
 231   delete request;
 232
 233   return true;
 234 }
 235
 236 void UDPSocketWrapper::StartReceivingData(
 237     InterfaceRequest<UDPSocketReceiver> request) {
 238   binding_.Bind(request.Pass());
 239   socket_->ReceiveMore(max_receive_queue_size_);
 240 }
 241
 242 }  // namespace mojo