Refactor WebSocket throttling feature.

net/socket_stream/socket_stream.cc

 // Copyright (c) 2010 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.
 //
 // TODO(ukai): code is similar with http_network_transaction.cc.  We should
 //   think about ways to share code, if possible.
 
 #include "net/socket_stream/socket_stream.h"
 
 #include <string>
 
 #include "base/compiler_specific.h"
 #include "base/logging.h"
 #include "base/message_loop.h"
 #include "base/string_util.h"
 #include "net/base/auth.h"
 #include "net/base/host_resolver.h"
 #include "net/base/io_buffer.h"
 #include "net/base/net_errors.h"
 #include "net/base/net_util.h"
 #include "net/http/http_response_headers.h"
 #include "net/http/http_util.h"
 #include "net/socket/client_socket_factory.h"
 #include "net/socket/ssl_client_socket.h"
 #include "net/socket/socks5_client_socket.h"
 #include "net/socket/socks_client_socket.h"
 #include "net/socket/tcp_client_socket.h"
 #include "net/socket_stream/socket_stream_metrics.h"
-#include "net/socket_stream/socket_stream_throttle.h"
 #include "net/url_request/url_request.h"
 
 static const int kMaxPendingSendAllowed = 32768;  // 32 kilobytes.
 static const int kReadBufferSize = 4096;
 
 namespace net {
 
 void SocketStream::ResponseHeaders::Realloc(size_t new_size) {
   headers_.reset(static_cast<char*>(realloc(headers_.release(), new_size)));
 }
@@ skipping 12 matching lines @@
           io_callback_(this, &SocketStream::OnIOCompleted)),
       ALLOW_THIS_IN_INITIALIZER_LIST(
           read_callback_(this, &SocketStream::OnReadCompleted)),
       ALLOW_THIS_IN_INITIALIZER_LIST(
           write_callback_(this, &SocketStream::OnWriteCompleted)),
       read_buf_(NULL),
       write_buf_(NULL),
       current_write_buf_(NULL),
       write_buf_offset_(0),
       write_buf_size_(0),
-      throttle_(
-          SocketStreamThrottle::GetSocketStreamThrottleForScheme(
-              url.scheme())),
       metrics_(new SocketStreamMetrics(url)) {
   DCHECK(MessageLoop::current()) <<
       "The current MessageLoop must exist";
   DCHECK_EQ(MessageLoop::TYPE_IO, MessageLoop::current()->type()) <<
       "The current MessageLoop must be TYPE_IO";
   DCHECK(delegate_);
-  DCHECK(throttle_);
 }
 
 SocketStream::~SocketStream() {
   set_context(NULL);
   DCHECK(!delegate_);
 }
 
 SocketStream::UserData* SocketStream::GetUserData(
     const void* key) const {
   UserDataMap::const_iterator found = user_data_.find(key);
@@ skipping 146 matching lines @@
   DLOG(INFO) << "Finish result=" << net::ErrorToString(result);
   if (delegate_)
     delegate_->OnError(this, result);
 
   metrics_->OnClose();
   Delegate* delegate = delegate_;
   delegate_ = NULL;
   if (delegate) {
     delegate->OnClose(this);
   }
-  throttle_->OnClose(this);
   Release();
 }
 
 void SocketStream::SetHostResolver(HostResolver* host_resolver) {
   DCHECK(host_resolver);
   host_resolver_ = host_resolver;
 }
 
 void SocketStream::SetClientSocketFactory(
     ClientSocketFactory* factory) {
@@ skipping 19 matching lines @@
 
   return OK;
 }
 
 int SocketStream::DidReceiveData(int result) {
   DCHECK(read_buf_);
   DCHECK_GT(result, 0);
   net_log_.AddEvent(NetLog::TYPE_SOCKET_STREAM_RECEIVED);
   int len = result;
   metrics_->OnRead(len);
-  result = throttle_->OnRead(this, read_buf_->data(), len, &io_callback_);
   if (delegate_) {
     // Notify recevied data to delegate.
     delegate_->OnReceivedData(this, read_buf_->data(), len);
   }
   read_buf_ = NULL;
-  return result;
+  return OK;
 }
 
 int SocketStream::DidSendData(int result) {
   DCHECK_GT(result, 0);
   net_log_.AddEvent(NetLog::TYPE_SOCKET_STREAM_SENT);
   int len = result;
   metrics_->OnWrite(len);
-  result = throttle_->OnWrite(this, current_write_buf_->data(), len,
-                              &io_callback_);
   current_write_buf_ = NULL;
   if (delegate_)
     delegate_->OnSentData(this, len);
 
   int remaining_size = write_buf_size_ - write_buf_offset_ - len;
   if (remaining_size == 0) {
     if (!pending_write_bufs_.empty()) {
       write_buf_size_ = pending_write_bufs_.front()->size();
       write_buf_ = pending_write_bufs_.front();
       pending_write_bufs_.pop_front();
     } else {
       write_buf_size_ = 0;
       write_buf_ = NULL;
     }
     write_buf_offset_ = 0;
   } else {
     write_buf_offset_ += len;
   }
-  return result;
+  return OK;
 }
 
 void SocketStream::OnIOCompleted(int result) {
   DoLoop(result);
 }
 
 void SocketStream::OnReadCompleted(int result) {
   if (result == 0) {
     // 0 indicates end-of-file, so socket was closed.
     next_state_ = STATE_CLOSE;
@@ skipping 160 matching lines @@
 
   DCHECK(host_resolver_.get());
   resolver_.reset(new SingleRequestHostResolver(host_resolver_.get()));
   return resolver_->Resolve(resolve_info, &addresses_, &io_callback_,
                             net_log_);
 }
 
 int SocketStream::DoResolveHostComplete(int result) {
   if (result == OK) {
     next_state_ = STATE_TCP_CONNECT;
-    result = throttle_->OnStartOpenConnection(this, &io_callback_);
+    result = delegate_->OnStartOpenConnection(this, &io_callback_);
     if (result == net::ERR_IO_PENDING)
       metrics_->OnWaitConnection();
   } else {
     next_state_ = STATE_CLOSE;
   }
   // TODO(ukai): if error occured, reconsider proxy after error.
   return result;
 }
 
 int SocketStream::DoTcpConnect() {
@@ skipping 438 matching lines @@
 
 SSLConfigService* SocketStream::ssl_config_service() const {
   return context_->ssl_config_service();
 }
 
 ProxyService* SocketStream::proxy_service() const {
   return context_->proxy_service();
 }
 
 }  // namespace net