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..478512714f1af5b37aeac4b16882532edaeb0373
--- /dev/null
+++ b/blimp/net/blob_channel/blob_channel_receiver.cc
@@ -0,0 +1,76 @@
+// 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::BlobChannelReceiver(BlobCache* cache,
+                                         BlobReceiverBindings* bindings)
+    : cache_(cache), bindings_(bindings) {
+  DCHECK(cache_);
+  DCHECK(bindings_);
+  bindings_->SetDelegate(this);
+}
+
+BlobChannelReceiver::~BlobChannelReceiver() {}
+
+scoped_refptr<RefCountedVector> BlobChannelReceiver::Get(
+    const std::string& 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);
+  }
+
+  if (active_read_callbacks_.find(id) == active_read_callbacks_.end()) {
+    // Only Get() from the bindings if there isn't already a request inflight.
+    bindings_->Get(id);
+  }
+
+  // Store the read callback for asynchronous completion.
+  active_read_callbacks_.insert(std::make_pair(id, callback));
+  return nullptr;
+}
+
+void BlobChannelReceiver::OnBlobReceived(
+    const std::string& id,
+    std::unique_ptr<std::vector<uint8_t>> data) {
+  scoped_refptr<RefCountedVector> cached_item(new RefCountedVector);
+  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());
+
+    cached_item->data.swap(*data);
+    cache_->Put(id, cached_item);
+
+    // 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;
+         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, cached_item);
+  }
+}
+
+}  // namespace blimp