Blimp: Add BlobChannelReceiver and bindings interface.

blimp/net/blob_channel/blob_channel_receiver.cc

Index: blimp/net/blob_channel/blob_channel_receiver.cc
diff --git a/blimp/net/blob_channel/blob_channel_receiver.cc b/blimp/net/blob_channel/blob_channel_receiver.cc
new file mode 100644
index 0000000000000000000000000000000000000000..68bc45d58e0a5fa657c1399380bd968146c8ecad
--- /dev/null
+++ b/blimp/net/blob_channel/blob_channel_receiver.cc
@@ -0,0 +1,89 @@
+// 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/net/blob_channel/blob_channel_receiver.h"
+
+#include "base/logging.h"
+#include "base/macros.h"
+#include "base/strings/string_number_conversions.h"
+#include "blimp/common/blob_cache/blob_cache.h"
+
+namespace blimp {
+
+BlobChannelReceiver::Delegate::Delegate() {}
+
+BlobChannelReceiver::Delegate::~Delegate() {}

[Wez, 2016/04/18 22:40:40]

Your API contract is that Delegate MUST out-live the BlobChannelReceiver, so you
can DCHECK(!receiver.isValid()) during destruction to ensure that.

However, you don't seem to need a WeakPtr here at all; you can just have
BlobChannelReceiver call SetReceiver(nullptr) during destruction and avoid that
extra overhead.

+
+void BlobChannelReceiver::Delegate::SetReceiver(
+    base::WeakPtr<BlobChannelReceiver> receiver) {
+  DCHECK(receiver);
+  DCHECK(!receiver_);
+  receiver_ = receiver;
+}
+
+void BlobChannelReceiver::Delegate::OnBlobReceived(const BlobId& id,
+                                                   BlobData data) {
+  if (receiver_) {
+    receiver_->OnBlobReceived(id, data);
+  }
+}
+
+BlobChannelReceiver::BlobChannelReceiver(BlobCache* cache, Delegate* delegate)
+    : cache_(cache), delegate_(delegate), weak_factory_(this) {
+  DCHECK(cache_);
+  DCHECK(delegate_);
+  delegate_->SetReceiver(weak_factory_.GetWeakPtr());
+}
+
+BlobChannelReceiver::~BlobChannelReceiver() {}
+
+scoped_refptr<RefCountedVector> BlobChannelReceiver::Get(
+    const BlobId& id,
+    const BlobReceivedCallback& callback) {
+  base::AutoLock lock(lock_);
+
+  // Return the value synchronously if the data is already available.
+  if (cache_->Contains(id)) {
+    return cache_->Get(id);

[Wez, 2016/04/18 22:40:40]

See below; like Put(), cache_->Get() seems likely to need to be either always-
or at least sometimes- asynchronous, in general?

[Wez, 2016/04/18 22:44:47]

Having this in here also means that the cache implementation needs to be
thread-clean (i.e. OK to call from different threads, even though it's not fully
thread-safe).

+  }
+
+  // Only Request() from the delegate if there isn't already a request inflight.
+  if (active_read_callbacks_.find(id) == active_read_callbacks_.end()) {
+    delegate_->Request(id);

[Wez, 2016/04/18 22:40:40]

Minimize the amount of code inside the Lock - set a local bool to indicate
whether to dispatch the request, and dispatch it after releasing the lock (this
is safe since you only ever dispatch a single request per-Id).

[Wez, 2016/04/18 22:44:47]

Note that calling delegate_->Request() here means that the Delegate is being
called on multiple threads, and that it's being called on caller's raster
threads, neither of which seem desirable - it would be cleaner to PostTask the
Request() call back to the BlobChannelReceiver's home thread.

+  }
+
+  // Store the read callback for asynchronous completion.
+  active_read_callbacks_.insert(std::make_pair(id, callback));
+  return nullptr;
+}
+
+void BlobChannelReceiver::OnBlobReceived(const BlobId& id, BlobData data) {
+  std::vector<BlobReceivedCallback> pending_callbacks;
+
+  {
+    base::AutoLock lock(lock_);
+
+    DLOG_IF(WARNING, cache_->Contains(id))
+        << "Redundant blob transfer detected: "
+        << base::HexEncode(id.data(), id.size());
+
+    cache_->Put(id, data);

[Wez, 2016/04/18 22:40:40]

In general adding a blob to the cache will be time-consuming & may involve I/O
(e.g. to persist to disk), so will be asynchronous - should we be calling Put()
while we're holding the lock?

+
+    // Gather the list of pending read callbacks for the blob |id|.
+    auto callbacks_for_id = active_read_callbacks_.equal_range(id);
+    for (auto callback_it = callbacks_for_id.first;

[Wez, 2016/04/18 22:40:40]

nit: callback_it is actually a more confusing name than e.g. it alone would be.

Why can't you just use std::vector::assign to copy the range into
pending_callbacks, though?

+         callback_it != callbacks_for_id.second; ++callback_it) {
+      pending_callbacks.push_back(callback_it->second);
+    }
+    active_read_callbacks_.erase(id);
+  }
+
+  // |lock_| is released before the callbacks are run, to prevent potential
+  // reentrant deadlocking issues.
+  for (const BlobReceivedCallback& next_callback : pending_callbacks) {
+    next_callback.Run(id, data);
+  }
+}
+
+}  // namespace blimp