Add a QuicHttpStream class.

Review URL: https://chromiumcodereview.appspot.com/11364068

git-svn-id: svn://svn.chromium.org/chrome/trunk/src@170968 0039d316-1c4b-4281-b951-d872f2087c98

net/http/http_stream_base.h

@@ -35,6 +35,7 @@ class NET_EXPORT_PRIVATE HttpStreamBase {
   virtual ~HttpStreamBase() {}
 
   // Initialize stream.  Must be called before calling SendRequest().
+  // |request_info| must outlive the HttpStreamBase.
   // Returns a net error code, possibly ERR_IO_PENDING.
   virtual int InitializeStream(const HttpRequestInfo* request_info,
                                const BoundNetLog& net_log,
@@ -44,6 +45,7 @@ class NET_EXPORT_PRIVATE HttpStreamBase {
   // ERR_IO_PENDING is returned if the operation could not be completed
   // synchronously, in which case the result will be passed to the callback
   // when available. Returns OK on success.
+  // |response| must outlive the HttpStreamBase.
   virtual int SendRequest(const HttpRequestHeaders& request_headers,
                           HttpResponseInfo* response,
                           const CompletionCallback& callback) = 0;
@@ -89,6 +91,9 @@ class NET_EXPORT_PRIVATE HttpStreamBase {
   // the response headers indicate either chunked encoding or content length.
   // If neither is sent, the server must close the connection for us to detect
   // the end of the response.
+  // TODO(rch): Rename this method, so that it is clear why it exists
+  // particularly as it applies to QUIC and SPDY for which the end of the
+  // response is always findable.
   virtual bool CanFindEndOfResponse() const = 0;
 
   // A stream exists on top of a connection.  If the connection has been used

net/net.gyp

@@ -684,6 +684,8 @@
         'quic/quic_fec_group.h',
         'quic/quic_framer.cc',
         'quic/quic_framer.h',
+        'quic/quic_http_stream.cc',
+        'quic/quic_http_stream.h',
         'quic/quic_packet_creator.cc',
         'quic/quic_packet_creator.h',
         'quic/quic_protocol.cc',
@@ -1447,6 +1449,7 @@
         'quic/quic_crypto_stream_test.cc',
         'quic/quic_fec_group_test.cc',
         'quic/quic_framer_test.cc',
+        'quic/quic_http_stream_test.cc',
         'quic/quic_packet_creator_test.cc',
         'quic/quic_reliable_client_stream_test.cc',
         'quic/quic_session_test.cc',

net/quic/quic_client_session.cc

@@ -14,6 +14,7 @@ QuicClientSession::QuicClientSession(QuicConnection* connection)
 }
 
 QuicClientSession::~QuicClientSession() {
+  STLDeleteValues(&streams_);
 }
 
 QuicReliableClientStream* QuicClientSession::CreateOutgoingReliableStream() {
@@ -28,6 +29,7 @@ QuicReliableClientStream* QuicClientSession::CreateOutgoingReliableStream() {
   }
   QuicReliableClientStream* stream =
        new QuicReliableClientStream(GetNextStreamId(), this);
+  streams_[stream->id()] = stream;
 
   ActivateStream(stream);
   return stream;
@@ -56,6 +58,18 @@ ReliableQuicStream* QuicClientSession::CreateIncomingReliableStream(
   return NULL;
 }
 
+void QuicClientSession::CloseStream(QuicStreamId stream_id) {
+  QuicSession::CloseStream(stream_id);
+
+  StreamMap::iterator it = streams_.find(stream_id);
+  DCHECK(it != streams_.end());
+  if (it != streams_.end()) {
+    ReliableQuicStream* stream = it->second;
+    streams_.erase(it);
+    delete stream;
+  }
+}
+
 void QuicClientSession::OnCryptoHandshakeComplete(QuicErrorCode error) {
   if (!callback_.is_null()) {
     callback_.Run(error == QUIC_NO_ERROR ? OK : ERR_UNEXPECTED);

net/quic/quic_client_session.h

@@ -7,6 +7,7 @@
 #ifndef NET_QUIC_QUIC_CLIENT_SESSION_H_
 #define NET_QUIC_QUIC_CLIENT_SESSION_H_
 
+#include "base/hash_tables.h"
 #include "net/base/completion_callback.h"
 #include "net/quic/quic_crypto_client_stream.h"
 #include "net/quic/quic_reliable_client_stream.h"
@@ -23,6 +24,7 @@ class NET_EXPORT_PRIVATE QuicClientSession : public QuicSession {
   // QuicSession methods:
   virtual QuicReliableClientStream* CreateOutgoingReliableStream() OVERRIDE;
   virtual QuicCryptoClientStream* GetCryptoStream() OVERRIDE;
+  virtual void CloseStream(QuicStreamId stream_id) OVERRIDE;
   virtual void OnCryptoHandshakeComplete(QuicErrorCode error) OVERRIDE;
 
   // Perform a crypto handshake with the server.
@@ -34,7 +36,9 @@ class NET_EXPORT_PRIVATE QuicClientSession : public QuicSession {
       QuicStreamId id) OVERRIDE;
 
  private:
+  typedef base::hash_map<QuicStreamId, ReliableQuicStream*> StreamMap;
   QuicCryptoClientStream crypto_stream_;
+  StreamMap streams_;
 
   CompletionCallback callback_;
 

net/quic/quic_client_session_test.cc

@@ -60,17 +60,14 @@ TEST_F(QuicClientSessionTest, MaxNumConnections) {
   std::vector<QuicReliableClientStream*> streams;
   for (size_t i = 0; i < kDefaultMaxStreamsPerConnection; i++) {
     QuicReliableClientStream* stream = session_.CreateOutgoingReliableStream();
-    streams.push_back(stream);
     EXPECT_TRUE(stream);
+    streams.push_back(stream);
   }
   EXPECT_FALSE(session_.CreateOutgoingReliableStream());
 
   // Close a stream and ensure I can now open a new one.
   session_.CloseStream(streams[0]->id());
-  scoped_ptr<QuicReliableClientStream> stream(
-      session_.CreateOutgoingReliableStream());
-  EXPECT_TRUE(stream.get());
-  STLDeleteElements(&streams);
+  EXPECT_TRUE(session_.CreateOutgoingReliableStream());
 }
 
 }  // namespace

net/quic/quic_connection_helper.cc

@@ -66,7 +66,6 @@ void QuicConnectionHelper::SetResendAlarm(
 }
 
 void QuicConnectionHelper::SetSendAlarm(QuicTime::Delta delay) {
-  DCHECK(!send_alarm_registered_);
   send_alarm_registered_ = true;
   task_runner_->PostDelayedTask(
       FROM_HERE,

net/quic/quic_http_stream.h

+// 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.
+
+#ifndef NET_QUIC_QUIC_HTTP_STREAM_H_
+#define NET_QUIC_QUIC_HTTP_STREAM_H_
+
+#include <list>
+
+#include "base/memory/weak_ptr.h"
+#include "net/base/io_buffer.h"
+#include "net/http/http_stream.h"
+#include "net/quic/quic_reliable_client_stream.h"
+
+namespace net {
+
+// The QuicHttpStream is a QUIC-specific HttpStream subclass.  It holds a
+// non-owning pointer to a QuicReliableClientStream which it uses to
+// send and receive data.
+class NET_EXPORT_PRIVATE QuicHttpStream :
+      public QuicReliableClientStream::Delegate,
+      public HttpStream {
+ public:
+  explicit QuicHttpStream(QuicReliableClientStream* stream);
+
+  virtual ~QuicHttpStream();
+
+  // HttpStream implementation.
+  virtual int InitializeStream(const HttpRequestInfo* request_info,
+                               const BoundNetLog& net_log,
+                               const CompletionCallback& callback) OVERRIDE;
+  virtual int SendRequest(const HttpRequestHeaders& request_headers,
+                          HttpResponseInfo* response,
+                          const CompletionCallback& callback) OVERRIDE;
+  virtual UploadProgress GetUploadProgress() const OVERRIDE;
+  virtual int ReadResponseHeaders(const CompletionCallback& callback) OVERRIDE;
+  virtual const HttpResponseInfo* GetResponseInfo() const OVERRIDE;
+  virtual int ReadResponseBody(IOBuffer* buf,
+                               int buf_len,
+                               const CompletionCallback& callback) OVERRIDE;
+  virtual void Close(bool not_reusable) OVERRIDE;
+  virtual HttpStream* RenewStreamForAuth() OVERRIDE;
+  virtual bool IsResponseBodyComplete() const OVERRIDE;
+  virtual bool CanFindEndOfResponse() const OVERRIDE;
+  virtual bool IsMoreDataBuffered() const OVERRIDE;
+  virtual bool IsConnectionReused() const OVERRIDE;
+  virtual void SetConnectionReused() OVERRIDE;
+  virtual bool IsConnectionReusable() const OVERRIDE;
+  virtual void GetSSLInfo(SSLInfo* ssl_info) OVERRIDE;
+  virtual void GetSSLCertRequestInfo(
+      SSLCertRequestInfo* cert_request_info) OVERRIDE;
+  virtual bool IsSpdyHttpStream() const OVERRIDE;
+  virtual void LogNumRttVsBytesMetrics() const OVERRIDE {}
+  virtual void Drain(HttpNetworkSession* session) OVERRIDE;
+
+  // QuicReliableClientStream::Delegate implementation
+  virtual int OnSendData() OVERRIDE;
+  virtual int OnSendDataComplete(int status, bool* eof) OVERRIDE;
+  virtual int OnDataReceived(const char* data, int length) OVERRIDE;
+  virtual void OnClose(QuicErrorCode error) OVERRIDE;
+
+ private:
+  enum State {
+    STATE_NONE,
+    STATE_SEND_HEADERS,
+    STATE_SEND_HEADERS_COMPLETE,
+    STATE_READ_REQUEST_BODY,
+    STATE_READ_REQUEST_BODY_COMPLETE,
+    STATE_SEND_BODY,
+    STATE_SEND_BODY_COMPLETE,
+    STATE_OPEN,
+  };
+
+  void OnIOComplete(int rv);
+  void DoCallback(int rv);
+
+  int DoLoop(int);
+  int DoSendHeaders();
+  int DoSendHeadersComplete(int rv);
+  int DoReadRequestBody();
+  int DoReadRequestBodyComplete(int rv);
+  int DoSendBody();
+  int DoSendBodyComplete(int rv);
+  int DoReadResponseHeaders();
+  int DoReadResponseHeadersComplete(int rv);
+
+  int ParseResponseHeaders();
+
+  void BufferResponseBody(const char* data, int length);
+
+  State io_state_;
+
+  QuicReliableClientStream* stream_;  // Non-owning.
+
+  // The following three fields are all owned by the caller and must
+  // outlive this object, according to the HttpStream contract.
+
+  // The request to send.
+  const HttpRequestInfo* request_info_;
+  // The request body to send, if any, owned by the caller.
+  UploadDataStream* request_body_stream_;
+  // |response_info_| is the HTTP response data object which is filled in
+  // when a the response headers are read.  It is not owned by this stream.
+  HttpResponseInfo* response_info_;
+
+  bool response_headers_received_;
+
+  // Serialized HTTP request.
+  std::string request_;
+
+  // Buffer into which response header data is read.
+  scoped_refptr<GrowableIOBuffer> read_buf_;
+
+  // We buffer the response body as it arrives asynchronously from the stream.
+  // TODO(rch): This is infinite buffering, which is bad.
+  std::list<scoped_refptr<IOBufferWithSize> > response_body_;
+
+  // The caller's callback to be used for asynchronous operations.
+  CompletionCallback callback_;
+
+  // Caller provided buffer for the ReadResponseBody() response.
+  scoped_refptr<IOBuffer> user_buffer_;
+  int user_buffer_len_;
+
+  // Temporary buffer used to read the request body from UploadDataStream.
+  scoped_refptr<IOBufferWithSize> raw_request_body_buf_;
+  // Wraps raw_request_body_buf_ to read the remaining data progressively.
+  scoped_refptr<DrainableIOBuffer> request_body_buf_;
+
+  base::WeakPtrFactory<QuicHttpStream> weak_factory_;
+};
+
+}  // namespace net
+
+#endif  // NET_QUIC_QUIC_HTTP_STREAM_H_

net/quic/quic_protocol.h

@@ -261,7 +261,7 @@ struct NET_EXPORT_PRIVATE QuicAckFrame {
 struct NET_EXPORT_PRIVATE QuicRstStreamFrame {
   QuicRstStreamFrame() {}
   QuicRstStreamFrame(QuicStreamId stream_id, uint64 offset,
-                        QuicErrorCode error_code)
+                     QuicErrorCode error_code)
       : stream_id(stream_id), offset(offset), error_code(error_code) {
     DCHECK_LE(error_code, std::numeric_limits<uint8>::max());
   }

net/quic/quic_reliable_client_stream.cc

@@ -5,19 +5,26 @@
 #include "net/quic/quic_reliable_client_stream.h"
 
 #include "net/base/net_errors.h"
+#include "net/quic/quic_session.h"
 
 namespace net {
 
 QuicReliableClientStream::QuicReliableClientStream(QuicStreamId id,
                                                    QuicSession* session)
-    : ReliableQuicStream(id, session) {
+    : ReliableQuicStream(id, session),
+      delegate_(NULL) {
 }
 
 QuicReliableClientStream::~QuicReliableClientStream() {
+  if (delegate_) {
+    delegate_->OnClose(error());
+  }
 }
 
 uint32 QuicReliableClientStream::ProcessData(const char* data,
                                              uint32 data_len) {
+  // TODO(rch): buffer data if we don't have a delegate.
+  DCHECK(delegate_);
   int rv = delegate_->OnDataReceived(data, data_len);
   if (rv != OK) {
     DLOG(ERROR) << "Delegate refused data, rv: " << rv;
@@ -28,11 +35,15 @@ uint32 QuicReliableClientStream::ProcessData(const char* data,
 }
 
 void QuicReliableClientStream::TerminateFromPeer(bool half_close) {
-  delegate_->OnClose(error());
+  if (delegate_) {
+    delegate_->OnClose(error());
+    delegate_ = NULL;
+  }
 }
 
 void QuicReliableClientStream::SetDelegate(
     QuicReliableClientStream::Delegate* delegate) {
+  DCHECK((!delegate_ && delegate) || (delegate_ && !delegate));
   delegate_ = delegate;
 }
 

net/quic/quic_reliable_client_stream.h

@@ -7,6 +7,7 @@
 #ifndef NET_QUIC_QUIC_RELIABLE_CLIENT_STREAM_H_
 #define NET_QUIC_QUIC_RELIABLE_CLIENT_STREAM_H_
 
+#include "net/base/ip_endpoint.h"
 #include "net/base/upload_data_stream.h"
 #include "net/http/http_request_info.h"
 #include "net/http/http_response_info.h"
@@ -17,10 +18,12 @@ namespace net {
 
 class QuicClientSession;
 
+// A client-initiated ReliableQuicStream.  Instances of this class
+// are owned by the QuicClientSession which created them.
 class NET_EXPORT_PRIVATE QuicReliableClientStream : public ReliableQuicStream {
  public:
   // Delegate handles protocol specific behavior of a quic stream.
-  class Delegate {
+  class NET_EXPORT_PRIVATE Delegate {
    public:
     Delegate() {}
 
@@ -57,6 +60,7 @@ class NET_EXPORT_PRIVATE QuicReliableClientStream : public ReliableQuicStream {
   // ReliableQuicStream
   virtual uint32 ProcessData(const char* data, uint32 data_len) OVERRIDE;
   virtual void TerminateFromPeer(bool half_close) OVERRIDE;
+  using ReliableQuicStream::WriteData;
 
   // Set new |delegate|. |delegate| must not be NULL.
   // If this stream has already received data, OnDataReceived() will be

net/quic/quic_reliable_client_stream_test.cc

@@ -52,6 +52,7 @@ TEST_F(QuicReliableClientStreamTest, TerminateFromPeer) {
 TEST_F(QuicReliableClientStreamTest, ProcessData) {
   const char data[] = "hello world!";
   EXPECT_CALL(delegate_, OnDataReceived(StrEq(data), arraysize(data)));
+  EXPECT_CALL(delegate_, OnClose(QUIC_NO_ERROR));
 
   EXPECT_EQ(arraysize(data), stream_.ProcessData(data, arraysize(data)));
 }
@@ -61,6 +62,8 @@ TEST_F(QuicReliableClientStreamTest, ProcessDataWithError) {
   EXPECT_CALL(delegate_,
               OnDataReceived(StrEq(data),
                              arraysize(data))).WillOnce(Return(ERR_UNEXPECTED));
+  EXPECT_CALL(delegate_, OnClose(QUIC_BAD_APPLICATION_PAYLOAD));
+
 
   EXPECT_EQ(0u, stream_.ProcessData(data, arraysize(data)));
 }

net/quic/reliable_quic_stream.cc

@@ -90,18 +90,22 @@ bool ReliableQuicStream::HasBytesToRead() const {
   return sequencer_.HasBytesToRead();
 }
 
+const IPEndPoint& ReliableQuicStream::GetPeerAddress() const {
+  return session_->peer_address();
+}
+
 int ReliableQuicStream::WriteData(StringPiece data, bool fin) {
   if (write_side_closed_) {
     DLOG(ERROR) << "Attempt to write when the write side is closed";
     return 0;
   }
 
-  session()->WriteData(id(), data, offset_, fin);
+  int rv = session()->WriteData(id(), data, offset_, fin);
   offset_ += data.length();
   if (fin) {
     CloseWriteSide();
   }
-  return data.length();
+  return rv;
 }
 
 void ReliableQuicStream::CloseReadSide() {
@@ -112,6 +116,7 @@ void ReliableQuicStream::CloseReadSide() {
 
   read_side_closed_ = true;
   if (write_side_closed_) {
+    DLOG(INFO) << "Closing stream: " << id();
     session_->CloseStream(id());
   }
 }
@@ -124,6 +129,7 @@ void ReliableQuicStream::CloseWriteSide() {
 
   write_side_closed_ = true;
   if (read_side_closed_) {
+    DLOG(INFO) << "Closing stream: " << id();
     session_->CloseStream(id());
   }
 }

net/quic/reliable_quic_stream.h

@@ -11,6 +11,7 @@
 
 namespace net {
 
+class IPEndPoint;
 class QuicSession;
 
 // All this does right now is send data to subclasses via the sequencer.
@@ -55,6 +56,8 @@ class NET_EXPORT_PRIVATE ReliableQuicStream {
   bool read_side_closed() const { return read_side_closed_; }
   bool write_side_closed() const { return write_side_closed_; }
 
+  const IPEndPoint& GetPeerAddress() const;
+
  protected:
   virtual int WriteData(base::StringPiece data, bool fin);
   // Close the read side of the socket.  Further frames will not be accepted.