Factor out the QUIC socket reading code into a stand alone QuicPacketReader

net/quic/quic_client_session.cc

 // Copyright (c) 2012 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 "net/quic/quic_client_session.h"
 
 #include "base/callback_helpers.h"
 #include "base/message_loop/message_loop.h"
 #include "base/metrics/histogram.h"
 #include "base/metrics/sparse_histogram.h"
@@ skipping 148 matching lines @@
     scoped_ptr<QuicServerInfo> server_info,
     const QuicConfig& config,
     const char* const connection_description,
     base::TimeTicks dns_resolution_end_time,
     base::TaskRunner* task_runner,
     NetLog* net_log)
     : QuicClientSessionBase(connection, config),
       require_confirmation_(false),
       stream_factory_(stream_factory),
       socket_(socket.Pass()),
-      read_buffer_(new IOBufferWithSize(kMaxPacketSize)),
       transport_security_state_(transport_security_state),
       server_info_(server_info.Pass()),
-      read_pending_(false),
       num_total_streams_(0),
       task_runner_(task_runner),
       net_log_(BoundNetLog::Make(net_log, NetLog::SOURCE_QUIC_SESSION)),
+      packet_reader_(socket_.get(), this, net_log_),
       dns_resolution_end_time_(dns_resolution_end_time),
       logger_(new QuicConnectionLogger(this, connection_description, net_log_)),
-      num_packets_read_(0),
       going_away_(false),
       weak_factory_(this) {
   connection->set_debug_visitor(logger_.get());
   IPEndPoint address;
   if (socket && socket->GetLocalAddress(&address) == OK &&
       address.GetFamily() == ADDRESS_FAMILY_IPV6) {
     connection->set_max_packet_length(
         connection->max_packet_length() - kAdditionalOverheadForIPv6);
   }
 }
@@ skipping 571 matching lines @@
   const ProofVerifyDetailsChromium* verify_details_chromium =
       reinterpret_cast<const ProofVerifyDetailsChromium*>(&verify_details);
   CertVerifyResult* result_copy = new CertVerifyResult;
   result_copy->CopyFrom(verify_details_chromium->cert_verify_result);
   cert_verify_result_.reset(result_copy);
   pinning_failure_log_ = verify_details_chromium->pinning_failure_log;
   logger_->OnCertificateVerified(*cert_verify_result_);
 }
 
 void QuicClientSession::StartReading() {
-  if (read_pending_) {
-    return;
-  }
-  read_pending_ = true;
-  int rv = socket_->Read(read_buffer_.get(),
-                         read_buffer_->size(),
-                         base::Bind(&QuicClientSession::OnReadComplete,
-                                    weak_factory_.GetWeakPtr()));
-  UMA_HISTOGRAM_BOOLEAN("Net.QuicSession.AsyncRead", rv == ERR_IO_PENDING);
-  if (rv == ERR_IO_PENDING) {
-    num_packets_read_ = 0;
-    return;
-  }
-
-  if (++num_packets_read_ > 32) {
-    num_packets_read_ = 0;
-    // Data was read, process it.
-    // Schedule the work through the message loop to 1) prevent infinite
-    // recursion and 2) avoid blocking the thread for too long.
-    base::MessageLoop::current()->PostTask(
-        FROM_HERE,
-        base::Bind(&QuicClientSession::OnReadComplete,
-                   weak_factory_.GetWeakPtr(), rv));
-  } else {
-    OnReadComplete(rv);
-  }
+  packet_reader_.StartReading();
 }
 
 void QuicClientSession::CloseSessionOnError(int error) {
   UMA_HISTOGRAM_SPARSE_SLOWLY("Net.QuicSession.CloseSessionOnError", -error);
   CloseSessionOnErrorInner(error, QUIC_INTERNAL_ERROR);
   NotifyFactoryOfSessionClosed();
 }
 
 void QuicClientSession::CloseSessionOnErrorInner(int net_error,
                                                  QuicErrorCode quic_error) {
@@ skipping 61 matching lines @@
   }
   dict->Set("aliases", alias_list);
 
   return dict;
 }
 
 base::WeakPtr<QuicClientSession> QuicClientSession::GetWeakPtr() {
   return weak_factory_.GetWeakPtr();
 }
 
-void QuicClientSession::OnReadComplete(int result) {
-  read_pending_ = false;
-  if (result == 0)
-    result = ERR_CONNECTION_CLOSED;
+void QuicClientSession::OnReadError(int result) {
+  DVLOG(1) << "Closing session on read error: " << result;
+  UMA_HISTOGRAM_SPARSE_SLOWLY("Net.QuicSession.ReadError", -result);
+  NotifyFactoryOfSessionGoingAway();
+  CloseSessionOnErrorInner(result, QUIC_PACKET_READ_ERROR);
+  NotifyFactoryOfSessionClosedLater();
+}
 
-  if (result < 0) {
-    DVLOG(1) << "Closing session on read error: " << result;
-    UMA_HISTOGRAM_SPARSE_SLOWLY("Net.QuicSession.ReadError", -result);
-    NotifyFactoryOfSessionGoingAway();
-    CloseSessionOnErrorInner(result, QUIC_PACKET_READ_ERROR);
-    NotifyFactoryOfSessionClosedLater();
-    return;
-  }
-
-  QuicEncryptedPacket packet(read_buffer_->data(), result);
-  IPEndPoint local_address;
-  IPEndPoint peer_address;
-  socket_->GetLocalAddress(&local_address);
-  socket_->GetPeerAddress(&peer_address);
-  // ProcessUdpPacket might result in |this| being deleted, so we
-  // use a weak pointer to be safe.
+bool QuicClientSession::OnPacket(const QuicEncryptedPacket& packet,
+                                 IPEndPoint local_address,
+                                 IPEndPoint peer_address) {
   connection()->ProcessUdpPacket(local_address, peer_address, packet);
   if (!connection()->connected()) {
     NotifyFactoryOfSessionClosedLater();
-    return;
+    return false;
   }
-  StartReading();
+  return true;
 }
 
 void QuicClientSession::NotifyFactoryOfSessionGoingAway() {
   going_away_ = true;
   if (stream_factory_)
     stream_factory_->OnSessionGoingAway(this);
 }
 
 void QuicClientSession::NotifyFactoryOfSessionClosedLater() {
   if (!streams()->empty())
@@ skipping 33 matching lines @@
     return;
 
   // TODO(rch): re-enable this code once beta is cut.
   //  if (stream_factory_)
   //    stream_factory_->OnSessionConnectTimeout(this);
   //  CloseAllStreams(ERR_QUIC_HANDSHAKE_FAILED);
   //  DCHECK_EQ(0u, GetNumOpenStreams());
 }
 
 }  // namespace net