[OBSOLETE] Browser-side implementation of WebSocket Blob receive.

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 <utility>
 
 #include "base/location.h"
 #include "base/macros.h"
@@ skipping 135 matching lines @@
 
   DISALLOW_COPY_AND_ASSIGN(WebSocketEventHandler);
 };
 
 WebSocketEventHandler::WebSocketEventHandler(
     WebSocketDispatcherHost* dispatcher,
     int routing_id,
     int render_frame_id)
     : dispatcher_(dispatcher),
       routing_id_(routing_id),
-      render_frame_id_(render_frame_id) {
-}
+      render_frame_id_(render_frame_id) {}
 
 WebSocketEventHandler::~WebSocketEventHandler() {
   DVLOG(1) << "WebSocketEventHandler destroyed routing_id=" << routing_id_;
 }
 
 ChannelState WebSocketEventHandler::OnAddChannelResponse(
     const std::string& selected_protocol,
     const std::string& extensions) {
   DVLOG(3) << "WebSocketEventHandler::OnAddChannelResponse"
-           << " routing_id=" << routing_id_
-           << " selected_protocol=\"" << selected_protocol << "\""
+           << " routing_id=" << routing_id_ << " selected_protocol=\""
+           << selected_protocol << "\""
            << " extensions=\"" << extensions << "\"";
 
   return StateCast(dispatcher_->SendAddChannelResponse(
       routing_id_, selected_protocol, extensions));
 }
 
 ChannelState 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() {
   DVLOG(3) << "WebSocketEventHandler::OnClosingHandshake"
            << " routing_id=" << routing_id_;
 
   return StateCast(dispatcher_->NotifyClosingHandshake(routing_id_));
 }
 
 ChannelState WebSocketEventHandler::OnFlowControl(int64_t quota) {
   DVLOG(3) << "WebSocketEventHandler::OnFlowControl"
            << " routing_id=" << routing_id_ << " quota=" << quota;
 
   return StateCast(dispatcher_->SendFlowControl(routing_id_, quota));
 }
 
 ChannelState 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) {
   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(
     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(
     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(
     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;
+    return WebSocketEventInterface::CHANNEL_ALIVE;
 }
 
 WebSocketEventHandler::SSLErrorHandlerDelegate::SSLErrorHandlerDelegate(
     scoped_ptr<net::WebSocketEventInterface::SSLErrorCallbacks> callbacks)
     : callbacks_(std::move(callbacks)), weak_ptr_factory_(this) {}
 
 WebSocketEventHandler::SSLErrorHandlerDelegate::~SSLErrorHandlerDelegate() {}
 
 base::WeakPtr<SSLErrorHandler::Delegate>
 WebSocketEventHandler::SSLErrorHandlerDelegate::GetWeakPtr() {
@@ skipping 10 matching lines @@
   callbacks_->CancelSSLRequest(error, ssl_info);
 }
 
 void WebSocketEventHandler::SSLErrorHandlerDelegate::ContinueSSLRequest() {
   DVLOG(3) << "SSLErrorHandlerDelegate::ContinueSSLRequest";
   callbacks_->ContinueSSLRequest();
 }
 
 }  // namespace
 
+WebSocketHost::ReceiveQuotaMultiplexer::ReceiveQuotaMultiplexer()
+    : channel_(nullptr), published_quota_(0), current_consumer_(nullptr) {}
+
+// Quota can be added to any consumer at any time.
+bool WebSocketHost::ReceiveQuotaMultiplexer::AddQuota(
+    ReceiveQuotaProvider* provider,
+    int64_t quota) {
+  DCHECK(channel_);
+  DCHECK_GE(quota, 0);
+  *provider += quota;
+  if (provider == current_consumer_) {
+    PublishMoreQuotaIfNeeded();
+    return true;
+  }
+  return false;
+}
+
+// Quota can only be used from the current consumer.
+int64_t WebSocketHost::ReceiveQuotaMultiplexer::AvailableQuota() const {
+  DCHECK(current_consumer_);
+  return *current_consumer_;
+}
+
+// Quota is only consumed by the current consumer.
+void WebSocketHost::ReceiveQuotaMultiplexer::ConsumedQuota(int64_t quota) {
+  DCHECK_GE(quota, 0);
+  DCHECK_GE(*current_consumer_, quota);
+  DCHECK_GE(published_quota_, *current_consumer_);
+  *current_consumer_ -= quota;
+  published_quota_ -= quota;
+}
+
+// Changing the current consumer may result in more quota being published.
+void WebSocketHost::ReceiveQuotaMultiplexer::SetConsumer(
+    ReceiveQuotaConsumer* consumer) {
+  DCHECK(channel_);
+  DCHECK(consumer);
+  current_consumer_ = consumer;
+  PublishMoreQuotaIfNeeded();
+}
+
+void WebSocketHost::ReceiveQuotaMultiplexer::PublishMoreQuotaIfNeeded() {
+  DCHECK(channel_);
+  DCHECK(current_consumer_);
+  if (*current_consumer_ > published_quota_) {
+    channel_->SendFlowControl(published_quota_ - *current_consumer_);
+    published_quota_ = *current_consumer_;
+  }
+}
+
 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),
@@ skipping 122 matching lines @@
                                    "");
     DCHECK_EQ(WebSocketDispatcherHost::WEBSOCKET_HOST_DELETED, result);
     return;
   }
 
   // TODO(yhirano): Handle |was_clean| appropriately.
   channel_->StartClosingHandshake(code, reason);
 }
 
 }  // namespace content