Move device/usb classes from the FILE thread to UI thread.

chrome/browser/devtools/device/usb/android_usb_device.cc

 // 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 "chrome/browser/devtools/device/usb/android_usb_device.h"
 
 #include <set>
 
 #include "base/barrier_closure.h"
 #include "base/base64.h"
 #include "base/lazy_instance.h"
-#include "base/message_loop/message_loop.h"
+#include "base/single_thread_task_runner.h"
 #include "base/stl_util.h"
 #include "base/strings/string_util.h"
 #include "base/strings/stringprintf.h"
 #include "base/strings/utf_string_conversions.h"
+#include "base/thread_task_runner_handle.h"
 #include "chrome/browser/devtools/device/usb/android_rsa.h"
 #include "chrome/browser/devtools/device/usb/android_usb_socket.h"
 #include "content/public/browser/browser_thread.h"
 #include "crypto/rsa_private_key.h"
 #include "device/core/device_client.h"
 #include "device/usb/usb_descriptors.h"
 #include "device/usb/usb_device.h"
 #include "device/usb/usb_service.h"
 #include "net/base/ip_endpoint.h"
 #include "net/base/net_errors.h"
@@ skipping 35 matching lines @@
   if (interface.alternate_setting != 0 ||
       interface.interface_class != kAdbClass ||
       interface.interface_subclass != kAdbSubclass ||
       interface.interface_protocol != kAdbProtocol ||
       interface.endpoints.size() != 2) {
     return false;
   }
   return true;
 }
 
