Add an async ReadInitialHeaders method to QuicChromiumClientStream::Handle

net/quic/chromium/quic_chromium_client_stream.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/chromium/quic_chromium_client_stream.h"
 
 #include <utility>
 
 #include "base/bind_helpers.h"
 #include "base/callback_helpers.h"
 #include "base/location.h"
 #include "base/threading/thread_task_runner_handle.h"
 #include "net/base/io_buffer.h"
 #include "net/base/net_errors.h"
 #include "net/log/net_log_event_type.h"
 #include "net/quic/chromium/quic_chromium_client_session.h"
 #include "net/quic/chromium/quic_http_utils.h"
 #include "net/quic/core/quic_spdy_session.h"
 #include "net/quic/core/quic_write_blocked_list.h"
 #include "net/quic/core/spdy_utils.h"
 
 namespace net {
 
 QuicChromiumClientStream::Handle::Handle(QuicChromiumClientStream* stream,
                                          Delegate* delegate)
-    : stream_(stream), delegate_(delegate) {
+    : stream_(stream), delegate_(delegate), read_headers_buffer_(nullptr) {
   SaveState();
 }
 
 QuicChromiumClientStream::Handle::~Handle() {
   if (stream_) {
     stream_->ClearHandle();
     // TODO(rch): If stream_ is still valid, it should probably be Reset()
     // so that it does not leak.
     // stream_->Reset(QUIC_STREAM_CANCELLED);
   }
 }
 
 void QuicChromiumClientStream::Handle::ClearDelegate() {
   delegate_ = nullptr;
 }
 
-void QuicChromiumClientStream::Handle::OnInitialHeadersAvailable(
-    const SpdyHeaderBlock& headers,
-    size_t frame_len) {
-  delegate_->OnInitialHeadersAvailable(headers, frame_len);
+void QuicChromiumClientStream::Handle::OnInitialHeadersAvailable() {
+  if (!read_headers_callback_)
+    return;  // Wait for ReadInitialHeaders to be called.
+
+  int rv = ERR_QUIC_PROTOCOL_ERROR;
+  if (!stream_->DeliverInitialHeaders(read_headers_buffer_, &rv))
+    rv = ERR_QUIC_PROTOCOL_ERROR;
+
+  ResetAndReturn(&read_headers_callback_).Run(rv);
 }
 
 void QuicChromiumClientStream::Handle::OnTrailingHeadersAvailable(
     const SpdyHeaderBlock& headers,
     size_t frame_len) {
   delegate_->OnTrailingHeadersAvailable(headers, frame_len);
 }
 
 void QuicChromiumClientStream::Handle::OnDataAvailable() {
   delegate_->OnDataAvailable();
@@ skipping 14 matching lines @@
   if (stream_)
     SaveState();
   stream_ = nullptr;
   if (delegate_) {
     auto* delegate = delegate_;
     delegate_ = nullptr;
     delegate->OnError(error);
   }
 }
 
+int QuicChromiumClientStream::Handle::ReadInitialHeaders(
+    SpdyHeaderBlock* header_block,
+    const CompletionCallback& callback) {
+  if (!stream_)
+    return ERR_CONNECTION_CLOSED;
+
+  int frame_len = 0;
+  if (stream_->DeliverInitialHeaders(header_block, &frame_len))
+    return frame_len;
+
+  read_headers_buffer_ = header_block;
+  read_headers_callback_ = callback;
+  return ERR_IO_PENDING;
+}
+
 size_t QuicChromiumClientStream::Handle::WriteHeaders(
     SpdyHeaderBlock header_block,
     bool fin,
     QuicReferenceCountedPointer<QuicAckListenerInterface>
         ack_notifier_delegate) {
   if (!stream_)
     return 0;
   return stream_->WriteHeaders(std::move(header_block), fin,
                                ack_notifier_delegate);
 }
@@ skipping 173 matching lines @@
   if (!SpdyUtils::CopyAndValidateHeaders(header_list, &length, &header_block)) {
     DLOG(ERROR) << "Failed to parse header list: " << header_list.DebugString();
     ConsumeHeaderList();
     Reset(QUIC_BAD_APPLICATION_PAYLOAD);
     return;
   }
 
   ConsumeHeaderList();
   session_->OnInitialHeadersComplete(id(), header_block);
 
-  if (handle_) {
-    // The handle will receive the headers via a posted task.
-    NotifyHandleOfInitialHeadersAvailableLater(std::move(header_block),
-                                               frame_len);
-    return;
-  }
-
   // Buffer the headers and deliver them when the handle arrives.
   initial_headers_ = std::move(header_block);
   initial_headers_frame_len_ = frame_len;
+
+  if (handle_) {
+    // The handle will be notified of the headers via a posted task.
+    NotifyHandleOfInitialHeadersAvailableLater();
+  }
 }
 
 void QuicChromiumClientStream::OnTrailingHeadersComplete(
     bool fin,
     size_t frame_len,
     const QuicHeaderList& header_list) {
   QuicSpdyStream::OnTrailingHeadersComplete(fin, frame_len, header_list);
   NotifyHandleOfTrailingHeadersAvailableLater(received_trailers().Clone(),
                                               frame_len);
 }
@@ skipping 113 matching lines @@
 
 std::unique_ptr<QuicChromiumClientStream::Handle>
 QuicChromiumClientStream::CreateHandle(
     QuicChromiumClientStream::Delegate* delegate) {
   DCHECK(!handle_);
   auto handle = std::unique_ptr<QuicChromiumClientStream::Handle>(
       new QuicChromiumClientStream::Handle(this, delegate));
   handle_ = handle.get();
 
   // Should this perhaps be via PostTask to make reasoning simpler?
-  if (!initial_headers_.empty()) {
-    handle_->OnInitialHeadersAvailable(std::move(initial_headers_),
-                                       initial_headers_frame_len_);
-  }
+  if (!initial_headers_.empty())
+    handle_->OnInitialHeadersAvailable();
 
   return handle;
 }
 
 void QuicChromiumClientStream::ClearHandle() {
   handle_ = nullptr;
 }
 
 void QuicChromiumClientStream::OnError(int error) {
   if (handle_) {
@@ skipping 12 matching lines @@
 
   iovec iov;
   iov.iov_base = buf->data();
   iov.iov_len = buf_len;
   size_t bytes_read = Readv(&iov, 1);
   // Since HasBytesToRead is true, Readv() must of read some data.
   DCHECK_NE(0u, bytes_read);
   return bytes_read;
 }
 
-void QuicChromiumClientStream::NotifyHandleOfInitialHeadersAvailableLater(
-    SpdyHeaderBlock headers,
-    size_t frame_len) {
+void QuicChromiumClientStream::NotifyHandleOfInitialHeadersAvailableLater() {
   DCHECK(handle_);
   base::ThreadTaskRunnerHandle::Get()->PostTask(
       FROM_HERE,
       base::Bind(
           &QuicChromiumClientStream::NotifyHandleOfInitialHeadersAvailable,
-          weak_factory_.GetWeakPtr(), base::Passed(std::move(headers)),
-          frame_len));
+          weak_factory_.GetWeakPtr()));
 }
 
-void QuicChromiumClientStream::NotifyHandleOfInitialHeadersAvailable(
-    SpdyHeaderBlock headers,
-    size_t frame_len) {
+void QuicChromiumClientStream::NotifyHandleOfInitialHeadersAvailable() {
   if (!handle_)
     return;
 
-  DCHECK(!headers_delivered_);
-  headers_delivered_ = true;
-  net_log_.AddEvent(
-      NetLogEventType::QUIC_CHROMIUM_CLIENT_STREAM_READ_RESPONSE_HEADERS,
-      base::Bind(&SpdyHeaderBlockNetLogCallback, &headers));
-
-  handle_->OnInitialHeadersAvailable(headers, frame_len);
+  if (!headers_delivered_)
+    handle_->OnInitialHeadersAvailable();
 }
 
 void QuicChromiumClientStream::NotifyHandleOfTrailingHeadersAvailableLater(
     SpdyHeaderBlock headers,
     size_t frame_len) {
   DCHECK(handle_);
   base::ThreadTaskRunnerHandle::Get()->PostTask(
       FROM_HERE,
       base::Bind(
           &QuicChromiumClientStream::NotifyHandleOfTrailingHeadersAvailable,
           weak_factory_.GetWeakPtr(), base::Passed(std::move(headers)),
           frame_len));
 }
 
 void QuicChromiumClientStream::NotifyHandleOfTrailingHeadersAvailable(
     SpdyHeaderBlock headers,
     size_t frame_len) {
   if (!handle_)
     return;
 
   DCHECK(headers_delivered_);
   // Only mark trailers consumed when we are about to notify delegate.
   MarkTrailersConsumed();
   // Post an async task to notify delegate of the FIN flag.
   NotifyHandleOfDataAvailableLater();
   net_log_.AddEvent(
       NetLogEventType::QUIC_CHROMIUM_CLIENT_STREAM_READ_RESPONSE_TRAILERS,
       base::Bind(&SpdyHeaderBlockNetLogCallback, &headers));
+  handle_->OnTrailingHeadersAvailable(headers, frame_len);
+}
 
-  handle_->OnTrailingHeadersAvailable(headers, frame_len);
+bool QuicChromiumClientStream::DeliverInitialHeaders(SpdyHeaderBlock* headers,
+                                                     int* frame_len) {
+  if (initial_headers_.empty())
+    return false;
+
+  headers_delivered_ = true;
+  net_log_.AddEvent(
+      NetLogEventType::QUIC_CHROMIUM_CLIENT_STREAM_READ_RESPONSE_HEADERS,
+      base::Bind(&SpdyHeaderBlockNetLogCallback, &initial_headers_));
+
+  *headers = std::move(initial_headers_);
+  *frame_len = initial_headers_frame_len_;
+  return true;
 }
 
 void QuicChromiumClientStream::NotifyHandleOfDataAvailableLater() {
   DCHECK(handle_);
   base::ThreadTaskRunnerHandle::Get()->PostTask(
       FROM_HERE,
       base::Bind(&QuicChromiumClientStream::NotifyHandleOfDataAvailable,
                  weak_factory_.GetWeakPtr()));
 }
 
 void QuicChromiumClientStream::NotifyHandleOfDataAvailable() {
   if (handle_)
     handle_->OnDataAvailable();
 }
 
 void QuicChromiumClientStream::DisableConnectionMigration() {
   can_migrate_ = false;
 }
 
 bool QuicChromiumClientStream::IsFirstStream() {
   return id() == kHeadersStreamId + 2;
 }
 
 }  // namespace net