Remove QuicChromiumClientStream::Delegate in favor of async methods.

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 {
+namespace {
+// Sets a boolean to a value, and restores it to the previous value once
+// the saver goes out of scope.
+class ScopedBoolSaver {
+ public:
+  ScopedBoolSaver(bool* var, bool new_val) : var_(var), old_val_(*var) {
+    *var_ = new_val;
+  }
 
-QuicChromiumClientStream::Handle::Handle(QuicChromiumClientStream* stream,
-                                         Delegate* delegate)
+  ~ScopedBoolSaver() { *var_ = old_val_; }
+
+ private:
+  bool* var_;
+  bool old_val_;
+};
+}  // namespace
+
+QuicChromiumClientStream::Handle::Handle(QuicChromiumClientStream* stream)
     : stream_(stream),
-      delegate_(delegate),
+      may_invoke_callbacks_(true),
       read_headers_buffer_(nullptr),
-      read_body_buffer_len_(0) {
+      read_body_buffer_len_(0),
+      net_error_(ERR_UNEXPECTED),
+      weak_factory_(this) {
   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() {
   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);
+  ResetAndRun(&read_headers_callback_, rv);
 }
 
 void QuicChromiumClientStream::Handle::OnTrailingHeadersAvailable() {
   if (!read_headers_callback_)
     return;  // Wait for ReadInitialHeaders to be called.
 
   int rv = ERR_QUIC_PROTOCOL_ERROR;
   if (!stream_->DeliverTrailingHeaders(read_headers_buffer_, &rv))
     rv = ERR_QUIC_PROTOCOL_ERROR;
 
-  ResetAndReturn(&read_headers_callback_).Run(rv);
+  ResetAndRun(&read_headers_callback_, rv);
 }
 
 void QuicChromiumClientStream::Handle::OnDataAvailable() {
   if (!read_body_callback_)
     return;  // Wait for ReadBody to be called.
 
   int rv = stream_->Read(read_body_buffer_, read_body_buffer_len_);
   if (rv == ERR_IO_PENDING)
     return;  // Spurrious, likely because of trailers?
 
   read_body_buffer_ = nullptr;
   read_body_buffer_len_ = 0;
-  ResetAndReturn(&read_body_callback_).Run(rv);
+  ResetAndRun(&read_body_callback_, rv);
 }
 
 void QuicChromiumClientStream::Handle::OnCanWrite() {
   if (!write_callback_)
     return;
 
-  base::ResetAndReturn(&write_callback_).Run(OK);
+  ResetAndRun(&write_callback_, OK);
 }
 
 void QuicChromiumClientStream::Handle::OnClose() {
+  if (net_error_ == ERR_UNEXPECTED) {
+    if (stream_error() == QUIC_STREAM_NO_ERROR &&
+        connection_error() == QUIC_NO_ERROR && fin_sent() && fin_received()) {
+      net_error_ = ERR_CONNECTION_CLOSED;
+    } else {
+      net_error_ = ERR_QUIC_PROTOCOL_ERROR;
+    }
+  }
+  OnError(net_error_);
+}
+
+void QuicChromiumClientStream::Handle::OnError(int error) {
+  net_error_ = error;
   if (stream_)
     SaveState();
   stream_ = nullptr;
-  if (delegate_) {
-    auto* delegate = delegate_;
-    delegate_ = nullptr;
-    delegate->OnClose();
-  }
+
+  // Post a task to invoke the callbacks to ensure that there is no reentrancy.
+  // A ScopedPacketBundler might cause an error which closes the stream under
+  // the call stack of the owner of the handle.
+  base::ThreadTaskRunnerHandle::Get()->PostTask(
+      FROM_HERE,
+      base::Bind(&QuicChromiumClientStream::Handle::InvokeCallbacksOnClose,
+                 weak_factory_.GetWeakPtr(), error));
 }
 
-void QuicChromiumClientStream::Handle::OnError(int error) {
-  if (stream_)
-    SaveState();
-  stream_ = nullptr;
-  if (delegate_) {
-    auto* delegate = delegate_;
-    delegate_ = nullptr;
-    delegate->OnError(error);
+void QuicChromiumClientStream::Handle::InvokeCallbacksOnClose(int error) {
+  // Invoking a callback may cause |this| to be deleted. If this happens, no
+  // more callbacks should be invoked. Guard against this by holding a WeakPtr
+  // to |this| and ensuring it's still valid.
+  auto guard(weak_factory_.GetWeakPtr());
+  for (auto* callback :
+       {&read_headers_callback_, &read_body_callback_, &write_callback_}) {
+    if (*callback)
+      ResetAndRun(callback, error);
+    if (!guard.get())
+      return;
   }
 }
 
 int QuicChromiumClientStream::Handle::ReadInitialHeaders(
     SpdyHeaderBlock* header_block,
     const CompletionCallback& callback) {
+  ScopedBoolSaver saver(&may_invoke_callbacks_, false);
   if (!stream_)
-    return ERR_CONNECTION_CLOSED;
+    return net_error_;
 
   int frame_len = 0;
   if (stream_->DeliverInitialHeaders(header_block, &frame_len))
     return frame_len;
 
   read_headers_buffer_ = header_block;
-  read_headers_callback_ = callback;
+  SetCallback(callback, &read_headers_callback_);
   return ERR_IO_PENDING;
 }
 
 int QuicChromiumClientStream::Handle::ReadBody(
     IOBuffer* buffer,
     int buffer_len,
     const CompletionCallback& callback) {
+  ScopedBoolSaver saver(&may_invoke_callbacks_, false);
+  if (IsDoneReading())
+    return OK;
+
   if (!stream_)
-    return ERR_CONNECTION_CLOSED;
+    return net_error_;
 
   int rv = stream_->Read(buffer, buffer_len);
   if (rv != ERR_IO_PENDING)
     return rv;
 
-  read_body_callback_ = callback;
+  SetCallback(callback, &read_body_callback_);
   read_body_buffer_ = buffer;
   read_body_buffer_len_ = buffer_len;
   return ERR_IO_PENDING;
 }
 
 int QuicChromiumClientStream::Handle::ReadTrailingHeaders(
     SpdyHeaderBlock* header_block,
     const CompletionCallback& callback) {
+  ScopedBoolSaver saver(&may_invoke_callbacks_, false);
   if (!stream_)
-    return ERR_CONNECTION_CLOSED;
+    return net_error_;
 
   int frame_len = 0;
   if (stream_->DeliverTrailingHeaders(header_block, &frame_len))
     return frame_len;
 
   read_headers_buffer_ = header_block;
-  read_headers_callback_ = callback;
+  SetCallback(callback, &read_headers_callback_);
   return ERR_IO_PENDING;
 }
 
-size_t QuicChromiumClientStream::Handle::WriteHeaders(
+int 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);
+  return HandleIOComplete(stream_->WriteHeaders(std::move(header_block), fin,
+                                                ack_notifier_delegate));
 }
 
 int QuicChromiumClientStream::Handle::WriteStreamData(
     base::StringPiece data,
     bool fin,
     const CompletionCallback& callback) {
+  ScopedBoolSaver saver(&may_invoke_callbacks_, false);
   if (!stream_)
-    return ERR_CONNECTION_CLOSED;
+    return net_error_;
 
   if (stream_->WriteStreamData(data, fin))
-    return OK;
+    return HandleIOComplete(OK);
 
-  write_callback_ = callback;
+  SetCallback(callback, &write_callback_);
   return ERR_IO_PENDING;
 }
 
 int QuicChromiumClientStream::Handle::WritevStreamData(
     const std::vector<scoped_refptr<IOBuffer>>& buffers,
     const std::vector<int>& lengths,
     bool fin,
     const CompletionCallback& callback) {
+  ScopedBoolSaver saver(&may_invoke_callbacks_, false);
   if (!stream_)
-    return ERR_CONNECTION_CLOSED;
+    return net_error_;
 
   if (stream_->WritevStreamData(buffers, lengths, fin))
-    return OK;
+    return HandleIOComplete(OK);
 
-  write_callback_ = callback;
+  SetCallback(callback, &write_callback_);
   return ERR_IO_PENDING;
 }
 
 int QuicChromiumClientStream::Handle::Read(IOBuffer* buf, int buf_len) {
   if (!stream_)
-    return ERR_CONNECTION_CLOSED;
+    return net_error_;
   return stream_->Read(buf, buf_len);
 }
 
 void QuicChromiumClientStream::Handle::OnFinRead() {
+  read_headers_callback_.Reset();
   if (stream_)
     stream_->OnFinRead();
 }
 
 void QuicChromiumClientStream::Handle::DisableConnectionMigration() {
   if (stream_)
     stream_->DisableConnectionMigration();
 }
 
 void QuicChromiumClientStream::Handle::SetPriority(SpdyPriority priority) {
@@ skipping 79 matching lines @@
     return priority_;
   return stream_->priority();
 }
 
 bool QuicChromiumClientStream::Handle::can_migrate() {
   if (!stream_)
     return false;
   return stream_->can_migrate();
 }
 
-QuicChromiumClientStream::Delegate*
-QuicChromiumClientStream::Handle::GetDelegate() {
-  return delegate_;
-}
-
 void QuicChromiumClientStream::Handle::SaveState() {
   DCHECK(stream_);
   fin_sent_ = stream_->fin_sent();
   fin_received_ = stream_->fin_received();
   num_bytes_consumed_ = stream_->sequencer()->NumBytesConsumed();
   id_ = stream_->id();
   connection_error_ = stream_->connection_error();
   stream_error_ = stream_->stream_error();
   is_done_reading_ = stream_->IsDoneReading();
   is_first_stream_ = stream_->IsFirstStream();
   stream_bytes_read_ = stream_->stream_bytes_read();
   stream_bytes_written_ = stream_->stream_bytes_written();
   priority_ = stream_->priority();
 }
 
+void QuicChromiumClientStream::Handle::SetCallback(
+    const CompletionCallback& new_callback,
+    CompletionCallback* callback) {
+  // TODO(rch): Convert this to a DCHECK once we ensure the API is stable and
+  // bug free.
+  CHECK(!may_invoke_callbacks_);
+  *callback = new_callback;
+}
+
+void QuicChromiumClientStream::Handle::ResetAndRun(CompletionCallback* callback,
+                                                   int rv) {
+  // TODO(rch): Convert this to a DCHECK once we ensure the API is stable and
+  // bug free.
+  CHECK(may_invoke_callbacks_);
+  ResetAndReturn(callback).Run(rv);
+}
+
+int QuicChromiumClientStream::Handle::HandleIOComplete(int rv) {
+  // If |stream_| is still valid the stream has not been closed. If the stream
+  // has not been closed, then just return |rv|.
+  if (rv < 0 || stream_)
+    return rv;
+
+  if (stream_error_ == QUIC_STREAM_NO_ERROR &&
+      connection_error_ == QUIC_NO_ERROR && fin_sent_ && fin_received_) {
+    return rv;
+  }
+
+  return net_error_;
+}
+
 QuicChromiumClientStream::QuicChromiumClientStream(
     QuicStreamId id,
     QuicClientSessionBase* session,
     const NetLogWithSource& net_log)
     : QuicSpdyStream(id, session),
       net_log_(net_log),
       handle_(nullptr),
       headers_delivered_(false),
       initial_headers_sent_(false),
       session_(session),
@@ skipping 139 matching lines @@
   // Writes the data, or buffers it.
   for (size_t i = 0; i < buffers.size(); ++i) {
     bool is_fin = fin && (i == buffers.size() - 1);
     QuicStringPiece string_data(buffers[i]->data(), lengths[i]);
     WriteOrBufferData(string_data, is_fin, nullptr);
   }
   return !HasBufferedData();  // Was all data written?
 }
 
 std::unique_ptr<QuicChromiumClientStream::Handle>
-QuicChromiumClientStream::CreateHandle(
-    QuicChromiumClientStream::Delegate* delegate) {
+QuicChromiumClientStream::CreateHandle() {
   DCHECK(!handle_);
   auto handle = std::unique_ptr<QuicChromiumClientStream::Handle>(
-      new QuicChromiumClientStream::Handle(this, delegate));
+      new QuicChromiumClientStream::Handle(this));
   handle_ = handle.get();
 
   // Should this perhaps be via PostTask to make reasoning simpler?
   if (!initial_headers_.empty())
     handle_->OnInitialHeadersAvailable();
 
   return handle;
 }
 
 void QuicChromiumClientStream::ClearHandle() {
@@ skipping 48 matching lines @@
       base::Bind(
           &QuicChromiumClientStream::NotifyHandleOfTrailingHeadersAvailable,
           weak_factory_.GetWeakPtr()));
 }
 
 void QuicChromiumClientStream::NotifyHandleOfTrailingHeadersAvailable() {
   if (!handle_)
     return;
 
   DCHECK(headers_delivered_);
-  // Post an async task to notify delegate of the FIN flag.
+  // Post an async task to notify handle of the FIN flag.
   NotifyHandleOfDataAvailableLater();
   handle_->OnTrailingHeadersAvailable();
 }
 
 bool QuicChromiumClientStream::DeliverInitialHeaders(SpdyHeaderBlock* headers,
                                                      int* frame_len) {
   if (initial_headers_.empty())
     return false;
 
   headers_delivered_ = true;
@@ skipping 37 matching lines @@
 
 void QuicChromiumClientStream::DisableConnectionMigration() {
   can_migrate_ = false;
 }
 
 bool QuicChromiumClientStream::IsFirstStream() {
   return id() == kHeadersStreamId + 2;
 }
 
 }  // namespace net