-scoped_refptr<AndroidUsbDevice> ClaimInterface(
-    crypto::RSAPrivateKey* rsa_key,
-    scoped_refptr<UsbDeviceHandle> usb_handle,
-    const base::string16& serial,
-    const UsbInterfaceDescriptor& interface) {
-  int inbound_address = 0;
-  int outbound_address = 0;
-  int zero_mask = 0;
-
-  for (const UsbEndpointDescriptor& endpoint : interface.endpoints) {
-    if (endpoint.transfer_type != device::USB_TRANSFER_BULK)
-      continue;
-    if (endpoint.direction == device::USB_DIRECTION_INBOUND)
-      inbound_address = endpoint.address;
-    else
-      outbound_address = endpoint.address;
-    zero_mask = endpoint.maximum_packet_size - 1;
+void CountAndroidDevices(const base::Callback<void(int)>& callback,
+                         const UsbDevices& devices) {
+  int device_count = 0;
+  for (const scoped_refptr<UsbDevice>& device : devices) {
+    const UsbConfigDescriptor* config = device->GetConfiguration();
+    if (config) {
+      for (const UsbInterfaceDescriptor& iface : config->interfaces) {
+        if (IsAndroidInterface(iface)) {
+          ++device_count;
+        }
+      }
+    }
   }
 
-  if (inbound_address == 0 || outbound_address == 0)
-    return NULL;
-
-  if (!usb_handle->ClaimInterface(interface.interface_number))
-    return NULL;
-
-  return new AndroidUsbDevice(rsa_key,
-                              usb_handle,
-                              base::UTF16ToASCII(serial),
-                              inbound_address,
-                              outbound_address,
-                              zero_mask,
-                              interface.interface_number);
+  BrowserThread::PostTask(BrowserThread::UI, FROM_HERE,
+                          base::Bind(callback, device_count));
 }
 
 uint32 Checksum(const std::string& data) {
   unsigned char* x = (unsigned char*)data.data();
   int count = data.length();
   uint32 sum = 0;
   while (count-- > 0)
     sum += *x++;
   return sum;
 }
@@ skipping 26 matching lines @@
   LOG(ERROR) << (outgoing ? "[out] " : "[ in] ") << result;
 #endif  // 0
 }
 
 void ReleaseInterface(scoped_refptr<UsbDeviceHandle> usb_device,
                       int interface_id) {
   usb_device->ReleaseInterface(interface_id);
   usb_device->Close();
 }
 
-}  // namespace
-
-AdbMessage::AdbMessage(uint32 command,
-                       uint32 arg0,
-                       uint32 arg1,
-                       const std::string& body)
-    : command(command),
-      arg0(arg0),
-      arg1(arg1),
-      body(body) {
-}
-
-AdbMessage::~AdbMessage() {
-}
-
-static void RespondOnCallerThread(const AndroidUsbDevicesCallback& callback,
-                                  AndroidUsbDevices* new_devices) {
+void RespondOnCallerThread(const AndroidUsbDevicesCallback& callback,
+                           AndroidUsbDevices* new_devices) {
   scoped_ptr<AndroidUsbDevices> devices(new_devices);
 
   // Add raw pointers to the newly claimed devices.
   for (const scoped_refptr<AndroidUsbDevice>& device : *devices) {
     g_devices.Get().push_back(device.get());
   }
 
   // Return all claimed devices.
   AndroidUsbDevices result(g_devices.Get().begin(), g_devices.Get().end());
   callback.Run(result);
 }
 
-static void RespondOnFileThread(
+void RespondOnUIThread(
     const AndroidUsbDevicesCallback& callback,
     AndroidUsbDevices* devices,
-    scoped_refptr<base::MessageLoopProxy> caller_message_loop_proxy) {
-  DCHECK_CURRENTLY_ON(BrowserThread::FILE);
-  caller_message_loop_proxy->PostTask(
-      FROM_HERE,
-      base::Bind(&RespondOnCallerThread, callback, devices));
+    scoped_refptr<base::SingleThreadTaskRunner> caller_task_runner) {
+  DCHECK_CURRENTLY_ON(BrowserThread::UI);
+  caller_task_runner->PostTask(
+      FROM_HERE, base::Bind(&RespondOnCallerThread, callback, devices));
 }
 
-static void OpenAndroidDeviceOnFileThread(
-    AndroidUsbDevices* devices,
-    crypto::RSAPrivateKey* rsa_key,
-    const base::Closure& barrier,
-    scoped_refptr<UsbDevice> device,
-    int interface_id,
-    bool success) {
-  DCHECK_CURRENTLY_ON(BrowserThread::FILE);
+void CreateDeviceOnInterfaceClaimed(AndroidUsbDevices* devices,
+                                    crypto::RSAPrivateKey* rsa_key,
+                                    scoped_refptr<UsbDeviceHandle> usb_handle,
+                                    int inbound_address,
+                                    int outbound_address,
+                                    int zero_mask,
+                                    int interface_number,
+                                    const base::Closure& barrier,
+                                    bool success) {
   if (success) {
-    base::string16 serial;
-    if (device->GetSerialNumber(&serial) && !serial.empty()) {
-      const UsbConfigDescriptor* config = device->GetConfiguration();
-      if (config) {
-        scoped_refptr<UsbDeviceHandle> usb_handle = device->Open();
-        if (usb_handle.get()) {
-          scoped_refptr<AndroidUsbDevice> android_device = ClaimInterface(
-              rsa_key, usb_handle, serial, config->interfaces[interface_id]);
-          if (android_device.get())
-            devices->push_back(android_device);
-          else
-            usb_handle->Close();
-        }
-      }
-    }
+    devices->push_back(new AndroidUsbDevice(
+        rsa_key, usb_handle,
+        base::UTF16ToASCII(usb_handle->GetDevice()->serial_number()),
+        inbound_address, outbound_address, zero_mask, interface_number));
+  } else {
+    usb_handle->Close();
   }
   barrier.Run();
 }
 
-static int CountOnFileThread() {
-  DCHECK_CURRENTLY_ON(BrowserThread::FILE);
-  UsbService* service = device::DeviceClient::Get()->GetUsbService();
-  UsbDevices usb_devices;
-  if (service != NULL)
-    service->GetDevices(&usb_devices);
-  int device_count = 0;
-  for (const scoped_refptr<UsbDevice>& device : usb_devices) {
-    const UsbConfigDescriptor* config = device->GetConfiguration();
-    if (config) {
-      for (const UsbInterfaceDescriptor& iface : config->interfaces) {
-        if (IsAndroidInterface(iface)) {
-          ++device_count;
-        }
-      }
-    }
+void OnDeviceOpened(AndroidUsbDevices* devices,
+                    crypto::RSAPrivateKey* rsa_key,
+                    int inbound_address,
+                    int outbound_address,
+                    int zero_mask,
+                    int interface_number,
+                    const base::Closure& barrier,
+                    scoped_refptr<UsbDeviceHandle> usb_handle) {
+  if (usb_handle.get()) {
+    usb_handle->ClaimInterface(
+        interface_number,
+        base::Bind(&CreateDeviceOnInterfaceClaimed, devices, rsa_key,
+                   usb_handle, inbound_address, outbound_address, zero_mask,
+                   interface_number, barrier));
+  } else {
+    barrier.Run();
   }
-  return device_count;
 }
 
-static void EnumerateOnFileThread(
+void OpenAndroidDevice(AndroidUsbDevices* devices,
+                       crypto::RSAPrivateKey* rsa_key,
+                       const base::Closure& barrier,
+                       scoped_refptr<UsbDevice> device,
+                       int interface_id,
+                       bool success) {
+  if (!success) {
+    barrier.Run();
+    return;
+  }
+
+  if (device->serial_number().empty()) {
+    barrier.Run();
+    return;
+  }
+
+  const UsbConfigDescriptor* config = device->GetConfiguration();
+  if (!config) {
+    barrier.Run();
+    return;
+  }
+
+  const UsbInterfaceDescriptor& interface = config->interfaces[interface_id];
+  int inbound_address = 0;
+  int outbound_address = 0;
+  int zero_mask = 0;
+
+  for (const UsbEndpointDescriptor& endpoint : interface.endpoints) {
+    if (endpoint.transfer_type != device::USB_TRANSFER_BULK)
+      continue;
+    if (endpoint.direction == device::USB_DIRECTION_INBOUND)
+      inbound_address = endpoint.address;
+    else
+      outbound_address = endpoint.address;
+    zero_mask = endpoint.maximum_packet_size - 1;
+  }
+
+  if (inbound_address == 0 || outbound_address == 0) {
+    barrier.Run();
+    return;
+  }
+
+  device->Open(base::Bind(&OnDeviceOpened, devices, rsa_key, inbound_address,
+                          outbound_address, zero_mask,
+                          interface.interface_number, barrier));
+}
+
+void OpenAndroidDevices(
     crypto::RSAPrivateKey* rsa_key,
     const AndroidUsbDevicesCallback& callback,
-    scoped_refptr<base::MessageLoopProxy> caller_message_loop_proxy) {
-  DCHECK_CURRENTLY_ON(BrowserThread::FILE);
-
-  UsbService* service = device::DeviceClient::Get()->GetUsbService();
-  UsbDevices usb_devices;
-  if (service != NULL)
-    service->GetDevices(&usb_devices);
-
+    scoped_refptr<base::SingleThreadTaskRunner> caller_task_runner,
+    const UsbDevices& usb_devices) {
   // Add new devices.
   AndroidUsbDevices* devices = new AndroidUsbDevices();
   base::Closure barrier = base::BarrierClosure(
-      usb_devices.size(), base::Bind(&RespondOnFileThread,
-                                     callback,
-                                     devices,
-                                     caller_message_loop_proxy));
+      usb_devices.size(),
+      base::Bind(&RespondOnUIThread, callback, devices, caller_task_runner));
 
   for (const scoped_refptr<UsbDevice>& device : usb_devices) {
     const UsbConfigDescriptor* config = device->GetConfiguration();
     if (!config) {
       barrier.Run();
       continue;
     }
     bool has_android_interface = false;
     for (size_t j = 0; j < config->interfaces.size(); ++j) {
       if (!IsAndroidInterface(config->interfaces[j])) {
         continue;
       }
 
       // Request permission on Chrome OS.
 #if defined(OS_CHROMEOS)
-      device->RequestUsbAccess(
-          j, base::Bind(&OpenAndroidDeviceOnFileThread, devices, rsa_key,
-                        barrier, device, j));
+      device->RequestUsbAccess(j, base::Bind(&OpenAndroidDevice, devices,
+                                             rsa_key, barrier, device, j));
 #else
-      OpenAndroidDeviceOnFileThread(devices, rsa_key, barrier, device, j, true);
+      OpenAndroidDevice(devices, rsa_key, barrier, device, j, true);
 #endif  // defined(OS_CHROMEOS)
 
       has_android_interface = true;
       break;
     }
-    if (!has_android_interface)
+    if (!has_android_interface) {
       barrier.Run();
+    }
   }
 }
 
+void EnumerateOnUIThread(
+    crypto::RSAPrivateKey* rsa_key,
+    const AndroidUsbDevicesCallback& callback,
+    scoped_refptr<base::SingleThreadTaskRunner> caller_task_runner) {
+  DCHECK_CURRENTLY_ON(BrowserThread::UI);
+
+  UsbService* service = device::DeviceClient::Get()->GetUsbService();
+  if (service == NULL) {
+    caller_task_runner->PostTask(FROM_HERE,
+                                 base::Bind(callback, AndroidUsbDevices()));
+  } else {
+    service->GetDevices(
+        base::Bind(&OpenAndroidDevices, rsa_key, callback, caller_task_runner));
+  }
+}
+
+}  // namespace
+
+AdbMessage::AdbMessage(uint32 command,
+                       uint32 arg0,
+                       uint32 arg1,
+                       const std::string& body)
+    : command(command), arg0(arg0), arg1(arg1), body(body) {
+}
+
+AdbMessage::~AdbMessage() {
+}
+
 // static
-void AndroidUsbDevice::CountDevices(
-    const base::Callback<void(int)>& callback) {
-  BrowserThread::PostTaskAndReplyWithResult(
-      BrowserThread::FILE,
-      FROM_HERE,
-      base::Bind(&CountOnFileThread),
-      callback);
+void AndroidUsbDevice::CountDevices(const base::Callback<void(int)>& callback) {
+  UsbService* service = device::DeviceClient::Get()->GetUsbService();
+  if (service != NULL) {
+    service->GetDevices(base::Bind(&CountAndroidDevices, callback));
+  } else {
+    BrowserThread::PostTask(BrowserThread::UI, FROM_HERE,
+                            base::Bind(callback, 0));
+  }
 }
 
 // static
 void AndroidUsbDevice::Enumerate(crypto::RSAPrivateKey* rsa_key,
                                  const AndroidUsbDevicesCallback& callback) {
-
   // Collect devices with closed handles.
   for (AndroidUsbDevice* device : g_devices.Get()) {
     if (device->usb_handle_.get()) {
-      BrowserThread::PostTask(BrowserThread::FILE, FROM_HERE,
-          base::Bind(&AndroidUsbDevice::TerminateIfReleased, device,
-                     device->usb_handle_));
+      BrowserThread::PostTask(BrowserThread::UI, FROM_HERE,
+                              base::Bind(&AndroidUsbDevice::TerminateIfReleased,
+                                         device, device->usb_handle_));
     }
   }
 
   // Then look for the new devices.
-  BrowserThread::PostTask(BrowserThread::FILE, FROM_HERE,
-                          base::Bind(&EnumerateOnFileThread, rsa_key, callback,
+  BrowserThread::PostTask(BrowserThread::UI, FROM_HERE,
+                          base::Bind(&EnumerateOnUIThread, rsa_key, callback,
                                      base::MessageLoopProxy::current()));
 }
 
 AndroidUsbDevice::AndroidUsbDevice(crypto::RSAPrivateKey* rsa_key,
                                    scoped_refptr<UsbDeviceHandle> usb_device,
                                    const std::string& serial,
                                    int inbound_address,
                                    int outbound_address,
                                    int zero_mask,
                                    int interface_id)
-    : message_loop_(NULL),
-      rsa_key_(rsa_key->Copy()),
+    : rsa_key_(rsa_key->Copy()),
       usb_handle_(usb_device),
       serial_(serial),
       inbound_address_(inbound_address),
       outbound_address_(outbound_address),
       zero_mask_(zero_mask),
       interface_id_(interface_id),
       is_connected_(false),
       signature_sent_(false),
       last_socket_id_(256),
       weak_factory_(this) {
 }
 
 void AndroidUsbDevice::InitOnCallerThread() {
-  if (message_loop_)
+  if (task_runner_)
     return;
-  message_loop_ = base::MessageLoop::current();
+  task_runner_ = base::ThreadTaskRunnerHandle::Get();
   Queue(make_scoped_ptr(new AdbMessage(AdbMessage::kCommandCNXN, kVersion,
                                        kMaxPayload, kHostConnectMessage)));
   ReadHeader();
 }
 
 net::StreamSocket* AndroidUsbDevice::CreateSocket(const std::string& command) {
   if (!usb_handle_.get())
     return NULL;
 
   uint32 socket_id = ++last_socket_id_;
   sockets_[socket_id] = new AndroidUsbSocket(this, socket_id, command,
       base::Bind(&AndroidUsbDevice::SocketDeleted, this, socket_id));
   return sockets_[socket_id];
 }
 
 void AndroidUsbDevice::Send(uint32 command,
                             uint32 arg0,
                             uint32 arg1,
                             const std::string& body) {
   scoped_ptr<AdbMessage> message(new AdbMessage(command, arg0, arg1, body));
   // Delay open request if not yet connected.
   if (!is_connected_) {
     pending_messages_.push_back(message.release());
     return;
   }
   Queue(message.Pass());
 }
 
 AndroidUsbDevice::~AndroidUsbDevice() {
-  DCHECK(message_loop_ == base::MessageLoop::current());
+  DCHECK(task_runner_->BelongsToCurrentThread());
   Terminate();
 }
 
 void AndroidUsbDevice::Queue(scoped_ptr<AdbMessage> message) {
-  DCHECK(message_loop_ == base::MessageLoop::current());
+  DCHECK(task_runner_->BelongsToCurrentThread());
 
   // Queue header.
   std::vector<uint32> header;
   header.push_back(message->command);
   header.push_back(message->arg0);
   header.push_back(message->arg1);
   bool append_zero = true;
   if (message->body.empty())
     append_zero = false;
   if (message->command == AdbMessage::kCommandAUTH &&
@@ skipping 21 matching lines @@
     outgoing_queue_.push(body_buffer);
     if (zero_mask_ && (body_length & zero_mask_) == 0) {
       // Send a zero length packet.
       outgoing_queue_.push(new net::IOBufferWithSize(0));
     }
   }
   ProcessOutgoing();
 }
 
 void AndroidUsbDevice::ProcessOutgoing() {
-  DCHECK(message_loop_ == base::MessageLoop::current());
+  DCHECK(task_runner_->BelongsToCurrentThread());
 
   if (outgoing_queue_.empty() || !usb_handle_.get())
     return;
 
   BulkMessage message = outgoing_queue_.front();
   outgoing_queue_.pop();
   DumpMessage(true, message->data(), message->size());
-  usb_handle_->BulkTransfer(device::USB_DIRECTION_OUTBOUND,
-                            outbound_address_,
-                            message.get(),
-                            message->size(),
-                            kUsbTimeout,
-                            base::Bind(&AndroidUsbDevice::OutgoingMessageSent,
-                                       weak_factory_.GetWeakPtr()));
+
+  BrowserThread::PostTask(
+      BrowserThread::UI, FROM_HERE,
+      base::Bind(&UsbDeviceHandle::BulkTransfer, usb_handle_,
+                 device::USB_DIRECTION_OUTBOUND, outbound_address_, message,
+                 message->size(), kUsbTimeout,
+                 base::Bind(&AndroidUsbDevice::OutgoingMessageSent,
+                            weak_factory_.GetWeakPtr(), task_runner_)));
 }
 
-void AndroidUsbDevice::OutgoingMessageSent(UsbTransferStatus status,
-                                           scoped_refptr<net::IOBuffer> buffer,
-                                           size_t result) {
-  DCHECK(message_loop_ == base::MessageLoop::current());
+// static
+void AndroidUsbDevice::OutgoingMessageSent(
+    base::WeakPtr<AndroidUsbDevice> device,
+    scoped_refptr<base::SingleThreadTaskRunner> task_runner,
+    UsbTransferStatus status,
+    scoped_refptr<net::IOBuffer> buffer,
+    size_t result) {
+  if (status != device::USB_TRANSFER_COMPLETED) {
+    return;
+  }
 
-  if (status != device::USB_TRANSFER_COMPLETED)
-    return;
-  message_loop_->PostTask(FROM_HERE,
-                          base::Bind(&AndroidUsbDevice::ProcessOutgoing, this));
+  task_runner->PostTask(FROM_HERE,
+                        base::Bind(&AndroidUsbDevice::ProcessOutgoing, device));
 }
 
 void AndroidUsbDevice::ReadHeader() {
-  DCHECK(message_loop_ == base::MessageLoop::current());
+  DCHECK(task_runner_->BelongsToCurrentThread());
 
-  if (!usb_handle_.get())
+  if (!usb_handle_.get()) {
     return;
+  }
+
   scoped_refptr<net::IOBuffer> buffer = new net::IOBuffer(kHeaderSize);
-  usb_handle_->BulkTransfer(
-      device::USB_DIRECTION_INBOUND,
-      inbound_address_,
-      buffer.get(),
-      kHeaderSize,
-      kUsbTimeout,
-      base::Bind(&AndroidUsbDevice::ParseHeader, weak_factory_.GetWeakPtr()));
+  BrowserThread::PostTask(
+      BrowserThread::UI, FROM_HERE,
+      base::Bind(&UsbDeviceHandle::BulkTransfer, usb_handle_,
+                 device::USB_DIRECTION_INBOUND, inbound_address_, buffer,
+                 kHeaderSize, kUsbTimeout,
+                 base::Bind(&AndroidUsbDevice::ReceivedHeader,
+                            weak_factory_.GetWeakPtr(), task_runner_)));
+}
+
+// static
+void AndroidUsbDevice::ReceivedHeader(
+    base::WeakPtr<AndroidUsbDevice> device,
+    scoped_refptr<base::SingleThreadTaskRunner> task_runner,
+    device::UsbTransferStatus status,
+    scoped_refptr<net::IOBuffer> buffer,
+    size_t result) {
+  task_runner->PostTask(FROM_HERE, base::Bind(&AndroidUsbDevice::ParseHeader,
+                                              device, status, buffer, result));
 }
 
 void AndroidUsbDevice::ParseHeader(UsbTransferStatus status,
                                    scoped_refptr<net::IOBuffer> buffer,
                                    size_t result) {
-  DCHECK(message_loop_ == base::MessageLoop::current());
+  DCHECK(task_runner_->BelongsToCurrentThread());
 
   if (status == device::USB_TRANSFER_TIMEOUT) {
-    message_loop_->PostTask(FROM_HERE,
-                            base::Bind(&AndroidUsbDevice::ReadHeader, this));
+    ReadHeader();
     return;
   }
 
   if (status != device::USB_TRANSFER_COMPLETED || result != kHeaderSize) {
     TransferError(status);
     return;
   }
 
   DumpMessage(false, buffer->data(), result);
   std::vector<uint32> header(6);
   memcpy(&header[0], buffer->data(), result);
   scoped_ptr<AdbMessage> message(
       new AdbMessage(header[0], header[1], header[2], ""));
   uint32 data_length = header[3];
   uint32 data_check = header[4];
   uint32 magic = header[5];
   if ((message->command ^ 0xffffffff) != magic) {
     TransferError(device::USB_TRANSFER_ERROR);
     return;
   }
 
   if (data_length == 0) {
-    message_loop_->PostTask(FROM_HERE,
-                            base::Bind(&AndroidUsbDevice::HandleIncoming, this,
-                                       base::Passed(&message)));
-    return;
+    HandleIncoming(message.Pass());
+  } else {
+    ReadBody(message.Pass(), data_length, data_check);
   }
-
-  message_loop_->PostTask(FROM_HERE,
-      base::Bind(&AndroidUsbDevice::ReadBody, this,
-                 base::Passed(&message), data_length, data_check));
 }
 
 void AndroidUsbDevice::ReadBody(scoped_ptr<AdbMessage> message,
                                 uint32 data_length,
                                 uint32 data_check) {
-  DCHECK(message_loop_ == base::MessageLoop::current());
+  DCHECK(task_runner_->BelongsToCurrentThread());
 
-  if (!usb_handle_.get())
+  if (!usb_handle_.get()) {
     return;
+  }
+
   scoped_refptr<net::IOBuffer> buffer = new net::IOBuffer(data_length);
-  usb_handle_->BulkTransfer(device::USB_DIRECTION_INBOUND,
-                            inbound_address_,
-                            buffer.get(),
-                            data_length,
-                            kUsbTimeout,
-                            base::Bind(&AndroidUsbDevice::ParseBody,
-                                       weak_factory_.GetWeakPtr(),
-                                       base::Passed(&message),
-                                       data_length,
-                                       data_check));
+  BrowserThread::PostTask(
+      BrowserThread::UI, FROM_HERE,
+      base::Bind(&UsbDeviceHandle::BulkTransfer, usb_handle_,
+                 device::USB_DIRECTION_INBOUND, inbound_address_, buffer,
+                 data_length, kUsbTimeout,
+                 base::Bind(&AndroidUsbDevice::ReceivedBody,
+                            weak_factory_.GetWeakPtr(), task_runner_,
+                            base::Passed(&message), data_length, data_check)));
+}
+
+// static
+void AndroidUsbDevice::ReceivedBody(
+    base::WeakPtr<AndroidUsbDevice> device,
+    scoped_refptr<base::SingleThreadTaskRunner> task_runner,
+    scoped_ptr<AdbMessage> message,
+    uint32 data_length,
+    uint32 data_check,
+    device::UsbTransferStatus status,
+    scoped_refptr<net::IOBuffer> buffer,
+    size_t result) {
+  task_runner->PostTask(
+      FROM_HERE,
+      base::Bind(&AndroidUsbDevice::ParseBody, device, base::Passed(&message),
+                 data_length, data_check, status, buffer, result));
 }
 
 void AndroidUsbDevice::ParseBody(scoped_ptr<AdbMessage> message,
                                  uint32 data_length,
                                  uint32 data_check,
                                  UsbTransferStatus status,
                                  scoped_refptr<net::IOBuffer> buffer,
                                  size_t result) {
-  DCHECK(message_loop_ == base::MessageLoop::current());
+  DCHECK(task_runner_->BelongsToCurrentThread());
 
   if (status == device::USB_TRANSFER_TIMEOUT) {
-    message_loop_->PostTask(FROM_HERE,
-                            base::Bind(&AndroidUsbDevice::ReadBody, this,
-                            base::Passed(&message), data_length, data_check));
+    ReadBody(message.Pass(), data_length, data_check);
     return;
   }
 
   if (status != device::USB_TRANSFER_COMPLETED ||
       static_cast<uint32>(result) != data_length) {
     TransferError(status);
     return;
   }
 
   DumpMessage(false, buffer->data(), data_length);
   message->body = std::string(buffer->data(), result);
   if (Checksum(message->body) != data_check) {
     TransferError(device::USB_TRANSFER_ERROR);
     return;
   }
 
-  message_loop_->PostTask(FROM_HERE,
-                          base::Bind(&AndroidUsbDevice::HandleIncoming, this,
-                                     base::Passed(&message)));
+  HandleIncoming(message.Pass());
 }
 
 void AndroidUsbDevice::HandleIncoming(scoped_ptr<AdbMessage> message) {
-  DCHECK(message_loop_ == base::MessageLoop::current());
+  DCHECK(task_runner_->BelongsToCurrentThread());
 
   switch (message->command) {
     case AdbMessage::kCommandAUTH:
       {
         DCHECK_EQ(message->arg0, static_cast<uint32>(AdbMessage::kAuthToken));
         if (signature_sent_) {
           Queue(make_scoped_ptr(new AdbMessage(
               AdbMessage::kCommandAUTH,
               AdbMessage::kAuthRSAPublicKey, 0,
               AndroidRSAPublicKey(rsa_key_.get()))));
@@ skipping 33 matching lines @@
           it->second->HandleIncoming(message.Pass());
       }
       break;
     default:
       break;
   }
   ReadHeader();
 }
 
 void AndroidUsbDevice::TransferError(UsbTransferStatus status) {
-  DCHECK(message_loop_ == base::MessageLoop::current());
+  DCHECK(task_runner_->BelongsToCurrentThread());
 
-  message_loop_->PostTask(FROM_HERE,
-                          base::Bind(&AndroidUsbDevice::Terminate, this));
+  Terminate();
 }
 
 void AndroidUsbDevice::TerminateIfReleased(
     scoped_refptr<UsbDeviceHandle> usb_handle) {
-  DCHECK_CURRENTLY_ON(BrowserThread::FILE);
-  if (usb_handle->GetDevice().get())
+  DCHECK_CURRENTLY_ON(BrowserThread::UI);
+  if (usb_handle->GetDevice().get()) {
     return;
-  message_loop_->PostTask(FROM_HERE,
-                          base::Bind(&AndroidUsbDevice::Terminate, this));
+  }
+
+  task_runner_->PostTask(FROM_HERE,
+                         base::Bind(&AndroidUsbDevice::Terminate, this));
 }
 
 void AndroidUsbDevice::Terminate() {
-  DCHECK(message_loop_ == base::MessageLoop::current());
+  DCHECK(task_runner_->BelongsToCurrentThread());
 
   std::vector<AndroidUsbDevice*>::iterator it =
       std::find(g_devices.Get().begin(), g_devices.Get().end(), this);
   if (it != g_devices.Get().end())
     g_devices.Get().erase(it);
 
   if (!usb_handle_.get())
     return;
 
   // Make sure we zero-out handle so that closing connections did not open
   // new connections.
   scoped_refptr<UsbDeviceHandle> usb_handle = usb_handle_;
   usb_handle_ = NULL;
 
   // Iterate over copy.
   AndroidUsbSockets sockets(sockets_);
   for (AndroidUsbSockets::iterator it = sockets.begin();
        it != sockets.end(); ++it) {
     it->second->Terminated(true);
   }
   DCHECK(sockets_.empty());
 
   BrowserThread::PostTask(
-      BrowserThread::FILE, FROM_HERE,
+      BrowserThread::UI, FROM_HERE,
       base::Bind(&ReleaseInterface, usb_handle, interface_id_));
 }
 
 void AndroidUsbDevice::SocketDeleted(uint32 socket_id) {
-  DCHECK(message_loop_ == base::MessageLoop::current());
+  DCHECK(task_runner_->BelongsToCurrentThread());
 
   sockets_.erase(socket_id);
 }