Clean up thread handling in Blimp browser tests.

blimp/engine/mojo/blob_channel_service.cc

 // Copyright 2016 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 "blimp/engine/mojo/blob_channel_service.h"
 
 #include <string>
 #include <unordered_map>
 #include <utility>
 #include <vector>
 
 #include "base/memory/ptr_util.h"
+#include "base/task_runner_util.h"
+#include "base/threading/thread.h"
 #include "blimp/net/blob_channel/blob_channel_sender.h"
 #include "mojo/public/cpp/system/buffer.h"
 
 namespace blimp {
 namespace engine {
 
-BlobChannelService::BlobChannelService(BlobChannelSender* blob_channel_sender)
-    : blob_channel_sender_(blob_channel_sender) {
-  DCHECK(blob_channel_sender_);
+BlobChannelService::BlobChannelService(
+    base::WeakPtr<BlobChannelSender> blob_channel_sender,
+    scoped_refptr<base::SingleThreadTaskRunner> task_runner)
+    : blob_channel_sender_(blob_channel_sender), task_runner_(task_runner) {
+  DCHECK(task_runner_.get());
 }
 
-BlobChannelService::~BlobChannelService() {}
+BlobChannelService::~BlobChannelService() {
+  DCHECK(thread_checker_.CalledOnValidThread());
+}
 
 void BlobChannelService::GetCachedBlobIds(
     const BlobChannelService::GetCachedBlobIdsCallback& response_callback) {
+  DCHECK(thread_checker_.CalledOnValidThread());
   VLOG(1) << "BlobChannel::GetCachedBlobIds called.";
+
+  // Pull the list of blob IDs from the UI thread.
+  base::PostTaskAndReplyWithResult(
+      task_runner_.get(),
+      FROM_HERE,
+      base::Bind(&BlobChannelService::GetCachedBlobIdsOnUiThread,
+                 base::Unretained(this)),

[Wez, 2016/08/19 19:15:50]

What happens if you just Bind(BlobChannelSender::GetCachedBlobIds,
blob_channel_sender_) here?  Then the task-runner should check the bound WeakPtr
and skip the task if it is null, but I'm not sure what that means for the
return-value...?

[Kevin M, 2016/08/19 21:43:33]

There's a static assert that prevents one from Binding() non-void methods of a
WeakPtr'd object.

[Wez, 2016/08/20 00:40:24]

Awww, sad panda. But also yay for static asserts preventing heinous choices. :P

+      base::Bind(&BlobChannelService::ReceivedBlobIds,
+                 base::Unretained(this),
+                 response_callback));

[Wez, 2016/08/19 19:15:50]

What prevents BlobChannelService being torn-down before ReceivedBlobIds is
processed?

[Kevin M, 2016/08/19 21:43:33]

Nothing... switched to WeakPtr and moved the UI thread getter outside the class
to an anonymous namespace function.

+}
+
+std::vector<BlobChannelSender::CacheStateEntry>
+BlobChannelService::GetCachedBlobIdsOnUiThread() {

[Wez, 2016/08/19 19:15:50]

If we must have this method (i.e if Bind()ing directly won't work) then let's
make it a static function so that the BlobChannelService is only ever "touched"
on the IO thread.

[Kevin M, 2016/08/19 21:43:33]

Done.

+  DCHECK(task_runner_->BelongsToCurrentThread());
+
+  if (blob_channel_sender_) {
+    return blob_channel_sender_->GetCachedBlobIds();
+  } else {
+    return std::vector<BlobChannelSender::CacheStateEntry>();
+  }
+}
+
+void BlobChannelService::ReceivedBlobIds(
+    const BlobChannelService::GetCachedBlobIdsCallback& response_callback,
+    const std::vector<BlobChannelSender::CacheStateEntry>& ids) {
   std::unordered_map<std::string, bool> cache_state;
-  for (const auto& next_entry : blob_channel_sender_->GetCachedBlobIds()) {
+  for (const auto& next_entry : ids) {

[CJ, 2016/08/19 23:22:50]

I'd suggest renaming |next_entry| to just entry, since in a context of a loop it
can be a little confusing. 

     cache_state[next_entry.id] = next_entry.was_delivered;
   }
   response_callback.Run(std::move(cache_state));
 }
 
 void BlobChannelService::PutBlob(const std::string& id,
                                  mojo::ScopedSharedBufferHandle data,
                                  uint32_t size) {
+  DCHECK(thread_checker_.CalledOnValidThread());
+
   // Map |data| into the address space and copy out its contents.
   if (!data.is_valid()) {
     LOG(ERROR) << "Invalid data handle received from renderer process.";
     return;
   }
 
   if (size > kMaxBlobSizeBytes) {
     LOG(ERROR) << "Blob size too big: " << size;
     return;
   }
 
   mojo::ScopedSharedBufferMapping mapping = data->Map(size);
   CHECK(mapping) << "Failed to mmap region of " << size << " bytes.";
 
   scoped_refptr<BlobData> new_blob(new BlobData);
   new_blob->data.assign(reinterpret_cast<const char*>(mapping.get()), size);
-  blob_channel_sender_->PutBlob(id, std::move(new_blob));
+
+  task_runner_->PostTask(FROM_HERE,
+                         base::Bind(&BlobChannelSender::PutBlob,
+                                    blob_channel_sender_,
+                                    id,
+                                    base::Passed(std::move(new_blob))));
 }
 
 void BlobChannelService::DeliverBlob(const std::string& id) {
-  blob_channel_sender_->DeliverBlob(id);
+  DCHECK(thread_checker_.CalledOnValidThread());
+
+  task_runner_->PostTask(FROM_HERE,
+                         base::Bind(&BlobChannelSender::DeliverBlob,
+                                    blob_channel_sender_,
+                                    id));
 }
 
 void BlobChannelService::BindRequest(
     mojo::InterfaceRequest<mojom::BlobChannel> request) {
+  DCHECK(thread_checker_.CalledOnValidThread());
+
   bindings_.AddBinding(this, std::move(request));
 }
 
 }  // namespace engine
 }  // namespace blimp