src/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/memory/weak_ptr.h"
   6 #include "base/message_loop/message_loop.h"
   7 #include "base/rand_util.h"
   8 #include "chrome/browser/devtools/device/android_device_manager.h"
   9 #include "content/public/browser/browser_thread.h"
  10 #include "net/base/io_buffer.h"
  11 #include "net/base/net_errors.h"
  12 #include "net/server/web_socket.h"
  13 #include "net/socket/stream_socket.h"
  14
  15 using content::BrowserThread;
  16 using net::WebSocket;
  17
  18 namespace {
  19
  20 const int kBufferSize = 16 * 1024;
  21
  22 class WebSocketImpl {
  23  public:
  24   typedef AndroidDeviceManager::AndroidWebSocket::Delegate Delegate;
  25
  26   WebSocketImpl(Delegate* delegate,
  27                 scoped_ptr<net::StreamSocket> socket);
  28   void StartListening();
  29   void SendFrame(const std::string& message);
  30
  31  private:
  32   void OnBytesRead(scoped_refptr<net::IOBuffer> response_buffer, int result);
  33   void SendPendingRequests(int result);
  34   void Disconnect();
  35
  36   Delegate* delegate_;
  37   scoped_ptr<net::StreamSocket> socket_;
  38   std::string response_buffer_;
  39   std::string request_buffer_;
  40   base::ThreadChecker thread_checker_;
  41   DISALLOW_COPY_AND_ASSIGN(WebSocketImpl);
  42 };
  43
  44 class DelegateWrapper
  45     : public AndroidDeviceManager::AndroidWebSocket::Delegate {
  46  public:
  47   DelegateWrapper(base::WeakPtr<Delegate> weak_delegate,
  48                   scoped_refptr<base::MessageLoopProxy> message_loop)
  49       : weak_delegate_(weak_delegate),
  50         message_loop_(message_loop) {
  51   }
  52
  53   virtual ~DelegateWrapper() {}
  54
  55   // AndroidWebSocket::Delegate implementation
  56   virtual void OnSocketOpened() OVERRIDE {
  57     message_loop_->PostTask(FROM_HERE,
  58         base::Bind(&Delegate::OnSocketOpened, weak_delegate_));
  59   }
  60
  61   virtual void OnFrameRead(const std::string& message) OVERRIDE {
  62     message_loop_->PostTask(FROM_HERE,
  63         base::Bind(&Delegate::OnFrameRead, weak_delegate_, message));
  64   }
  65
  66   virtual void OnSocketClosed() OVERRIDE {
  67     message_loop_->PostTask(FROM_HERE,
  68         base::Bind(&Delegate::OnSocketClosed, weak_delegate_));
  69   }
  70
  71  private:
  72   base::WeakPtr<Delegate> weak_delegate_;
  73   scoped_refptr<base::MessageLoopProxy> message_loop_;
  74 };
  75
  76 class AndroidWebSocketImpl
  77     : public AndroidDeviceManager::AndroidWebSocket,
  78       public AndroidDeviceManager::AndroidWebSocket::Delegate {
  79  public:
  80   typedef AndroidDeviceManager::Device Device;
  81   AndroidWebSocketImpl(
  82       scoped_refptr<base::MessageLoopProxy> device_message_loop,
  83       scoped_refptr<Device> device,
  84       const std::string& socket_name,
  85       const std::string& url,
  86       AndroidWebSocket::Delegate* delegate);
  87
  88   virtual ~AndroidWebSocketImpl();
  89
  90   // AndroidWebSocket implementation
  91   virtual void SendFrame(const std::string& message) OVERRIDE;
  92
  93   // AndroidWebSocket::Delegate implementation
  94   virtual void OnSocketOpened() OVERRIDE;
  95   virtual void OnFrameRead(const std::string& message) OVERRIDE;
  96   virtual void OnSocketClosed() OVERRIDE;
  97
  98  private:
  99   void Connected(int result, scoped_ptr<net::StreamSocket> socket);
 100
 101   scoped_refptr<base::MessageLoopProxy> device_message_loop_;
 102   scoped_refptr<Device> device_;
 103   std::string socket_name_;
 104   std::string url_;
 105   WebSocketImpl* connection_;
 106   DelegateWrapper* delegate_wrapper_;
 107   AndroidWebSocket::Delegate* delegate_;
 108   base::WeakPtrFactory<AndroidWebSocketImpl> weak_factory_;
 109   DISALLOW_COPY_AND_ASSIGN(AndroidWebSocketImpl);
 110 };
 111
 112 AndroidWebSocketImpl::AndroidWebSocketImpl(
 113     scoped_refptr<base::MessageLoopProxy> device_message_loop,
 114     scoped_refptr<Device> device,
 115     const std::string& socket_name,
 116     const std::string& url,
 117     AndroidWebSocket::Delegate* delegate)
 118     : device_message_loop_(device_message_loop),
 119       device_(device),
 120       socket_name_(socket_name),
 121       url_(url),
 122       delegate_(delegate),
 123       weak_factory_(this) {
 124   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
 125   DCHECK(delegate_);
 126   device_->HttpUpgrade(
 127       socket_name_, url_,
 128       base::Bind(&AndroidWebSocketImpl::Connected, weak_factory_.GetWeakPtr()));
 129 }
 130
 131 void AndroidWebSocketImpl::SendFrame(const std::string& message) {
 132   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
 133   device_message_loop_->PostTask(
 134       FROM_HERE,
 135       base::Bind(&WebSocketImpl::SendFrame,
 136                  base::Unretained(connection_), message));
 137 }
 138
 139 void WebSocketImpl::SendFrame(const std::string& message) {
 140   DCHECK(thread_checker_.CalledOnValidThread());
 141   if (!socket_)
 142     return;
 143   int mask = base::RandInt(0, 0x7FFFFFFF);
 144   std::string encoded_frame = WebSocket::EncodeFrameHybi17(message, mask);
 145   request_buffer_ += encoded_frame;
 146   if (request_buffer_.length() == encoded_frame.length())
 147     SendPendingRequests(0);
 148 }
 149
 150 AndroidWebSocketImpl::~AndroidWebSocketImpl() {
 151   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
 152   device_message_loop_->DeleteSoon(FROM_HERE, connection_);
 153   device_message_loop_->DeleteSoon(FROM_HERE, delegate_wrapper_);
 154 }
 155
 156 WebSocketImpl::WebSocketImpl(Delegate* delegate,
 157                              scoped_ptr<net::StreamSocket> socket)
 158                              : delegate_(delegate),
 159                                socket_(socket.Pass()) {
 160   thread_checker_.DetachFromThread();
 161 }
 162
 163 void AndroidWebSocketImpl::Connected(int result,
 164                                      scoped_ptr<net::StreamSocket> socket) {
 165   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
 166   if (result != net::OK || socket == NULL) {
 167     OnSocketClosed();
 168     return;
 169   }
 170   delegate_wrapper_ = new DelegateWrapper(weak_factory_.GetWeakPtr(),
 171                                           base::MessageLoopProxy::current());
 172   connection_ = new WebSocketImpl(delegate_wrapper_, socket.Pass());
 173   device_message_loop_->PostTask(
 174       FROM_HERE,
 175       base::Bind(&WebSocketImpl::StartListening,
 176                  base::Unretained(connection_)));
 177   OnSocketOpened();
 178 }
 179
 180 void WebSocketImpl::StartListening() {
 181   DCHECK(thread_checker_.CalledOnValidThread());
 182   DCHECK(socket_);
 183   scoped_refptr<net::IOBuffer> response_buffer =
 184       new net::IOBuffer(kBufferSize);
 185   int result = socket_->Read(
 186       response_buffer.get(),
 187       kBufferSize,
 188       base::Bind(&WebSocketImpl::OnBytesRead,
 189                  base::Unretained(this), response_buffer));
 190   if (result != net::ERR_IO_PENDING)
 191     OnBytesRead(response_buffer, result);
 192 }
 193
 194 void WebSocketImpl::OnBytesRead(scoped_refptr<net::IOBuffer> response_buffer,
 195                                 int result) {
 196   DCHECK(thread_checker_.CalledOnValidThread());
 197   if (result <= 0) {
 198     Disconnect();
 199     return;
 200   }
 201
 202   response_buffer_.append(response_buffer->data(), result);
 203
 204   int bytes_consumed;
 205   std::string output;
 206   WebSocket::ParseResult parse_result = WebSocket::DecodeFrameHybi17(
 207       response_buffer_, false, &bytes_consumed, &output);
 208
 209   while (parse_result == WebSocket::FRAME_OK) {
 210     response_buffer_ = response_buffer_.substr(bytes_consumed);
 211     delegate_->OnFrameRead(output);
 212     parse_result = WebSocket::DecodeFrameHybi17(
 213         response_buffer_, false, &bytes_consumed, &output);
 214   }
 215
 216   if (parse_result == WebSocket::FRAME_ERROR ||
 217       parse_result == WebSocket::FRAME_CLOSE) {
 218     Disconnect();
 219     return;
 220   }
 221
 222   result = socket_->Read(
 223       response_buffer.get(),
 224       kBufferSize,
 225       base::Bind(&WebSocketImpl::OnBytesRead,
 226                  base::Unretained(this), response_buffer));
 227   if (result != net::ERR_IO_PENDING)
 228     OnBytesRead(response_buffer, result);
 229 }
 230
 231 void WebSocketImpl::SendPendingRequests(int result) {
 232   DCHECK(thread_checker_.CalledOnValidThread());
 233   if (result < 0) {
 234     Disconnect();
 235     return;
 236   }
 237   request_buffer_ = request_buffer_.substr(result);
 238   if (request_buffer_.empty())
 239     return;
 240
 241   scoped_refptr<net::StringIOBuffer> buffer =
 242       new net::StringIOBuffer(request_buffer_);
 243   result = socket_->Write(buffer.get(), buffer->size(),
 244                           base::Bind(&WebSocketImpl::SendPendingRequests,
 245                                      base::Unretained(this)));
 246   if (result != net::ERR_IO_PENDING)
 247     SendPendingRequests(result);
 248 }
 249
 250 void WebSocketImpl::Disconnect() {
 251   DCHECK(thread_checker_.CalledOnValidThread());
 252   socket_.reset();
 253   delegate_->OnSocketClosed();
 254 }
 255
 256 void AndroidWebSocketImpl::OnSocketOpened() {
 257   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
 258   delegate_->OnSocketOpened();
 259 }
 260
 261 void AndroidWebSocketImpl::OnFrameRead(const std::string& message) {
 262   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
 263   delegate_->OnFrameRead(message);
 264 }
 265
 266 void AndroidWebSocketImpl::OnSocketClosed() {
 267   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
 268   delegate_->OnSocketClosed();
 269 }
 270
 271 }  // namespace
 272
 273 AndroidDeviceManager::AndroidWebSocket*
 274 AndroidDeviceManager::Device::CreateWebSocket(
 275     const std::string& socket,
 276     const std::string& url,
 277     AndroidDeviceManager::AndroidWebSocket::Delegate* delegate) {
 278   return new AndroidWebSocketImpl(
 279       device_message_loop_, this, socket, url, delegate);
 280 }