remoting/protocol/channel_socket_adapter.cc

   1 // Copyright 2015 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 "remoting/protocol/channel_socket_adapter.h"
   6
   7 #include <limits>
   8
   9 #include "base/callback.h"
  10 #include "base/logging.h"
  11 #include "net/base/io_buffer.h"
  12 #include "net/base/net_errors.h"
  13 #include "third_party/webrtc/p2p/base/transportchannel.h"
  14
  15 namespace remoting {
  16 namespace protocol {
  17
  18 TransportChannelSocketAdapter::TransportChannelSocketAdapter(
  19     cricket::TransportChannel* channel)
  20     : channel_(channel),
  21       closed_error_code_(net::OK) {
  22   DCHECK(channel_);
  23
  24   channel_->SignalReadPacket.connect(
  25       this, &TransportChannelSocketAdapter::OnNewPacket);
  26   channel_->SignalWritableState.connect(
  27       this, &TransportChannelSocketAdapter::OnWritableState);
  28   channel_->SignalDestroyed.connect(
  29       this, &TransportChannelSocketAdapter::OnChannelDestroyed);
  30 }
  31
  32 TransportChannelSocketAdapter::~TransportChannelSocketAdapter() {
  33   if (!destruction_callback_.is_null())
  34     destruction_callback_.Run();
  35 }
  36
  37 void TransportChannelSocketAdapter::SetOnDestroyedCallback(
  38     const base::Closure& callback) {
  39   destruction_callback_ = callback;
  40 }
  41
  42 int TransportChannelSocketAdapter::Recv(
  43     const scoped_refptr<net::IOBuffer>& buf, int buffer_size,
  44     const net::CompletionCallback& callback) {
  45   DCHECK(thread_checker_.CalledOnValidThread());
  46   DCHECK(buf);
  47   DCHECK(!callback.is_null());
  48   CHECK(read_callback_.is_null());
  49
  50   if (!channel_) {
  51     DCHECK(closed_error_code_ != net::OK);
  52     return closed_error_code_;
  53   }
  54
  55   read_callback_ = callback;
  56   read_buffer_ = buf;
  57   read_buffer_size_ = buffer_size;
  58
  59   return net::ERR_IO_PENDING;
  60 }
  61
  62 int TransportChannelSocketAdapter::Send(
  63     const scoped_refptr<net::IOBuffer>& buffer, int buffer_size,
  64     const net::CompletionCallback& callback) {
  65   DCHECK(thread_checker_.CalledOnValidThread());
  66   DCHECK(buffer);
  67   DCHECK(!callback.is_null());
  68   CHECK(write_callback_.is_null());
  69
  70   if (!channel_) {
  71     DCHECK(closed_error_code_ != net::OK);
  72     return closed_error_code_;
  73   }
  74
  75   int result;
  76   rtc::PacketOptions options;
  77   if (channel_->writable()) {
  78     result = channel_->SendPacket(buffer->data(), buffer_size, options);
  79     if (result < 0) {
  80       result = net::MapSystemError(channel_->GetError());
  81
  82       // If the underlying socket returns IO pending where it shouldn't we
  83       // pretend the packet is dropped and return as succeeded because no
  84       // writeable callback will happen.
  85       if (result == net::ERR_IO_PENDING)
  86         result = net::OK;
  87     }
  88   } else {
  89     // Channel is not writable yet.
  90     result = net::ERR_IO_PENDING;
  91     write_callback_ = callback;
  92     write_buffer_ = buffer;
  93     write_buffer_size_ = buffer_size;
  94   }
  95
  96   return result;
  97 }
  98
  99 void TransportChannelSocketAdapter::Close(int error_code) {
 100   DCHECK(thread_checker_.CalledOnValidThread());
 101
 102   if (!channel_)  // Already closed.
 103     return;
 104
 105   DCHECK(error_code != net::OK);
 106   closed_error_code_ = error_code;
 107   channel_->SignalReadPacket.disconnect(this);
 108   channel_->SignalDestroyed.disconnect(this);
 109   channel_ = NULL;
 110
 111   if (!read_callback_.is_null()) {
 112     net::CompletionCallback callback = read_callback_;
 113     read_callback_.Reset();
 114     read_buffer_ = NULL;
 115     callback.Run(error_code);
 116   }
 117
 118   if (!write_callback_.is_null()) {
 119     net::CompletionCallback callback = write_callback_;
 120     write_callback_.Reset();
 121     write_buffer_ = NULL;
 122     callback.Run(error_code);
 123   }
 124 }
 125
 126 void TransportChannelSocketAdapter::OnNewPacket(
 127     cricket::TransportChannel* channel,
 128     const char* data,
 129     size_t data_size,
 130     const rtc::PacketTime& packet_time,
 131     int flags) {
 132   DCHECK(thread_checker_.CalledOnValidThread());
 133   DCHECK_EQ(channel, channel_);
 134   if (!read_callback_.is_null()) {
 135     DCHECK(read_buffer_.get());
 136     CHECK_LT(data_size, static_cast<size_t>(std::numeric_limits<int>::max()));
 137
 138     if (read_buffer_size_ < static_cast<int>(data_size)) {
 139       LOG(WARNING) << "Data buffer is smaller than the received packet. "
 140                    << "Dropping the data that doesn't fit.";
 141       data_size = read_buffer_size_;
 142     }
 143
 144     memcpy(read_buffer_->data(), data, data_size);
 145
 146     net::CompletionCallback callback = read_callback_;
 147     read_callback_.Reset();
 148     read_buffer_ = NULL;
 149
 150     callback.Run(data_size);
 151   } else {
 152     LOG(WARNING)
 153         << "Data was received without a callback. Dropping the packet.";
 154   }
 155 }
 156
 157 void TransportChannelSocketAdapter::OnWritableState(
 158     cricket::TransportChannel* channel) {
 159   DCHECK(thread_checker_.CalledOnValidThread());
 160   // Try to send the packet if there is a pending write.
 161   if (!write_callback_.is_null()) {
 162     rtc::PacketOptions options;
 163     int result = channel_->SendPacket(write_buffer_->data(),
 164                                       write_buffer_size_,
 165                                       options);
 166     if (result < 0)
 167       result = net::MapSystemError(channel_->GetError());
 168
 169     if (result != net::ERR_IO_PENDING) {
 170       net::CompletionCallback callback = write_callback_;
 171       write_callback_.Reset();
 172       write_buffer_ = NULL;
 173       callback.Run(result);
 174     }
 175   }
 176 }
 177
 178 void TransportChannelSocketAdapter::OnChannelDestroyed(
 179     cricket::TransportChannel* channel) {
 180   DCHECK(thread_checker_.CalledOnValidThread());
 181   DCHECK_EQ(channel, channel_);
 182   Close(net::ERR_CONNECTION_ABORTED);
 183 }
 184
 185 }  // namespace protocol
 186 }  // namespace remoting