chrome/browser/devtools/device/android_web_socket.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 "base/location.h"
   6 #include "base/memory/weak_ptr.h"
   7 #include "base/rand_util.h"
   8 #include "base/single_thread_task_runner.h"
   9 #include "base/thread_task_runner_handle.h"
  10 #include "chrome/browser/devtools/device/android_device_manager.h"
  11 #include "content/public/browser/browser_thread.h"
  12 #include "net/base/io_buffer.h"
  13 #include "net/base/net_errors.h"
  14 #include "net/server/web_socket_encoder.h"
  15 #include "net/socket/stream_socket.h"
  16
  17 using content::BrowserThread;
  18 using net::WebSocket;
  19
  20 namespace {
  21
  22 const int kBufferSize = 16 * 1024;
  23
  24 }  // namespace
  25
  26 class AndroidDeviceManager::AndroidWebSocket::WebSocketImpl {
  27  public:
  28   WebSocketImpl(
  29       scoped_refptr<base::SingleThreadTaskRunner> response_task_runner,
  30       base::WeakPtr<AndroidWebSocket> weak_socket,
  31       const std::string& extensions,
  32       const std::string& body_head,
  33       scoped_ptr<net::StreamSocket> socket)
  34       : response_task_runner_(response_task_runner),
  35         weak_socket_(weak_socket),
  36         socket_(socket.Pass()),
  37         encoder_(net::WebSocketEncoder::CreateClient(extensions)),
  38         response_buffer_(body_head) {
  39     thread_checker_.DetachFromThread();
  40   }
  41
  42   void StartListening() {
  43     DCHECK(thread_checker_.CalledOnValidThread());
  44     DCHECK(socket_);
  45
  46     scoped_refptr<net::IOBuffer> buffer(new net::IOBuffer(kBufferSize));
  47
  48     if (response_buffer_.size() > 0)
  49       ProcessResponseBuffer(buffer);
  50     else
  51       Read(buffer);
  52   }
  53
  54   void SendFrame(const std::string& message) {
  55     DCHECK(thread_checker_.CalledOnValidThread());
  56     if (!socket_)
  57       return;
  58     int mask = base::RandInt(0, 0x7FFFFFFF);
  59     std::string encoded_frame;
  60     encoder_->EncodeFrame(message, mask, &encoded_frame);
  61     request_buffer_ += encoded_frame;
  62     if (request_buffer_.length() == encoded_frame.length())
  63       SendPendingRequests(0);
  64   }
  65
  66  private:
  67   void Read(scoped_refptr<net::IOBuffer> io_buffer) {
  68     int result = socket_->Read(
  69         io_buffer.get(),
  70         kBufferSize,
  71         base::Bind(&WebSocketImpl::OnBytesRead,
  72                    base::Unretained(this), io_buffer));
  73     if (result != net::ERR_IO_PENDING)
  74       OnBytesRead(io_buffer, result);
  75   }
  76
  77   void OnBytesRead(scoped_refptr<net::IOBuffer> io_buffer, int result) {
  78     DCHECK(thread_checker_.CalledOnValidThread());
  79     if (result <= 0) {
  80       Disconnect();
  81       return;
  82     }
  83     response_buffer_.append(io_buffer->data(), result);
  84
  85     ProcessResponseBuffer(io_buffer);
  86   }
  87
  88   void ProcessResponseBuffer(scoped_refptr<net::IOBuffer> io_buffer) {
  89     int bytes_consumed;
  90     std::string output;
  91     WebSocket::ParseResult parse_result = encoder_->DecodeFrame(
  92         response_buffer_, &bytes_consumed, &output);
  93
  94     while (parse_result == WebSocket::FRAME_OK) {
  95       response_buffer_ = response_buffer_.substr(bytes_consumed);
  96       response_task_runner_->PostTask(
  97           FROM_HERE,
  98           base::Bind(&AndroidWebSocket::OnFrameRead, weak_socket_, output));
  99       parse_result = encoder_->DecodeFrame(
 100           response_buffer_, &bytes_consumed, &output);
 101     }
 102
 103     if (parse_result == WebSocket::FRAME_ERROR ||
 104         parse_result == WebSocket::FRAME_CLOSE) {
 105       Disconnect();
 106       return;
 107     }
 108     Read(io_buffer);
 109   }
 110
 111   void SendPendingRequests(int result) {
 112     DCHECK(thread_checker_.CalledOnValidThread());
 113     if (result < 0) {
 114       Disconnect();
 115       return;
 116     }
 117     request_buffer_ = request_buffer_.substr(result);
 118     if (request_buffer_.empty())
 119       return;
 120
 121     scoped_refptr<net::StringIOBuffer> buffer =
 122         new net::StringIOBuffer(request_buffer_);
 123     result = socket_->Write(buffer.get(), buffer->size(),
 124                             base::Bind(&WebSocketImpl::SendPendingRequests,
 125                                        base::Unretained(this)));
 126     if (result != net::ERR_IO_PENDING)
 127       SendPendingRequests(result);
 128   }
 129
 130   void Disconnect() {
 131     DCHECK(thread_checker_.CalledOnValidThread());
 132     socket_.reset();
 133     response_task_runner_->PostTask(
 134         FROM_HERE, base::Bind(&AndroidWebSocket::OnSocketClosed, weak_socket_));
 135   }
 136
 137   scoped_refptr<base::SingleThreadTaskRunner> response_task_runner_;
 138   base::WeakPtr<AndroidWebSocket> weak_socket_;
 139   scoped_ptr<net::StreamSocket> socket_;
 140   scoped_ptr<net::WebSocketEncoder> encoder_;
 141   std::string response_buffer_;
 142   std::string request_buffer_;
 143   base::ThreadChecker thread_checker_;
 144   DISALLOW_COPY_AND_ASSIGN(WebSocketImpl);
 145 };
 146
 147 AndroidDeviceManager::AndroidWebSocket::AndroidWebSocket(
 148     scoped_refptr<Device> device,
 149     const std::string& socket_name,
 150     const std::string& path,
 151     Delegate* delegate)
 152     : device_(device.get()),
 153       socket_impl_(nullptr),
 154       delegate_(delegate),
 155       weak_factory_(this) {
 156   DCHECK_CURRENTLY_ON(BrowserThread::UI);
 157   DCHECK(delegate_);
 158   DCHECK(device_);
 159   device_->sockets_.insert(this);
 160   device_->HttpUpgrade(
 161       socket_name, path, net::WebSocketEncoder::kClientExtensions,
 162       base::Bind(&AndroidWebSocket::Connected, weak_factory_.GetWeakPtr()));
 163 }
 164
 165 AndroidDeviceManager::AndroidWebSocket::~AndroidWebSocket() {
 166   DCHECK_CURRENTLY_ON(BrowserThread::UI);
 167   Terminate();
 168 }
 169
 170 void AndroidDeviceManager::AndroidWebSocket::SendFrame(
 171     const std::string& message) {
 172   DCHECK_CURRENTLY_ON(BrowserThread::UI);
 173   DCHECK(socket_impl_);
 174   DCHECK(device_);
 175   device_->task_runner_->PostTask(
 176       FROM_HERE, base::Bind(&WebSocketImpl::SendFrame,
 177                             base::Unretained(socket_impl_), message));
 178 }
 179
 180 void AndroidDeviceManager::AndroidWebSocket::Connected(
 181     int result,
 182     const std::string& extensions,
 183     const std::string& body_head,
 184     scoped_ptr<net::StreamSocket> socket) {
 185   DCHECK_CURRENTLY_ON(BrowserThread::UI);
 186   if (result != net::OK || !socket.get()) {
 187     OnSocketClosed();
 188     return;
 189   }
 190   socket_impl_ = new WebSocketImpl(base::ThreadTaskRunnerHandle::Get(),
 191                                    weak_factory_.GetWeakPtr(), extensions,
 192                                    body_head, socket.Pass());
 193   device_->task_runner_->PostTask(FROM_HERE,
 194                                   base::Bind(&WebSocketImpl::StartListening,
 195                                              base::Unretained(socket_impl_)));
 196   delegate_->OnSocketOpened();
 197 }
 198
 199 void AndroidDeviceManager::AndroidWebSocket::OnFrameRead(
 200     const std::string& message) {
 201   DCHECK_CURRENTLY_ON(BrowserThread::UI);
 202   delegate_->OnFrameRead(message);
 203 }
 204
 205 void AndroidDeviceManager::AndroidWebSocket::OnSocketClosed() {
 206   DCHECK_CURRENTLY_ON(BrowserThread::UI);
 207   Terminate();
 208   delegate_->OnSocketClosed();
 209 }
 210
 211 void AndroidDeviceManager::AndroidWebSocket::Terminate() {
 212   DCHECK_CURRENTLY_ON(BrowserThread::UI);
 213   if (socket_impl_) {
 214     DCHECK(device_);
 215     device_->task_runner_->DeleteSoon(FROM_HERE, socket_impl_);
 216     socket_impl_ = nullptr;
 217   }
 218   if (device_) {
 219     device_->sockets_.erase(this);
 220     device_ = nullptr;
 221   }
 222 }
 223
 224 AndroidDeviceManager::AndroidWebSocket*
 225 AndroidDeviceManager::Device::CreateWebSocket(
 226     const std::string& socket_name,
 227     const std::string& path,
 228     AndroidWebSocket::Delegate* delegate) {
 229   return new AndroidWebSocket(this, socket_name, path, delegate);
 230 }