Remove QuicChromiumClientStream::Delegate in favor of async methods.

net/quic/chromium/bidirectional_stream_quic_impl.cc

 // Copyright 2016 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/bidirectional_stream_quic_impl.h"
 
 #include <utility>
 
 #include "base/bind.h"
 #include "base/location.h"
@@ skipping 26 matching lines @@
 }  // namespace
 
 BidirectionalStreamQuicImpl::BidirectionalStreamQuicImpl(
     std::unique_ptr<QuicChromiumClientSession::Handle> session)
     : session_(std::move(session)),
       stream_(nullptr),
       request_info_(nullptr),
       delegate_(nullptr),
       response_status_(OK),
       negotiated_protocol_(kProtoUnknown),
+      expect_trailers_(true),
       read_buffer_len_(0),
       headers_bytes_received_(0),
       headers_bytes_sent_(0),
       closed_stream_received_bytes_(0),
       closed_stream_sent_bytes_(0),
       closed_is_first_stream_(false),
       has_sent_headers_(false),
       send_request_headers_automatically_(true),
       may_invoke_callbacks_(true),
       weak_factory_(this) {}
@@ skipping 38 matching lines @@
     return;
   }
 
   base::ThreadTaskRunnerHandle::Get()->PostTask(
       FROM_HERE, base::Bind(&BidirectionalStreamQuicImpl::OnStreamReady,
                             weak_factory_.GetWeakPtr(), rv));
 }
 
 void BidirectionalStreamQuicImpl::SendRequestHeaders() {
   ScopedBoolSaver saver(&may_invoke_callbacks_, false);
-  // If this fails, a task will have been posted to notify the delegate
-  // asynchronously.
-  WriteHeaders();
+  int rv = WriteHeaders();
+  if (rv < 0) {
+    base::ThreadTaskRunnerHandle::Get()->PostTask(
+        FROM_HERE, base::Bind(&BidirectionalStreamQuicImpl::NotifyError,
+                              weak_factory_.GetWeakPtr(), rv));
+  }
 }
 
