[OBSOLETE][DO NOT SUBMIT][DO NOT COMMIT]] Browser-side implementation of WebSocket Blob receive.

content/browser/renderer_host/websocket_blob_receiver.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 "websocket_blob_receiver.h"
+
+#include <stddef.h>
+#include <string.h>
+#include <ostream>
+#include <utility>
+
+#include "base/bind.h"
+#include "base/bind_helpers.h"
+#include "base/files/file_util.h"
+#include "base/guid.h"
+#include "base/location.h"
+#include "base/logging.h"
+#include "base/message_loop/message_loop.h"
+#include "base/numerics/safe_conversions.h"
+#include "base/single_thread_task_runner.h"
+#include "content/browser/loader/temporary_file_stream.h"
+#include "content/public/browser/browser_thread.h"
+#include "net/base/file_stream.h"
+#include "net/base/io_buffer.h"
+#include "net/base/net_errors.h"
+#include "storage/browser/blob/blob_data_builder.h"
+#include "storage/browser/blob/blob_storage_context.h"
+#include "storage/browser/blob/shareable_file_reference.h"
+
+namespace content {
+
+namespace {
+// This should be the same as the maximum receive quota provided by the
+// renderer, otherwise WebSocketHost will have to do unnecessary buffering.
+const size_t kBufferSize = 64 * 1024;
+}
+
+// Needed to make CHECK_EQ(), etc. work
+std::ostream& operator<<(std::ostream& os, WebSocketBlobReceiver::State state) {
+  static const char* const kStateStrings[] = {
+      "NONE",
+      "CREATE_FILE",
+      "CREATE_FILE_COMPLETE",
+      "SEND_QUOTA",
+      "WRITE",
+      "WRITE_COMPLETE",
+      "GET_INFO",
+      "GET_INFO_COMPLETE",
+  };
+
+  if (state < WebSocketBlobReceiver::State::NONE ||
+      state > WebSocketBlobReceiver::State::GET_INFO_COMPLETE) {
+    return os << "Bad state (" << static_cast<int>(state) << ")";
+  }
+  return os << kStateStrings[static_cast<int>(state)];
+}
+
+// Helper object to call GetFileInfo() on the FILE thread
+class WebSocketBlobReceiver::FileInfoHelper {
+ public:
+  FileInfoHelper(const base::WeakPtr<WebSocketBlobReceiver>& owner)
+      : owner_(owner) {}
+
+  void GetFileInfo(const base::FilePath& path) {
+    result_ = base::GetFileInfo(path, &info_);
+  }
+
+  void DidGetInfo() {
+    if (owner_)
+      owner_->DidGetInfo(result_, info_);
+  }
+
+ private:
+  const base::WeakPtr<WebSocketBlobReceiver> owner_;
+  bool result_ = false;
+  base::File::Info info_;
+
+  DISALLOW_COPY_AND_ASSIGN(FileInfoHelper);
+};
+
+WebSocketBlobReceiver::WebSocketBlobReceiver(
+    scoped_ptr<Client> client,
+    storage::BlobStorageContext* blob_storage_context)
+    : client_(std::move(client)),
+      blob_storage_context_(blob_storage_context),
+      io_buffer_(new net::DrainableIOBuffer(new net::IOBuffer(kBufferSize),
+                                            kBufferSize)),
+      pending_quota_(kBufferSize),
+      weak_factory_(this) {
+  DCHECK(blob_storage_context_);
+  io_buffer_->SetOffset(kBufferSize);
+}
+
+WebSocketBlobReceiver::~WebSocketBlobReceiver() {}
+
+void WebSocketBlobReceiver::Start() {
+  // This doesn't limit the space used by a single origin or take into account
+  // incognito mode. TODO(ricea): Apply quota and take into account incognito
+  // mode.
+  next_state_ = State::CREATE_FILE;
+  int rv = DoLoop(net::OK);
+  DCHECK_EQ(net::ERR_IO_PENDING, rv);
+}
+
+int WebSocketBlobReceiver::AppendData(const std::vector<char>& data) {
+  CHECK_LE(data.size(), kBufferSize);
+  DCHECK_NE(next_state_, State::NONE);
+  if (data.size() == 0)
+    return net::OK;
+  if (io_in_progress_) {
+    pending_data_.insert(pending_data_.end(), data.begin(), data.end());
+    return net::ERR_IO_PENDING;
+  }
+  PrepareWrite(data);
+  return DoLoop(net::OK);
+}
+
+int WebSocketBlobReceiver::Finish() {
+  DCHECK(!finish_called_);
+  finish_called_ = true;
+  DCHECK_NE(next_state_, State::NONE);
+  DCHECK_NE(next_state_, State::GET_INFO);
+  DCHECK_NE(next_state_, State::GET_INFO_COMPLETE);
+  if (io_in_progress_)
+    return net::ERR_IO_PENDING;
+  return DoLoop(net::OK);
+}
+
+int WebSocketBlobReceiver::DoLoop(int result) {
+  int rv = result;
+  do {
+    State state = next_state_;
+    next_state_ = State::NONE;
+    switch (state) {
+      case State::CREATE_FILE:
+        DCHECK_EQ(net::OK, rv);
+        rv = DoCreateFile();
+        break;
+
+      case State::CREATE_FILE_COMPLETE:
+        rv = DoCreateFileComplete(rv);
+        break;
+
+      case State::SEND_QUOTA:
+        DCHECK_EQ(net::OK, rv);
+        rv = DoSendQuota();
+        break;
+
+      case State::WRITE:
+        DCHECK_EQ(net::OK, rv);
+        rv = DoWrite();
+        break;
+
+      case State::WRITE_COMPLETE:
+        rv = DoWriteComplete(rv);
+        break;
+
+      case State::GET_INFO:
+        DCHECK_EQ(net::OK, rv);
+        rv = DoGetInfo();
+        break;
+
+      case State::GET_INFO_COMPLETE:
+        rv = DoGetInfoComplete(rv);
+        break;
+
+      default:
+        NOTREACHED();
+    }
+  } while (rv != net::ERR_IO_PENDING && next_state_ != State::NONE);
+  return rv;
+}
+
+void WebSocketBlobReceiver::DoLoopAsync(int result) {
+  int rv = DoLoop(result);
+  if (rv == net::ERR_IO_PENDING)
+    return;
+  if (rv < 0)
+    client_->BlobFailed(rv);
+}
+
+int WebSocketBlobReceiver::DoCreateFile() {
+  next_state_ = State::CREATE_FILE_COMPLETE;
+  io_in_progress_ = true;
+  CreateTemporaryFileStream(
+      base::Bind(&WebSocketBlobReceiver::DidCreateTemporaryFileStream,
+                 weak_factory_.GetWeakPtr()));
+  return net::ERR_IO_PENDING;
+}
+
+int WebSocketBlobReceiver::DoCreateFileComplete(int result) {
+  io_in_progress_ = false;
+  if (result < 0)
+    return result;
+  next_state_ = State::SEND_QUOTA;
+  return result;
+}
+
+int WebSocketBlobReceiver::DoSendQuota() {
+  next_state_ = State::WRITE;
+  if (finish_called_)
+    return net::OK;
+  DCHECK_GT(pending_quota_, 0u);
+  base::MessageLoop::current()->PostTask(
+      FROM_HERE, base::Bind(&WebSocketBlobReceiver::DoSendQuotaAsync,
+                            weak_factory_.GetWeakPtr(), pending_quota_));
+  pending_quota_ = 0u;
+  return net::ERR_IO_PENDING;
+}
+
+int WebSocketBlobReceiver::DoWrite() {
+  next_state_ = State::WRITE_COMPLETE;
+  io_in_progress_ = true;
+  if (io_buffer_->BytesRemaining() == 0)
+    return net::OK;
+  // This use of base::Unretained() is safe because |file_stream_| will cancel
+  // writing on destruction, and its lifetime is tied to
+  // |deletable_file_|. |deletable_file_| was created by this object, and
+  // remains solely owned by this object until BlobCreated() is called.
+  // BlobCreated() will not be called while |io_in_progress_| is true.
+  return file_stream_->Write(io_buffer_.get(), io_buffer_->BytesRemaining(),
+                             base::Bind(&WebSocketBlobReceiver::OnWriteComplete,
+                                        base::Unretained(this)));
+}
+
+int WebSocketBlobReceiver::DoWriteComplete(int result) {
+  io_in_progress_ = false;
+  if (result < 0)
+    return result;
+  if (result > 0)
+    io_buffer_->DidConsume(result);
+  if (io_buffer_->BytesRemaining() > 0) {
+    next_state_ = State::WRITE;
+    return net::OK;
+  }
+  if (!pending_data_.empty()) {
+    next_state_ = State::WRITE;
+    std::vector<char> pending_data;
+    pending_data.swap(pending_data_);
+    PrepareWrite(pending_data);
+    return net::OK;
+  }
+  if (finish_called_) {
+    next_state_ = State::GET_INFO;
+    return net::OK;
+  }
+  DCHECK_GT(pending_quota_, 0u);
+  next_state_ = State::SEND_QUOTA;
+  return net::OK;
+}
+
+int WebSocketBlobReceiver::DoGetInfo() {
+  next_state_ = State::GET_INFO_COMPLETE;
+  scoped_refptr<base::SingleThreadTaskRunner> file_thread =
+      BrowserThread::GetMessageLoopProxyForThread(BrowserThread::FILE);
+  FileInfoHelper* file_info_helper =
+      new FileInfoHelper(weak_factory_.GetWeakPtr());
+  file_thread->PostTaskAndReply(
+      FROM_HERE,
+      base::Bind(&FileInfoHelper::GetFileInfo,
+                 base::Unretained(file_info_helper), deletable_file_->path()),
+      base::Bind(&FileInfoHelper::DidGetInfo, base::Owned(file_info_helper)));
+  return net::ERR_IO_PENDING;
+}
+
+int WebSocketBlobReceiver::DoGetInfoComplete(int result) {
+  return result;
+}
+
+void WebSocketBlobReceiver::OnWriteComplete(int result) {
+  DoLoopAsync(result);
+  // |this| may be destroyed here.
+}
+
+void WebSocketBlobReceiver::DidCreateTemporaryFileStream(
+    base::File::Error error_code,
+    scoped_ptr<net::FileStream> file_stream,
+    storage::ShareableFileReference* deletable_file) {
+  if (error_code == base::File::FILE_OK) {
+    file_stream_.swap(file_stream);
+    deletable_file_ = deletable_file;
+  }
+  DoLoopAsync(net::FileErrorToNetError(error_code));
+  // |this| may be destroyed here.
+}
+
+void WebSocketBlobReceiver::PrepareWrite(const std::vector<char>& data) {
+  CHECK_LE(data.size(), kBufferSize);
+  pending_quota_ += data.size();
+  io_buffer_->SetOffset(kBufferSize - data.size());
+  memcpy(io_buffer_->data(), data.data(), data.size());
+}
+
+void WebSocketBlobReceiver::DidGetInfo(bool result,
+                                       const base::File::Info& info) {
+  if (result) {
+    storage::BlobDataBuilder builder(base::GenerateGUID());
+    builder.AppendFile(deletable_file_->path(), UINT64_C(0),
+                       base::checked_cast<uint64_t>(info.size),
+                       info.last_modified);
+    scoped_ptr<storage::BlobDataHandle> blob_data_handle(
+        blob_storage_context_->AddFinishedBlob(builder));
+    if (!blob_data_handle) {
+      DoLoopAsync(net::ERR_OUT_OF_MEMORY);
+      // |this| may be destroyed here.
+      return;
+    }
+    client_->BlobCreated(std::move(blob_data_handle), info.size);
+    // |this| may be destroyed here.
+    return;
+  }
+  DoLoopAsync(net::ERR_FILE_NOT_FOUND);
+  // |this| may be destroyed here.
+}
+
+// This is called asynchronously rather than from the DoLoop() because it may
+// destroy |this|.
+void WebSocketBlobReceiver::DoSendQuotaAsync(size_t quota) {
+  client_->AddFlowControlQuota(quota);
+}
+
+}  // namespace content