[Cronet] Upload support for async APIs

components/cronet/android/cronet_upload_data_stream_adapter.cc

+// Copyright 2015 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 "components/cronet/android/cronet_upload_data_stream_adapter.h"
+
+#include "net/base/io_buffer.h"
+#include "net/base/net_errors.h"
+
+namespace cronet {
+
+CronetUploadDataStreamAdapter::CronetUploadDataStreamAdapter(Delegate* delegate,
+                                                             int64 size)
+    : UploadDataStream(size < 0, 0),
+      size_(size),
+      waiting_on_read_(false),
+      read_in_progress_(false),
+      waiting_on_rewind_(false),
+      rewind_in_progress_(false),
+      at_front_of_stream_(true),
+      delegate_(delegate),
+      weak_factory_(this) {
+}
+
+CronetUploadDataStreamAdapter::~CronetUploadDataStreamAdapter() {
+  delegate_->OnAdapterDestroyed();
+}
+
+int CronetUploadDataStreamAdapter::InitInternal() {
+  // ResetInternal should have been called before init, if the adapter was in
+  // use.
+  DCHECK(!waiting_on_read_);
+  DCHECK(!waiting_on_rewind_);
+
+  if (!weak_factory_.HasWeakPtrs())
+    delegate_->InitializeOnNetworkThread(weak_factory_.GetWeakPtr());
+
+  // Set size of non-chunked uploads.
+  if (size_ >= 0)
+    SetSize(static_cast<uint64>(size_));
+
+  // If already at the front of the stream, nothing to do.
+  if (at_front_of_stream_) {
+    // Being at the front of the stream implies there's no read or rewind in
+    // progress.
+    DCHECK(!read_in_progress_);
+    DCHECK(!rewind_in_progress_);
+    return net::OK;
+  }
+
+  // Otherwise, the request is now waiting for the stream to be rewound.
+  waiting_on_rewind_ = true;
+
+  // Start rewinding the stream if no operation is in progress.
+  if (!read_in_progress_ && !rewind_in_progress_)
+    StartRewind();
+  return net::ERR_IO_PENDING;
+}
+
+int CronetUploadDataStreamAdapter::ReadInternal(net::IOBuffer* buf,
+                                                int buf_len) {
+  // All pending operations should have completed before a read can start.
+  DCHECK(!waiting_on_read_);
+  DCHECK(!read_in_progress_);
+  DCHECK(!waiting_on_rewind_);
+  DCHECK(!rewind_in_progress_);
+
+  DCHECK(buf);
+  DCHECK_GT(buf_len, 0);
+
+  read_in_progress_ = true;
+  waiting_on_read_ = true;
+  at_front_of_stream_ = false;
+  delegate_->Read(buf, buf_len);
+  return net::ERR_IO_PENDING;
+}
+
+void CronetUploadDataStreamAdapter::ResetInternal() {
+  // Consumer is not waiting on any operation.  Note that the active operation,
+  // if any, will continue.
+  waiting_on_read_ = false;
+  waiting_on_rewind_ = false;
+}
+
+void CronetUploadDataStreamAdapter::OnReadSuccess(int bytes_read,
+                                                  bool final_chunk) {
+  DCHECK(read_in_progress_);
+  DCHECK(!rewind_in_progress_);
+  DCHECK(bytes_read > 0 || (final_chunk && bytes_read == 0));
+  DCHECK(!is_chunked() || !final_chunk);
+
+  read_in_progress_ = false;
+
+  if (waiting_on_rewind_) {
+    DCHECK(!waiting_on_read_);
+    // Since a read just completed, can't be at the front of the stream.
+    StartRewind();
+    return;
+  }
+  // ResetInternal has been called, but still waiting on InitInternal.
+  if (!waiting_on_read_)
+    return;
+
+  waiting_on_read_ = false;
+  if (final_chunk)
+    SetIsFinalChunk();
+  OnReadCompleted(bytes_read);
+}
+
+void CronetUploadDataStreamAdapter::OnRewindSuccess() {
+  DCHECK(!waiting_on_read_);
+  DCHECK(!read_in_progress_);
+  DCHECK(rewind_in_progress_);
+  DCHECK(!at_front_of_stream_);
+
+  rewind_in_progress_ = false;
+  at_front_of_stream_ = true;
+
+  // Possible that ResetInternal was called since the rewind was started, but
+  // InitInternal has not been.
+  if (!waiting_on_rewind_)
+    return;
+
+  waiting_on_rewind_ = false;
+  OnInitCompleted(net::OK);
+}
+
+void CronetUploadDataStreamAdapter::StartRewind() {
+  DCHECK(!waiting_on_read_);
+  DCHECK(!read_in_progress_);
+  DCHECK(waiting_on_rewind_);
+  DCHECK(!rewind_in_progress_);
+  DCHECK(!at_front_of_stream_);
+
+  rewind_in_progress_ = true;
+  delegate_->Rewind();
+}
+
+}  // namespace cronet