Implement content::WebSocketBlobSender

content/browser/renderer_host/websocket_host.cc

 // Copyright 2013 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 "content/browser/renderer_host/websocket_host.h"
 
+#include <inttypes.h>
 #include <utility>
 
+#include "base/bind.h"
+#include "base/bind_helpers.h"
 #include "base/location.h"
+#include "base/logging.h"
 #include "base/macros.h"
-#include "base/memory/weak_ptr.h"
 #include "base/single_thread_task_runner.h"
 #include "base/strings/string_util.h"
+#include "base/strings/stringprintf.h"
 #include "base/thread_task_runner_handle.h"
+#include "content/browser/renderer_host/websocket_blob_sender.h"
 #include "content/browser/renderer_host/websocket_dispatcher_host.h"
 #include "content/browser/ssl/ssl_error_handler.h"
 #include "content/browser/ssl/ssl_manager.h"
 #include "content/common/websocket_messages.h"
+#include "content/public/browser/browser_thread.h"
 #include "content/public/browser/render_frame_host.h"
+#include "content/public/browser/storage_partition.h"
 #include "ipc/ipc_message_macros.h"
+#include "net/base/net_errors.h"
 #include "net/http/http_request_headers.h"
 #include "net/http/http_response_headers.h"
 #include "net/http/http_util.h"
 #include "net/ssl/ssl_info.h"
 #include "net/websockets/websocket_channel.h"
 #include "net/websockets/websocket_errors.h"
 #include "net/websockets/websocket_event_interface.h"
 #include "net/websockets/websocket_frame.h"  // for WebSocketFrameHeader::OpCode
 #include "net/websockets/websocket_handshake_request_info.h"
 #include "net/websockets/websocket_handshake_response_info.h"
@@ skipping 48 matching lines @@
   // for the conversion.
   static_assert(static_cast<ChannelState>(WEBSOCKET_HOST_ALIVE) ==
                     net::WebSocketEventInterface::CHANNEL_ALIVE,
                 "enum values must match for state_alive");
   static_assert(static_cast<ChannelState>(WEBSOCKET_HOST_DELETED) ==
                     net::WebSocketEventInterface::CHANNEL_DELETED,
                 "enum values must match for state_deleted");
   return static_cast<ChannelState>(host_state);
 }
 
+// Implementation of WebSocketBlobSender::Channel
+class SendChannelImpl : public WebSocketBlobSender::Channel {
+ public:
+  explicit SendChannelImpl(net::WebSocketChannel* channel)
+      : channel_(channel) {}
+
+  // Implementation of WebSocketBlobSender::Channel
+  int GetSendQuota() const override { return channel_->current_send_quota(); }
+
+  ChannelState SendFrame(bool fin, const std::vector<char>& data) override {
+    int opcode = first_frame_ ? net::WebSocketFrameHeader::kOpCodeBinary
+                              : net::WebSocketFrameHeader::kOpCodeContinuation;
+    if (first_frame_)
+      first_frame_ = false;
+    return channel_->SendFrame(fin, opcode, data);
+  }
+
+ private:
+  net::WebSocketChannel* channel_;
+  bool first_frame_ = true;
+
+  DISALLOW_COPY_AND_ASSIGN(SendChannelImpl);
+};
+
+}  // namespace
+
 // Implementation of net::WebSocketEventInterface. Receives events from our
 // WebSocketChannel object. Each event is translated to an IPC and sent to the
 // renderer or child process via WebSocketDispatcherHost.