-bool BidirectionalStreamQuicImpl::WriteHeaders() {
+int BidirectionalStreamQuicImpl::WriteHeaders() {
   DCHECK(!has_sent_headers_);
-  if (!stream_) {
-    LOG(ERROR)
-        << "Trying to send request headers after stream has been destroyed.";
-    base::ThreadTaskRunnerHandle::Get()->PostTask(
-        FROM_HERE, base::Bind(&BidirectionalStreamQuicImpl::NotifyError,
-                              weak_factory_.GetWeakPtr(), ERR_UNEXPECTED));
-    return false;
-  }
 
   SpdyHeaderBlock headers;
   HttpRequestInfo http_request_info;
   http_request_info.url = request_info_->url;
   http_request_info.method = request_info_->method;
   http_request_info.extra_headers = request_info_->extra_headers;
 
   CreateSpdyHeadersFromHttpRequest(
       http_request_info, http_request_info.extra_headers, true, &headers);
-  // Sending the request might result in the stream being closed via OnClose
-  // which will post a task to notify the delegate asynchronously.
-  // TODO(rch): Clean up this interface when OnClose and OnError are removed.
-  size_t headers_bytes_sent = stream_->WriteHeaders(
-      std::move(headers), request_info_->end_stream_on_headers, nullptr);
-  if (!stream_)
-    return false;
-
-  headers_bytes_sent_ += headers_bytes_sent;
-  has_sent_headers_ = true;
-  return true;
+  int rv = stream_->WriteHeaders(std::move(headers),
+                                 request_info_->end_stream_on_headers, nullptr);
+  if (rv >= 0) {
+    headers_bytes_sent_ += rv;
+    has_sent_headers_ = true;
+  }
+  return rv;
 }
 
 int BidirectionalStreamQuicImpl::ReadData(IOBuffer* buffer, int buffer_len) {
   ScopedBoolSaver saver(&may_invoke_callbacks_, false);
   DCHECK(buffer);
   DCHECK(buffer_len);
 
-  if (!stream_) {
-    // If the stream is already closed, there is no body to read.
-    return response_status_;
-  }
   int rv = stream_->ReadBody(
       buffer, buffer_len,
       base::Bind(&BidirectionalStreamQuicImpl::OnReadDataComplete,
                  weak_factory_.GetWeakPtr()));
   if (rv == ERR_IO_PENDING) {
     read_buffer_ = buffer;
     read_buffer_len_ = buffer_len;
     return ERR_IO_PENDING;
   }
 
   if (rv < 0)
     return rv;
 
   if (stream_->IsDoneReading()) {
     // If the write side is closed, OnFinRead() will call
     // BidirectionalStreamQuicImpl::OnClose().
+    expect_trailers_ = false;
     stream_->OnFinRead();
   }
   return rv;
 }
 
 void BidirectionalStreamQuicImpl::SendData(const scoped_refptr<IOBuffer>& data,
                                            int length,
                                            bool end_stream) {
   ScopedBoolSaver saver(&may_invoke_callbacks_, false);
   DCHECK(length > 0 || (length == 0 && end_stream));
-  if (!stream_) {
-    LOG(ERROR) << "Trying to send data after stream has been destroyed.";
+  if (!stream_->IsOpen()) {
+    LOG(ERROR) << "Trying to send data after stream has been closed.";
     base::ThreadTaskRunnerHandle::Get()->PostTask(
         FROM_HERE, base::Bind(&BidirectionalStreamQuicImpl::NotifyError,
                               weak_factory_.GetWeakPtr(), ERR_UNEXPECTED));
     return;
   }
 
   std::unique_ptr<QuicConnection::ScopedPacketBundler> bundler;
   if (!has_sent_headers_) {
     DCHECK(!send_request_headers_automatically_);
     // Creates a bundler only if there are headers to be sent along with the
     // single data buffer.
     bundler =
         session_->CreatePacketBundler(QuicConnection::SEND_ACK_IF_PENDING);
     // Sending the request might result in the stream being closed.
-    if (!WriteHeaders())
+    int rv = WriteHeaders();
+    if (rv < 0) {
+      base::ThreadTaskRunnerHandle::Get()->PostTask(
+          FROM_HERE, base::Bind(&BidirectionalStreamQuicImpl::NotifyError,
+                                weak_factory_.GetWeakPtr(), rv));
       return;
+    }
   }
 
   QuicStringPiece string_data(data->data(), length);
   int rv = stream_->WriteStreamData(
       string_data, end_stream,
       base::Bind(&BidirectionalStreamQuicImpl::OnSendDataComplete,
                  weak_factory_.GetWeakPtr()));
-  DCHECK(rv == OK || rv == ERR_IO_PENDING);
-  if (rv == OK) {
+  if (rv != ERR_IO_PENDING) {
     base::ThreadTaskRunnerHandle::Get()->PostTask(
         FROM_HERE, base::Bind(&BidirectionalStreamQuicImpl::OnSendDataComplete,
-                              weak_factory_.GetWeakPtr(), OK));
+                              weak_factory_.GetWeakPtr(), rv));
   }
 }
 
 void BidirectionalStreamQuicImpl::SendvData(
     const std::vector<scoped_refptr<IOBuffer>>& buffers,
     const std::vector<int>& lengths,
     bool end_stream) {
   ScopedBoolSaver saver(&may_invoke_callbacks_, false);
   DCHECK_EQ(buffers.size(), lengths.size());
 
-  if (!stream_) {
-    LOG(ERROR) << "Trying to send data after stream has been destroyed.";
+  if (!stream_->IsOpen()) {
+    LOG(ERROR) << "Trying to send data after stream has been closed.";
     base::ThreadTaskRunnerHandle::Get()->PostTask(
         FROM_HERE, base::Bind(&BidirectionalStreamQuicImpl::NotifyError,
                               weak_factory_.GetWeakPtr(), ERR_UNEXPECTED));
     return;
   }
 
   std::unique_ptr<QuicConnection::ScopedPacketBundler> bundler(
       session_->CreatePacketBundler(QuicConnection::SEND_ACK_IF_PENDING));
   if (!has_sent_headers_) {
     DCHECK(!send_request_headers_automatically_);
-    // Sending the request might result in the stream being closed.
-    if (!WriteHeaders())
+    int rv = WriteHeaders();
+    if (rv < 0) {
+      base::ThreadTaskRunnerHandle::Get()->PostTask(
+          FROM_HERE, base::Bind(&BidirectionalStreamQuicImpl::NotifyError,
+                                weak_factory_.GetWeakPtr(), rv));

[Ryan Hamilton, 2017/06/01 23:20:30]

Since we do this same thing every time we call WriteHeaders (check rv, and
PostTask if it failed). Should I go back to returning a bool, and do the
PostTask inside of WriteHeaders?

       return;
+    }
   }
 
   int rv = stream_->WritevStreamData(
       buffers, lengths, end_stream,
       base::Bind(&BidirectionalStreamQuicImpl::OnSendDataComplete,
                  weak_factory_.GetWeakPtr()));
 
-  DCHECK(rv == OK || rv == ERR_IO_PENDING);
-  if (rv == OK) {
+  if (rv != ERR_IO_PENDING) {
     base::ThreadTaskRunnerHandle::Get()->PostTask(
         FROM_HERE, base::Bind(&BidirectionalStreamQuicImpl::OnSendDataComplete,
-                              weak_factory_.GetWeakPtr(), OK));
+                              weak_factory_.GetWeakPtr(), rv));
   }
 }
 
 NextProto BidirectionalStreamQuicImpl::GetProtocol() const {
   return negotiated_protocol_;
 }
 
 int64_t BidirectionalStreamQuicImpl::GetTotalReceivedBytes() const {
   if (stream_)
     return headers_bytes_received_ + stream_->stream_bytes_read();
@@ skipping 13 matching lines @@
     is_first_stream = stream_->IsFirstStream();
   if (is_first_stream) {
     load_timing_info->socket_reused = false;
     load_timing_info->connect_timing = connect_timing_;
   } else {
     load_timing_info->socket_reused = true;
   }
   return true;
 }
 
-void BidirectionalStreamQuicImpl::OnClose() {
-  DCHECK(stream_);
-
-  if (stream_->connection_error() != QUIC_NO_ERROR ||
-      stream_->stream_error() != QUIC_STREAM_NO_ERROR) {
-    OnError(session_->IsCryptoHandshakeConfirmed() ? ERR_QUIC_PROTOCOL_ERROR
-                                                   : ERR_QUIC_HANDSHAKE_FAILED);
-    return;
-  }
-
-  if (!stream_->fin_sent() || !stream_->fin_received()) {
-    // The connection must have been closed by the peer with QUIC_NO_ERROR,
-    // which is improper.
-    OnError(ERR_UNEXPECTED);
-    return;
-  }
-
-  // The connection was closed normally so there is no need to notify
-  // the delegate.
-  ResetStream();
-}
-
-void BidirectionalStreamQuicImpl::OnError(int error) {
-  // Avoid reentrancy by notifying the delegate asynchronously.
-  NotifyErrorImpl(error, /*notify_delegate_later*/ true);
-}
-
 void BidirectionalStreamQuicImpl::OnStreamReady(int rv) {
   DCHECK_NE(ERR_IO_PENDING, rv);
-  DCHECK(rv == OK || !stream_);
+  DCHECK(!stream_);
   if (rv != OK) {
     NotifyError(rv);
     return;
   }
 
-  stream_ = session_->ReleaseStream(this);
+  stream_ = session_->ReleaseStream();
 
   base::ThreadTaskRunnerHandle::Get()->PostTask(
       FROM_HERE, base::Bind(&BidirectionalStreamQuicImpl::ReadInitialHeaders,
                             weak_factory_.GetWeakPtr()));
 
   NotifyStreamReady();
 }
 
 void BidirectionalStreamQuicImpl::OnSendDataComplete(int rv) {
   CHECK(may_invoke_callbacks_);
-  DCHECK(rv == OK || !stream_);
-  if (rv != 0) {
+  DCHECK_NE(ERR_IO_PENDING, rv);
+  if (rv < 0) {
     NotifyError(rv);
     return;
   }
 
   if (delegate_)
     delegate_->OnDataSent();
 }
 
 void BidirectionalStreamQuicImpl::OnReadInitialHeadersComplete(int rv) {
   CHECK(may_invoke_callbacks_);
@@ skipping 30 matching lines @@
                  weak_factory_.GetWeakPtr()));
 
   if (rv != ERR_IO_PENDING)
     OnReadTrailingHeadersComplete(rv);
 }
 
 void BidirectionalStreamQuicImpl::OnReadTrailingHeadersComplete(int rv) {
   CHECK(may_invoke_callbacks_);
   DCHECK_NE(ERR_IO_PENDING, rv);
   if (rv < 0) {
-    NotifyError(rv);
+    if (expect_trailers_)
+      NotifyError(rv);
     return;
   }
 
   headers_bytes_received_ += rv;
 
   if (delegate_)
     delegate_->OnTrailersReceived(trailing_headers_);
 }
 
 void BidirectionalStreamQuicImpl::OnReadDataComplete(int rv) {
   CHECK(may_invoke_callbacks_);
-  DCHECK_GE(rv, 0);
+
   read_buffer_ = nullptr;
   read_buffer_len_ = 0;
 
   if (stream_->IsDoneReading()) {
     // If the write side is closed, OnFinRead() will call
     // BidirectionalStreamQuicImpl::OnClose().
     stream_->OnFinRead();
   }
 
-  if (delegate_)
+  if (!delegate_)
+    return;
+
+  if (rv < 0)
+    NotifyError(rv);
+  else
     delegate_->OnDataRead(rv);
 }
 
 void BidirectionalStreamQuicImpl::NotifyError(int error) {
   NotifyErrorImpl(error, /*notify_delegate_later*/ false);
 }
 
 void BidirectionalStreamQuicImpl::NotifyErrorImpl(int error,
                                                   bool notify_delegate_later) {
   DCHECK_NE(OK, error);
@@ skipping 20 matching lines @@
 void BidirectionalStreamQuicImpl::NotifyFailure(
     BidirectionalStreamImpl::Delegate* delegate,
     int error) {
   CHECK(may_invoke_callbacks_);
   delegate->OnFailed(error);
   // |this| might be destroyed at this point.
 }
 
 void BidirectionalStreamQuicImpl::NotifyStreamReady() {
   CHECK(may_invoke_callbacks_);
-  // Sending the request might result in the stream being closed.
-  if (send_request_headers_automatically_ && !WriteHeaders())
-    return;
+  if (send_request_headers_automatically_) {
+    int rv = WriteHeaders();
+    if (rv < 0) {
+      base::ThreadTaskRunnerHandle::Get()->PostTask(
+          FROM_HERE, base::Bind(&BidirectionalStreamQuicImpl::NotifyError,
+                                weak_factory_.GetWeakPtr(), rv));
+      return;
+    }
+  }
 
   if (delegate_)
     delegate_->OnStreamReady(has_sent_headers_);
 }
 
 void BidirectionalStreamQuicImpl::ResetStream() {
   if (!stream_)
     return;
   closed_stream_received_bytes_ = stream_->stream_bytes_read();
   closed_stream_sent_bytes_ = stream_->stream_bytes_written();
   closed_is_first_stream_ = stream_->IsFirstStream();
-  stream_->ClearDelegate();
-  stream_ = nullptr;
 }
 
 }  // namespace net