-class WebSocketEventHandler : public net::WebSocketEventInterface {
+class WebSocketHost::WebSocketEventHandler
+    : public net::WebSocketEventInterface {
  public:
   WebSocketEventHandler(WebSocketDispatcherHost* dispatcher,
+                        WebSocketHost* host,
                         int routing_id,
                         int render_frame_id);
   ~WebSocketEventHandler() override;
 
   // net::WebSocketEventInterface implementation
 
   ChannelState OnAddChannelResponse(const std::string& selected_subprotocol,
                                     const std::string& extensions) override;
   ChannelState OnDataFrame(bool fin,
                            WebSocketMessageType type,
@@ skipping 28 matching lines @@
     void ContinueSSLRequest() override;
 
    private:
     scoped_ptr<net::WebSocketEventInterface::SSLErrorCallbacks> callbacks_;
     base::WeakPtrFactory<SSLErrorHandlerDelegate> weak_ptr_factory_;
 
     DISALLOW_COPY_AND_ASSIGN(SSLErrorHandlerDelegate);
   };
 
   WebSocketDispatcherHost* const dispatcher_;
+  WebSocketHost* const host_;
   const int routing_id_;
   const int render_frame_id_;
   scoped_ptr<SSLErrorHandlerDelegate> ssl_error_handler_delegate_;
 
   DISALLOW_COPY_AND_ASSIGN(WebSocketEventHandler);
 };
 
-WebSocketEventHandler::WebSocketEventHandler(
+WebSocketHost::WebSocketEventHandler::WebSocketEventHandler(
     WebSocketDispatcherHost* dispatcher,
+    WebSocketHost* host,
     int routing_id,
     int render_frame_id)
     : dispatcher_(dispatcher),
+      host_(host),
       routing_id_(routing_id),
-      render_frame_id_(render_frame_id) {
-}
+      render_frame_id_(render_frame_id) {}
 
-WebSocketEventHandler::~WebSocketEventHandler() {
+WebSocketHost::WebSocketEventHandler::~WebSocketEventHandler() {
   DVLOG(1) << "WebSocketEventHandler destroyed routing_id=" << routing_id_;
 }
 
-ChannelState WebSocketEventHandler::OnAddChannelResponse(
+ChannelState WebSocketHost::WebSocketEventHandler::OnAddChannelResponse(
     const std::string& selected_protocol,
     const std::string& extensions) {
   DVLOG(3) << "WebSocketEventHandler::OnAddChannelResponse"
            << " routing_id=" << routing_id_
            << " selected_protocol=\"" << selected_protocol << "\""
            << " extensions=\"" << extensions << "\"";
 
   return StateCast(dispatcher_->SendAddChannelResponse(
       routing_id_, selected_protocol, extensions));
 }
 
-ChannelState WebSocketEventHandler::OnDataFrame(
+ChannelState WebSocketHost::WebSocketEventHandler::OnDataFrame(
     bool fin,
     net::WebSocketFrameHeader::OpCode type,
     const std::vector<char>& data) {
   DVLOG(3) << "WebSocketEventHandler::OnDataFrame"
            << " routing_id=" << routing_id_ << " fin=" << fin
            << " type=" << type << " data is " << data.size() << " bytes";
 
-  return StateCast(dispatcher_->SendFrame(
-      routing_id_, fin, OpCodeToMessageType(type), data));
+  return StateCast(dispatcher_->SendFrame(routing_id_, fin,
+                                          OpCodeToMessageType(type), data));
 }
 
-ChannelState WebSocketEventHandler::OnClosingHandshake() {
+ChannelState WebSocketHost::WebSocketEventHandler::OnClosingHandshake() {
   DVLOG(3) << "WebSocketEventHandler::OnClosingHandshake"
            << " routing_id=" << routing_id_;
 
   return StateCast(dispatcher_->NotifyClosingHandshake(routing_id_));
 }
 
-ChannelState WebSocketEventHandler::OnFlowControl(int64_t quota) {
+ChannelState WebSocketHost::WebSocketEventHandler::OnFlowControl(
+    int64_t quota) {
   DVLOG(3) << "WebSocketEventHandler::OnFlowControl"
            << " routing_id=" << routing_id_ << " quota=" << quota;
 
+  if (host_->blob_sender_)
+    host_->blob_sender_->OnNewSendQuota();
   return StateCast(dispatcher_->SendFlowControl(routing_id_, quota));
 }
 
-ChannelState WebSocketEventHandler::OnDropChannel(bool was_clean,
-                                                  uint16_t code,
-                                                  const std::string& reason) {
+ChannelState WebSocketHost::WebSocketEventHandler::OnDropChannel(
+    bool was_clean,
+    uint16_t code,
+    const std::string& reason) {
   DVLOG(3) << "WebSocketEventHandler::OnDropChannel"
            << " routing_id=" << routing_id_ << " was_clean=" << was_clean
            << " code=" << code << " reason=\"" << reason << "\"";
 
   return StateCast(
       dispatcher_->DoDropChannel(routing_id_, was_clean, code, reason));
 }
 
-ChannelState WebSocketEventHandler::OnFailChannel(const std::string& message) {
+ChannelState WebSocketHost::WebSocketEventHandler::OnFailChannel(
+    const std::string& message) {
   DVLOG(3) << "WebSocketEventHandler::OnFailChannel"
-           << " routing_id=" << routing_id_
-           << " message=\"" << message << "\"";
+           << " routing_id=" << routing_id_ << " message=\"" << message << "\"";
 
   return StateCast(dispatcher_->NotifyFailure(routing_id_, message));
 }
 
-ChannelState WebSocketEventHandler::OnStartOpeningHandshake(
+ChannelState WebSocketHost::WebSocketEventHandler::OnStartOpeningHandshake(
     scoped_ptr<net::WebSocketHandshakeRequestInfo> request) {
   bool should_send = dispatcher_->CanReadRawCookies();
   DVLOG(3) << "WebSocketEventHandler::OnStartOpeningHandshake "
            << "should_send=" << should_send;
 
   if (!should_send)
     return WebSocketEventInterface::CHANNEL_ALIVE;
 
   WebSocketHandshakeRequest request_to_pass;
   request_to_pass.url.Swap(&request->url);
   net::HttpRequestHeaders::Iterator it(request->headers);
   while (it.GetNext())
     request_to_pass.headers.push_back(std::make_pair(it.name(), it.value()));
   request_to_pass.headers_text =
       base::StringPrintf("GET %s HTTP/1.1\r\n",
                          request_to_pass.url.spec().c_str()) +
       request->headers.ToString();
   request_to_pass.request_time = request->request_time;
 
-  return StateCast(dispatcher_->NotifyStartOpeningHandshake(routing_id_,
-                                                            request_to_pass));
+  return StateCast(
+      dispatcher_->NotifyStartOpeningHandshake(routing_id_, request_to_pass));
 }
 
-ChannelState WebSocketEventHandler::OnFinishOpeningHandshake(
+ChannelState WebSocketHost::WebSocketEventHandler::OnFinishOpeningHandshake(
     scoped_ptr<net::WebSocketHandshakeResponseInfo> response) {
   bool should_send = dispatcher_->CanReadRawCookies();
   DVLOG(3) << "WebSocketEventHandler::OnFinishOpeningHandshake "
            << "should_send=" << should_send;
 
   if (!should_send)
     return WebSocketEventInterface::CHANNEL_ALIVE;
 
   WebSocketHandshakeResponse response_to_pass;
   response_to_pass.url.Swap(&response->url);
   response_to_pass.status_code = response->status_code;
   response_to_pass.status_text.swap(response->status_text);
   void* iter = NULL;
   std::string name, value;
   while (response->headers->EnumerateHeaderLines(&iter, &name, &value))
     response_to_pass.headers.push_back(std::make_pair(name, value));
   response_to_pass.headers_text =
       net::HttpUtil::ConvertHeadersBackToHTTPResponse(
           response->headers->raw_headers());
   response_to_pass.response_time = response->response_time;
 
-  return StateCast(dispatcher_->NotifyFinishOpeningHandshake(routing_id_,
-                                                             response_to_pass));
+  return StateCast(
+      dispatcher_->NotifyFinishOpeningHandshake(routing_id_, response_to_pass));
 }
 
-ChannelState WebSocketEventHandler::OnSSLCertificateError(
+ChannelState WebSocketHost::WebSocketEventHandler::OnSSLCertificateError(
     scoped_ptr<net::WebSocketEventInterface::SSLErrorCallbacks> callbacks,
     const GURL& url,
     const net::SSLInfo& ssl_info,
     bool fatal) {
   DVLOG(3) << "WebSocketEventHandler::OnSSLCertificateError"
            << " routing_id=" << routing_id_ << " url=" << url.spec()
            << " cert_status=" << ssl_info.cert_status << " fatal=" << fatal;
   ssl_error_handler_delegate_.reset(
       new SSLErrorHandlerDelegate(std::move(callbacks)));
   SSLManager::OnSSLCertificateSubresourceError(
       ssl_error_handler_delegate_->GetWeakPtr(), url,
       dispatcher_->render_process_id(), render_frame_id_, ssl_info, fatal);
   // The above method is always asynchronous.
   return WebSocketEventInterface::CHANNEL_ALIVE;
 }
 
-WebSocketEventHandler::SSLErrorHandlerDelegate::SSLErrorHandlerDelegate(
-    scoped_ptr<net::WebSocketEventInterface::SSLErrorCallbacks> callbacks)
+WebSocketHost::WebSocketEventHandler::SSLErrorHandlerDelegate::
+    SSLErrorHandlerDelegate(
+        scoped_ptr<net::WebSocketEventInterface::SSLErrorCallbacks> callbacks)
     : callbacks_(std::move(callbacks)), weak_ptr_factory_(this) {}
 
-WebSocketEventHandler::SSLErrorHandlerDelegate::~SSLErrorHandlerDelegate() {}
+WebSocketHost::WebSocketEventHandler::SSLErrorHandlerDelegate::
+    ~SSLErrorHandlerDelegate() {}
 
 base::WeakPtr<SSLErrorHandler::Delegate>
-WebSocketEventHandler::SSLErrorHandlerDelegate::GetWeakPtr() {
+WebSocketHost::WebSocketEventHandler::SSLErrorHandlerDelegate::GetWeakPtr() {
   return weak_ptr_factory_.GetWeakPtr();
 }
 
-void WebSocketEventHandler::SSLErrorHandlerDelegate::CancelSSLRequest(
-    int error,
-    const net::SSLInfo* ssl_info) {
+void WebSocketHost::WebSocketEventHandler::SSLErrorHandlerDelegate::
+    CancelSSLRequest(int error, const net::SSLInfo* ssl_info) {
   DVLOG(3) << "SSLErrorHandlerDelegate::CancelSSLRequest"
            << " error=" << error
            << " cert_status=" << (ssl_info ? ssl_info->cert_status
                                            : static_cast<net::CertStatus>(-1));
   callbacks_->CancelSSLRequest(error, ssl_info);
 }
 
-void WebSocketEventHandler::SSLErrorHandlerDelegate::ContinueSSLRequest() {
+void WebSocketHost::WebSocketEventHandler::SSLErrorHandlerDelegate::
+    ContinueSSLRequest() {
   DVLOG(3) << "SSLErrorHandlerDelegate::ContinueSSLRequest";
   callbacks_->ContinueSSLRequest();
 }
 
-}  // namespace
-
 WebSocketHost::WebSocketHost(int routing_id,
                              WebSocketDispatcherHost* dispatcher,
                              net::URLRequestContext* url_request_context,
                              base::TimeDelta delay)
     : dispatcher_(dispatcher),
       url_request_context_(url_request_context),
       routing_id_(routing_id),
       delay_(delay),
       pending_flow_control_quota_(0),
       handshake_succeeded_(false),
       weak_ptr_factory_(this) {
   DVLOG(1) << "WebSocketHost: created routing_id=" << routing_id;
 }
 
 WebSocketHost::~WebSocketHost() {}
 
 void WebSocketHost::GoAway() {
   OnDropChannel(false, static_cast<uint16_t>(net::kWebSocketErrorGoingAway),
                 "");
 }
 
 bool WebSocketHost::OnMessageReceived(const IPC::Message& message) {
   bool handled = true;
   IPC_BEGIN_MESSAGE_MAP(WebSocketHost, message)
     IPC_MESSAGE_HANDLER(WebSocketHostMsg_AddChannelRequest, OnAddChannelRequest)
+    IPC_MESSAGE_HANDLER(WebSocketHostMsg_SendBlob, OnSendBlob)
     IPC_MESSAGE_HANDLER(WebSocketMsg_SendFrame, OnSendFrame)
     IPC_MESSAGE_HANDLER(WebSocketMsg_FlowControl, OnFlowControl)
     IPC_MESSAGE_HANDLER(WebSocketMsg_DropChannel, OnDropChannel)
     IPC_MESSAGE_UNHANDLED(handled = false)
   IPC_END_MESSAGE_MAP()
   return handled;
 }
 
 void WebSocketHost::OnAddChannelRequest(
     const GURL& socket_url,
@@ skipping 26 matching lines @@
     int render_frame_id) {
   DVLOG(3) << "WebSocketHost::AddChannel"
            << " routing_id=" << routing_id_ << " socket_url=\"" << socket_url
            << "\" requested_protocols=\""
            << base::JoinString(requested_protocols, ", ") << "\" origin=\""
            << origin << "\"";
 
   DCHECK(!channel_);
 
   scoped_ptr<net::WebSocketEventInterface> event_interface(
-      new WebSocketEventHandler(dispatcher_, routing_id_, render_frame_id));
+      new WebSocketEventHandler(dispatcher_, this, routing_id_,
+                                render_frame_id));
   channel_.reset(new net::WebSocketChannel(std::move(event_interface),
                                            url_request_context_));
 
   if (pending_flow_control_quota_ > 0) {
     // channel_->SendFlowControl(pending_flow_control_quota_) must be called
     // after channel_->SendAddChannelRequest() below.
     // We post OnFlowControl() here using |weak_ptr_factory_| instead of
     // calling SendFlowControl directly, because |this| may have been deleted
     // after channel_->SendAddChannelRequest().
     base::ThreadTaskRunnerHandle::Get()->PostTask(
         FROM_HERE, base::Bind(&WebSocketHost::OnFlowControl,
                               weak_ptr_factory_.GetWeakPtr(),
                               pending_flow_control_quota_));
     pending_flow_control_quota_ = 0;
   }
 
   channel_->SendAddChannelRequest(socket_url, requested_protocols, origin);
   // |this| may have been deleted here.
 }
 
+void WebSocketHost::OnSendBlob(const std::string& uuid,
+                               uint64_t expected_size) {
+  DVLOG(3) << "WebSocketHost::OnSendBlob"
+           << " routing_id=" << routing_id_ << " uuid=" << uuid
+           << " expected_size=" << expected_size;
+
+  DCHECK(channel_);
+  DCHECK(!blob_sender_);
+  blob_sender_.reset(new WebSocketBlobSender(
+      make_scoped_ptr(new SendChannelImpl(channel_.get()))));
+  StoragePartition* partition = dispatcher_->storage_partition();
+  storage::FileSystemContext* file_system_context =
+      partition->GetFileSystemContext();
+
+  net::WebSocketEventInterface::ChannelState channel_state =
+      net::WebSocketEventInterface::CHANNEL_ALIVE;
+
+  // This use of base::Unretained is safe because the WebSocketBlobSender object
+  // is owned by this object and will not call it back after destruction.
+  int rv = blob_sender_->Start(
+      uuid, expected_size, dispatcher_->blob_storage_context(),
+      file_system_context,
+      BrowserThread::GetMessageLoopProxyForThread(BrowserThread::FILE).get(),
+      &channel_state,
+      base::Bind(&WebSocketHost::BlobSendComplete, base::Unretained(this)));
+  if (channel_state == net::WebSocketEventInterface::CHANNEL_ALIVE &&
+      rv != net::ERR_IO_PENDING)
+    BlobSendComplete(rv);
+  // |this| may be destroyed here.
+}
+
 void WebSocketHost::OnSendFrame(bool fin,
                                 WebSocketMessageType type,
                                 const std::vector<char>& data) {
   DVLOG(3) << "WebSocketHost::OnSendFrame"
            << " routing_id=" << routing_id_ << " fin=" << fin
            << " type=" << type << " data is " << data.size() << " bytes";
 
   DCHECK(channel_);
+  if (blob_sender_) {
+    // TODO(ricea): Kill the renderer.
+    return;
+  }
   channel_->SendFrame(fin, MessageTypeToOpCode(type), data);
 }
 
 void WebSocketHost::OnFlowControl(int64_t quota) {
   DVLOG(3) << "WebSocketHost::OnFlowControl"
            << " routing_id=" << routing_id_ << " quota=" << quota;
 
   if (!channel_) {
     // WebSocketChannel is not yet created due to the delay introduced by
     // per-renderer WebSocket throttling.
     // SendFlowControl() is called after WebSocketChannel is created.
     pending_flow_control_quota_ += quota;
     return;
   }
 
   channel_->SendFlowControl(quota);
 }
 
 void WebSocketHost::OnDropChannel(bool was_clean,
                                   uint16_t code,
                                   const std::string& reason) {
   DVLOG(3) << "WebSocketHost::OnDropChannel"
            << " routing_id=" << routing_id_ << " was_clean=" << was_clean
            << " code=" << code << " reason=\"" << reason << "\"";
 
   if (!channel_) {
     // WebSocketChannel is not yet created due to the delay introduced by
     // per-renderer WebSocket throttling.
     WebSocketDispatcherHost::WebSocketHostState result =
-        dispatcher_->DoDropChannel(routing_id_,
-                                   false,
-                                   net::kWebSocketErrorAbnormalClosure,
-                                   "");
+        dispatcher_->DoDropChannel(routing_id_, false,
+                                   net::kWebSocketErrorAbnormalClosure, "");
     DCHECK_EQ(WebSocketDispatcherHost::WEBSOCKET_HOST_DELETED, result);
     return;
   }
 
+  blob_sender_.reset();
   // TODO(yhirano): Handle |was_clean| appropriately.
   channel_->StartClosingHandshake(code, reason);
 }
 
+void WebSocketHost::BlobSendComplete(int result) {
+  DVLOG(3) << "WebSocketHost::BlobSendComplete"
+           << " routing_id=" << routing_id_
+           << " result=" << net::ErrorToString(result);
+
+  // All paths through this method must reset blob_sender_, so take ownership
+  // at the beginning.
+  scoped_ptr<WebSocketBlobSender> blob_sender;

[dcheng, 2016/01/20 08:30:06]

scoped_ptr<WebSocketBlobSender> blob_sender(std::move(blob_sender_)); is another
alternative (I don't have a strong preference over which one is better, but I
think this version might fit on one line =)

[Adam Rice, 2016/01/20 17:11:21]

Done.

+  blob_sender.swap(blob_sender_);
+  switch (result) {
+    case net::OK:
+      ignore_result(dispatcher_->BlobSendComplete(routing_id_));

[dcheng, 2016/01/20 08:30:06]

Can you help me understand why it's OK to ignore failures here?

[Adam Rice, 2016/01/20 17:11:21]

A failure here would mean the IPC failed to be sent to the renderer, in which
case we will have already
destroyed the channel including this object here.

If we needed to avoid touching member variables when |this| was deleted we could
check for CHANNEL_DELETED here. But in practice we always just return to the
event loop regardless, so there is no point in doing a check.

If you want the really long version, there's actually a design doc:
https://docs.google.com/document/d/1fMxcEFiVj-H5vmuxhPqsxJAl98GS6_yZ-wqB6j9Wm...

ChannelState may be the only enum in Chromium that has its own design doc.
Although it doesn't mention
the enum by name and is almost entirely concerned with ancient problems that no
longer exist since we
added the enum.

+      // |this| may be destroyed here.
+      return;
+
+    case net::ERR_UPLOAD_FILE_CHANGED: {
+      uint64_t expected_size = blob_sender->expected_size();
+      uint64_t actual_size = blob_sender->ActualSize();
+      if (expected_size != actual_size) {
+        ignore_result(dispatcher_->NotifyFailure(
+            routing_id_,
+            base::StringPrintf("Blob size mismatch; renderer size = %" PRIu64
+                               ", browser size = %" PRIu64,
+                               expected_size, actual_size)));
+        // |this| is destroyed here.
+        return;
+      }  // else fallthrough
+    }
+
+    default:
+      ignore_result(dispatcher_->NotifyFailure(
+          routing_id_, std::string("Failed to load Blob: error code = ") +

[dcheng, 2016/01/20 08:30:06]

Is the explicit string conversion here needed?

[Adam Rice, 2016/01/20 17:11:21]

No. Removed.

+                           net::ErrorToString(result)));
+      // |this| is destroyed here.
+      return;
+  }
+}
+
 }  // namespace content