[Cronet] Implement native UploadDataStream API.

Bug: 786559
Cq-Include-Trybots: master.tryserver.chromium.android:android_cronet_tester;master.tryserver.chromium.mac:ios-simulator-cronet
Change-Id: Ied0c3de493c817e0836fcf42e09736509acbd2d1
Reviewed-on: https://chromium-review.googlesource.com/1086127
Commit-Queue: Misha Efimov <mef@chromium.org>
Reviewed-by: Paul Jensen <pauljensen@chromium.org>
Cr-Commit-Position: refs/heads/master@{#584464}

components/cronet/native/BUILD.gn

@@ -55,6 +55,8 @@ source_set("cronet_native_impl") {
     "io_buffer_with_cronet_buffer.h",
     "runnables.cc",
     "runnables.h",
+    "upload_data_sink.cc",
+    "upload_data_sink.h",
     "url_request.cc",
     "url_request.h",
 

components/cronet/native/buffer.cc

@@ -4,9 +4,9 @@
 
 #include "components/cronet/native/generated/cronet.idl_impl_interface.h"
 
-#include "base/lazy_instance.h"
 #include "base/logging.h"
 #include "base/macros.h"
+#include "base/no_destructor.h"
 #include "base/numerics/safe_conversions.h"
 
 namespace {
@@ -23,9 +23,6 @@ class Cronet_BufferCallbackFree : public Cronet_BufferCallback {
   DISALLOW_COPY_AND_ASSIGN(Cronet_BufferCallbackFree);
 };
 
-base::LazyInstance<Cronet_BufferCallbackFree>::Leaky
-    g_cronet_buffer_callback_free = LAZY_INSTANCE_INITIALIZER;
-
 // Concrete implementation of abstract Cronet_Buffer interface.
 class Cronet_BufferImpl : public Cronet_Buffer {
  public:
@@ -69,7 +66,8 @@ void Cronet_BufferImpl::InitWithAlloc(uint64_t size) {
   if (!data_)
     return;
   size_ = size;
-  callback_ = g_cronet_buffer_callback_free.Pointer();
+  static base::NoDestructor<Cronet_BufferCallbackFree> static_callback;
+  callback_ = static_callback.get();
 }
 
 uint64_t Cronet_BufferImpl::GetSize() {

components/cronet/native/cronet.idl

@@ -855,28 +855,29 @@ interface UploadDataSink {
   /**
    * Called by UploadDataProvider when a read succeeds.
    *
+   * @param bytes_read number of bytes read into buffer passed to read().
    * @param final_chunk For chunked uploads, |true| if this is the final
    *     read. It must be |false| for non-chunked uploads.
    */
-  OnReadSucceeded(bool final_chunk);
+  OnReadSucceeded(uint64 bytes_read, bool final_chunk);
 
   /**
    * Called by UploadDataProvider when a read fails.
-   * @param error to pass on to UrlRequestCallback.onFailed().
+   * @param error_message to pass on to UrlRequestCallback.onFailed().
    */
-  OnReadError(Error error);
+  OnReadError(string error_message);
 
   /**
    * Called by UploadDataProvider when a rewind succeeds.
    */
-  OnRewindSucceded();
+  OnRewindSucceeded();
 
   /**
    * Called by UploadDataProvider when a rewind fails, or if rewinding
    * uploads is not supported.
-   * @param error to pass on to UrlRequestCallback.onFailed().
+   * @param error_message to pass on to UrlRequestCallback.onFailed().
    */
-  OnRewindError(Error error);
+  OnRewindError(string error_message);
 };
 
 /**
@@ -898,9 +899,7 @@ interface UploadDataProvider {
   GetLength() => (int64 length);
 
   /**
-   * Reads upload data into |buffer|. Upon completion, the buffer's
-   * position is updated to the end of the bytes that were read. The buffer's
-   * limit is not changed. Each call of this method must be followed be a
+   * Reads upload data into |buffer|. Each call of this method must be followed be a
    * single call, either synchronous or asynchronous, to
    * UploadDataSink.onReadSucceeded() on success
    * or UploadDataSink.onReadError() on failure. Neither read nor rewind
@@ -910,8 +909,7 @@ interface UploadDataProvider {
    *
    * @param upload_data_sink The object to notify when the read has completed,
    *            successfully or otherwise.
-   * @param buffer The buffer to copy the read bytes into. Do not change
-   *            byteBuffer's limit.
+   * @param buffer The buffer to copy the read bytes into.
    */
   Read(UploadDataSink upload_data_sink, Buffer buffer);
 

components/cronet/native/generated/cronet.idl_c.h

@@ -468,34 +468,36 @@ Cronet_UploadDataSink_GetClientContext(Cronet_UploadDataSinkPtr self);
 // The app calls them to manipulate Cronet_UploadDataSink.
 CRONET_EXPORT
 void Cronet_UploadDataSink_OnReadSucceeded(Cronet_UploadDataSinkPtr self,
+                                           uint64_t bytes_read,
                                            bool final_chunk);
 CRONET_EXPORT
 void Cronet_UploadDataSink_OnReadError(Cronet_UploadDataSinkPtr self,
-                                       Cronet_ErrorPtr error);
+                                       Cronet_String error_message);
 CRONET_EXPORT
-void Cronet_UploadDataSink_OnRewindSucceded(Cronet_UploadDataSinkPtr self);
+void Cronet_UploadDataSink_OnRewindSucceeded(Cronet_UploadDataSinkPtr self);
 CRONET_EXPORT
 void Cronet_UploadDataSink_OnRewindError(Cronet_UploadDataSinkPtr self,
-                                         Cronet_ErrorPtr error);
+                                         Cronet_String error_message);
 // Concrete interface Cronet_UploadDataSink is implemented by Cronet.
 // The app can implement these for testing / mocking.
 typedef void (*Cronet_UploadDataSink_OnReadSucceededFunc)(
     Cronet_UploadDataSinkPtr self,
+    uint64_t bytes_read,
     bool final_chunk);
 typedef void (*Cronet_UploadDataSink_OnReadErrorFunc)(
     Cronet_UploadDataSinkPtr self,
-    Cronet_ErrorPtr error);
-typedef void (*Cronet_UploadDataSink_OnRewindSuccededFunc)(
+    Cronet_String error_message);
+typedef void (*Cronet_UploadDataSink_OnRewindSucceededFunc)(
     Cronet_UploadDataSinkPtr self);
 typedef void (*Cronet_UploadDataSink_OnRewindErrorFunc)(
     Cronet_UploadDataSinkPtr self,
-    Cronet_ErrorPtr error);
+    Cronet_String error_message);
 // Concrete interface Cronet_UploadDataSink is implemented by Cronet.
 // The app can use this for testing / mocking.
 CRONET_EXPORT Cronet_UploadDataSinkPtr Cronet_UploadDataSink_CreateWith(
     Cronet_UploadDataSink_OnReadSucceededFunc OnReadSucceededFunc,
     Cronet_UploadDataSink_OnReadErrorFunc OnReadErrorFunc,
-    Cronet_UploadDataSink_OnRewindSuccededFunc OnRewindSuccededFunc,
+    Cronet_UploadDataSink_OnRewindSucceededFunc OnRewindSucceededFunc,
     Cronet_UploadDataSink_OnRewindErrorFunc OnRewindErrorFunc);
 
 ///////////////////////

components/cronet/native/generated/cronet.idl_impl_interface.cc

@@ -577,26 +577,27 @@ Cronet_ClientContext Cronet_UploadDataSink_GetClientContext(
 }
 
 void Cronet_UploadDataSink_OnReadSucceeded(Cronet_UploadDataSinkPtr self,
+                                           uint64_t bytes_read,
                                            bool final_chunk) {
   DCHECK(self);
-  self->OnReadSucceeded(final_chunk);
+  self->OnReadSucceeded(bytes_read, final_chunk);
 }
 
 void Cronet_UploadDataSink_OnReadError(Cronet_UploadDataSinkPtr self,
-                                       Cronet_ErrorPtr error) {
+                                       Cronet_String error_message) {
   DCHECK(self);
-  self->OnReadError(error);
+  self->OnReadError(error_message);
 }
 
-void Cronet_UploadDataSink_OnRewindSucceded(Cronet_UploadDataSinkPtr self) {
+void Cronet_UploadDataSink_OnRewindSucceeded(Cronet_UploadDataSinkPtr self) {
   DCHECK(self);
-  self->OnRewindSucceded();
+  self->OnRewindSucceeded();
 }
 
 void Cronet_UploadDataSink_OnRewindError(Cronet_UploadDataSinkPtr self,
-                                         Cronet_ErrorPtr error) {
+                                         Cronet_String error_message) {
   DCHECK(self);
-  self->OnRewindError(error);
+  self->OnRewindError(error_message);
 }
 
 // Implementation of Cronet_UploadDataSink that forwards calls to C functions
@@ -606,34 +607,34 @@ class Cronet_UploadDataSinkStub : public Cronet_UploadDataSink {
   Cronet_UploadDataSinkStub(
       Cronet_UploadDataSink_OnReadSucceededFunc OnReadSucceededFunc,
       Cronet_UploadDataSink_OnReadErrorFunc OnReadErrorFunc,
-      Cronet_UploadDataSink_OnRewindSuccededFunc OnRewindSuccededFunc,
+      Cronet_UploadDataSink_OnRewindSucceededFunc OnRewindSucceededFunc,
       Cronet_UploadDataSink_OnRewindErrorFunc OnRewindErrorFunc)
       : OnReadSucceededFunc_(OnReadSucceededFunc),
         OnReadErrorFunc_(OnReadErrorFunc),
-        OnRewindSuccededFunc_(OnRewindSuccededFunc),
+        OnRewindSucceededFunc_(OnRewindSucceededFunc),
         OnRewindErrorFunc_(OnRewindErrorFunc) {}
 
   ~Cronet_UploadDataSinkStub() override {}
 
  protected:
-  void OnReadSucceeded(bool final_chunk) override {
-    OnReadSucceededFunc_(this, final_chunk);
+  void OnReadSucceeded(uint64_t bytes_read, bool final_chunk) override {
+    OnReadSucceededFunc_(this, bytes_read, final_chunk);
   }
 
-  void OnReadError(Cronet_ErrorPtr error) override {
-    OnReadErrorFunc_(this, error);
+  void OnReadError(Cronet_String error_message) override {
+    OnReadErrorFunc_(this, error_message);
   }
 
-  void OnRewindSucceded() override { OnRewindSuccededFunc_(this); }
+  void OnRewindSucceeded() override { OnRewindSucceededFunc_(this); }
 
-  void OnRewindError(Cronet_ErrorPtr error) override {
-    OnRewindErrorFunc_(this, error);
+  void OnRewindError(Cronet_String error_message) override {
+    OnRewindErrorFunc_(this, error_message);
   }
 
  private:
   const Cronet_UploadDataSink_OnReadSucceededFunc OnReadSucceededFunc_;
   const Cronet_UploadDataSink_OnReadErrorFunc OnReadErrorFunc_;
-  const Cronet_UploadDataSink_OnRewindSuccededFunc OnRewindSuccededFunc_;
+  const Cronet_UploadDataSink_OnRewindSucceededFunc OnRewindSucceededFunc_;
   const Cronet_UploadDataSink_OnRewindErrorFunc OnRewindErrorFunc_;
 
   DISALLOW_COPY_AND_ASSIGN(Cronet_UploadDataSinkStub);
@@ -642,10 +643,11 @@ class Cronet_UploadDataSinkStub : public Cronet_UploadDataSink {
 Cronet_UploadDataSinkPtr Cronet_UploadDataSink_CreateWith(
     Cronet_UploadDataSink_OnReadSucceededFunc OnReadSucceededFunc,
     Cronet_UploadDataSink_OnReadErrorFunc OnReadErrorFunc,
-    Cronet_UploadDataSink_OnRewindSuccededFunc OnRewindSuccededFunc,
+    Cronet_UploadDataSink_OnRewindSucceededFunc OnRewindSucceededFunc,
     Cronet_UploadDataSink_OnRewindErrorFunc OnRewindErrorFunc) {
   return new Cronet_UploadDataSinkStub(OnReadSucceededFunc, OnReadErrorFunc,
-                                       OnRewindSuccededFunc, OnRewindErrorFunc);
+                                       OnRewindSucceededFunc,
+                                       OnRewindErrorFunc);
 }
 
 // C functions of Cronet_UploadDataProvider that forward calls to C++

components/cronet/native/generated/cronet.idl_impl_interface.h

@@ -166,10 +166,10 @@ struct Cronet_UploadDataSink {
   }
   Cronet_ClientContext client_context() const { return client_context_; }
 
-  virtual void OnReadSucceeded(bool final_chunk) = 0;
-  virtual void OnReadError(Cronet_ErrorPtr error) = 0;
-  virtual void OnRewindSucceded() = 0;
-  virtual void OnRewindError(Cronet_ErrorPtr error) = 0;
+  virtual void OnReadSucceeded(uint64_t bytes_read, bool final_chunk) = 0;
+  virtual void OnReadError(Cronet_String error_message) = 0;
+  virtual void OnRewindSucceeded() = 0;
+  virtual void OnRewindError(Cronet_String error_message) = 0;
 
  private:
   Cronet_ClientContext client_context_ = nullptr;

components/cronet/native/generated/cronet.idl_impl_interface_unittest.cc

@@ -66,7 +66,7 @@ Cronet_RawDataPtr TestCronet_Buffer_GetData(Cronet_BufferPtr self) {
   CHECK(test);
   test->GetData_called_ = true;
 
-  return static_cast<Cronet_RawDataPtr>(nullptr);
+  return static_cast<Cronet_RawDataPtr>(0);
 }
 }  // namespace
 
@@ -276,7 +276,7 @@ Cronet_String TestCronet_Engine_GetVersionString(Cronet_EnginePtr self) {
   CHECK(test);
   test->GetVersionString_called_ = true;
 
-  return static_cast<Cronet_String>(nullptr);
+  return static_cast<Cronet_String>(0);
 }
 Cronet_String TestCronet_Engine_GetDefaultUserAgent(Cronet_EnginePtr self) {
   CHECK(self);
@@ -285,7 +285,7 @@ Cronet_String TestCronet_Engine_GetDefaultUserAgent(Cronet_EnginePtr self) {
   CHECK(test);
   test->GetDefaultUserAgent_called_ = true;
 
-  return static_cast<Cronet_String>(nullptr);
+  return static_cast<Cronet_String>(0);
 }
 }  // namespace
 
@@ -485,7 +485,7 @@ class Cronet_UploadDataSinkTest : public ::testing::Test {
  public:
   bool OnReadSucceeded_called_ = false;
   bool OnReadError_called_ = false;
-  bool OnRewindSucceded_called_ = false;
+  bool OnRewindSucceeded_called_ = false;
   bool OnRewindError_called_ = false;
 
  private:
@@ -495,6 +495,7 @@ class Cronet_UploadDataSinkTest : public ::testing::Test {
 namespace {
 // Implementation of Cronet_UploadDataSink methods for testing.
 void TestCronet_UploadDataSink_OnReadSucceeded(Cronet_UploadDataSinkPtr self,
+                                               uint64_t bytes_read,
                                                bool final_chunk) {
   CHECK(self);
   Cronet_ClientContext client_context =
@@ -504,7 +505,7 @@ void TestCronet_UploadDataSink_OnReadSucceeded(Cronet_UploadDataSinkPtr self,
   test->OnReadSucceeded_called_ = true;
 }
 void TestCronet_UploadDataSink_OnReadError(Cronet_UploadDataSinkPtr self,
-                                           Cronet_ErrorPtr error) {
+                                           Cronet_String error_message) {
   CHECK(self);
   Cronet_ClientContext client_context =
       Cronet_UploadDataSink_GetClientContext(self);
@@ -512,16 +513,17 @@ void TestCronet_UploadDataSink_OnReadError(Cronet_UploadDataSinkPtr self,
   CHECK(test);
   test->OnReadError_called_ = true;
 }
-void TestCronet_UploadDataSink_OnRewindSucceded(Cronet_UploadDataSinkPtr self) {
+void TestCronet_UploadDataSink_OnRewindSucceeded(
+    Cronet_UploadDataSinkPtr self) {
   CHECK(self);
   Cronet_ClientContext client_context =
       Cronet_UploadDataSink_GetClientContext(self);
   auto* test = static_cast<Cronet_UploadDataSinkTest*>(client_context);
   CHECK(test);
-  test->OnRewindSucceded_called_ = true;
+  test->OnRewindSucceeded_called_ = true;
 }
 void TestCronet_UploadDataSink_OnRewindError(Cronet_UploadDataSinkPtr self,
-                                             Cronet_ErrorPtr error) {
+                                             Cronet_String error_message) {
   CHECK(self);
   Cronet_ClientContext client_context =
       Cronet_UploadDataSink_GetClientContext(self);
@@ -536,14 +538,14 @@ TEST_F(Cronet_UploadDataSinkTest, TestCreate) {
   Cronet_UploadDataSinkPtr test = Cronet_UploadDataSink_CreateWith(
       TestCronet_UploadDataSink_OnReadSucceeded,
       TestCronet_UploadDataSink_OnReadError,
-      TestCronet_UploadDataSink_OnRewindSucceded,
+      TestCronet_UploadDataSink_OnRewindSucceeded,
       TestCronet_UploadDataSink_OnRewindError);
   CHECK(test);
   Cronet_UploadDataSink_SetClientContext(test, this);
   CHECK(!OnReadSucceeded_called_);
   CHECK(!OnReadError_called_);
-  Cronet_UploadDataSink_OnRewindSucceded(test);
-  CHECK(OnRewindSucceded_called_);
+  Cronet_UploadDataSink_OnRewindSucceeded(test);
+  CHECK(OnRewindSucceeded_called_);
   CHECK(!OnRewindError_called_);
 
   Cronet_UploadDataSink_Destroy(test);

components/cronet/native/io_buffer_with_cronet_buffer.cc

@@ -4,8 +4,25 @@
 
 #include "components/cronet/native/io_buffer_with_cronet_buffer.h"
 
+#include "base/no_destructor.h"
 #include "components/cronet/native/generated/cronet.idl_impl_interface.h"
 
+namespace {
+// Implementation of Cronet_BufferCallback that doesn't free the data as it
+// is not owned by the buffer.
+class Cronet_BufferCallbackUnowned : public Cronet_BufferCallback {
+ public:
+  Cronet_BufferCallbackUnowned() = default;
+  ~Cronet_BufferCallbackUnowned() override = default;
+
+  void OnDestroy(Cronet_BufferPtr buffer) override {}
+
+ private:
+  DISALLOW_COPY_AND_ASSIGN(Cronet_BufferCallbackUnowned);
+};
+
+}  // namespace
+
 namespace cronet {
 
 IOBufferWithCronet_Buffer::IOBufferWithCronet_Buffer(
@@ -25,4 +42,16 @@ Cronet_BufferPtr IOBufferWithCronet_Buffer::Release() {
   return cronet_buffer_.release();
 }
 
+Cronet_BufferWithIOBuffer::Cronet_BufferWithIOBuffer(net::IOBuffer* io_buffer,
+                                                     size_t io_buffer_len)
+    : io_buffer_(io_buffer),
+      io_buffer_len_(io_buffer_len),
+      cronet_buffer_(Cronet_Buffer_Create()) {
+  static base::NoDestructor<Cronet_BufferCallbackUnowned> static_callback;
+  cronet_buffer_->InitWithDataAndCallback(io_buffer->data(), io_buffer_len,
+                                          static_callback.get());
+}
+
+Cronet_BufferWithIOBuffer::~Cronet_BufferWithIOBuffer() = default;
+
 }  // namespace cronet

components/cronet/native/io_buffer_with_cronet_buffer.h

@@ -32,6 +32,29 @@ class IOBufferWithCronet_Buffer : public net::WrappedIOBuffer {
   DISALLOW_COPY_AND_ASSIGN(IOBufferWithCronet_Buffer);
 };
 
+// Represents a Cronet_Buffer backed by a net::IOBuffer. Keeps both the
+// net::IOBuffer and the Cronet_Buffer object alive until destroyed.
+class Cronet_BufferWithIOBuffer {
+ public:
+  Cronet_BufferWithIOBuffer(net::IOBuffer* io_buffer, size_t io_buffer_len);
+  ~Cronet_BufferWithIOBuffer();
+
+  const net::IOBuffer* io_buffer() const { return io_buffer_.get(); }
+  size_t io_buffer_len() const { return io_buffer_len_; }
+
+  // Returns pointer to Cronet buffer owned by |this|.
+  Cronet_BufferPtr cronet_buffer() { return cronet_buffer_.get(); }
+
+ private:
+  scoped_refptr<net::IOBuffer> io_buffer_;
+  size_t io_buffer_len_;
+
+  // Cronet buffer owned by |this|.
+  std::unique_ptr<Cronet_Buffer> cronet_buffer_;
+
+  DISALLOW_COPY_AND_ASSIGN(Cronet_BufferWithIOBuffer);
+};
+
 }  // namespace cronet
 
 #endif  // COMPONENTS_CRONET_NATIVE_IO_BUFFER_WITH_CRONET_BUFFER_H_

components/cronet/native/test/BUILD.gn

@@ -44,6 +44,8 @@ source_set("cronet_native_tests") {
     "buffer_test.cc",
     "engine_test.cc",
     "executors_test.cc",
+    "test_upload_data_provider.cc",
+    "test_upload_data_provider.h",
     "test_url_request_callback.cc",
     "test_url_request_callback.h",
     "url_request_test.cc",

components/cronet/native/test/test_upload_data_provider.cc

+// Copyright 2018 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/native/test/test_upload_data_provider.h"
+
+#include "base/bind.h"
+#include "testing/gtest/include/gtest/gtest.h"
+
+namespace {
+
+// Helper class that runs base::OnceClosure.
+class TestRunnable {
+ public:
+  // Creates Cronet runnable that runs |task| once and destroys itself.
+  static Cronet_RunnablePtr CreateRunnable(base::OnceClosure task);
+
+ private:
+  explicit TestRunnable(base::OnceClosure task);
+  ~TestRunnable();
+
+  // Runs |self| and destroys it.
+  static void Run(Cronet_RunnablePtr self);
+
+  // Closure to run.
+  base::OnceClosure task_;
+
+  DISALLOW_COPY_AND_ASSIGN(TestRunnable);
+};
+
+TestRunnable::TestRunnable(base::OnceClosure task) : task_(std::move(task)) {}
+
+TestRunnable::~TestRunnable() = default;
+
+// static
+Cronet_RunnablePtr TestRunnable::CreateRunnable(base::OnceClosure task) {
+  Cronet_RunnablePtr runnable = Cronet_Runnable_CreateWith(TestRunnable::Run);
+  Cronet_Runnable_SetClientContext(runnable, new TestRunnable(std::move(task)));
+  return runnable;
+}
+
+// static
+void TestRunnable::Run(Cronet_RunnablePtr self) {
+  CHECK(self);
+  Cronet_ClientContext context = Cronet_Runnable_GetClientContext(self);
+  TestRunnable* runnable = static_cast<TestRunnable*>(context);
+  CHECK(runnable);
+  std::move(runnable->task_).Run();
+  delete runnable;
+  Cronet_Runnable_Destroy(self);
+}
+
+}  // namespace
+
+namespace cronet {
+// Various test utility functions for testing Cronet.
+namespace test {
+
+TestUploadDataProvider::TestUploadDataProvider(
+    SuccessCallbackMode success_callback_mode,
+    Cronet_ExecutorPtr executor)
+    : success_callback_mode_(success_callback_mode), executor_(executor) {}
+
+TestUploadDataProvider::~TestUploadDataProvider() = default;
+
+Cronet_UploadDataProviderPtr
+TestUploadDataProvider::CreateUploadDataProvider() {
+  Cronet_UploadDataProviderPtr upload_data_provider =
+      Cronet_UploadDataProvider_CreateWith(
+          TestUploadDataProvider::GetLength, TestUploadDataProvider::Read,
+          TestUploadDataProvider::Rewind, TestUploadDataProvider::Close);
+  Cronet_UploadDataProvider_SetClientContext(upload_data_provider, this);
+  return upload_data_provider;
+}
+
+void TestUploadDataProvider::AddRead(std::string read) {
+  EXPECT_TRUE(!started_) << "Adding bytes after read";
+  reads_.push_back(read);
+}
+
+void TestUploadDataProvider::SetReadFailure(int read_fail_index,
+                                            FailMode read_fail_mode) {
+  read_fail_index_ = read_fail_index;
+  read_fail_mode_ = read_fail_mode;
+}
+
+void TestUploadDataProvider::SetRewindFailure(FailMode rewind_fail_mode) {
+  rewind_fail_mode_ = rewind_fail_mode;
+}
+
+int64_t TestUploadDataProvider::GetLength() const {
+  EXPECT_TRUE(!closed_.IsSet()) << "Data Provider is closed";
+  if (bad_length_ != -1)
+    return bad_length_;
+
+  return GetUploadedLength();
+}
+
+int64_t TestUploadDataProvider::GetUploadedLength() const {
+  if (chunked_)
+    return -1ll;
+
+  int64_t length = 0ll;
+  for (const auto& read : reads_)
+    length += read.size();
+
+  return length;
+}
+
+void TestUploadDataProvider::Read(Cronet_UploadDataSinkPtr upload_data_sink,
+                                  Cronet_BufferPtr buffer) {
+  int current_read_call = num_read_calls_;
+  ++num_read_calls_;
+  EXPECT_TRUE(!closed_.IsSet()) << "Data Provider is closed";
+
+  AssertIdle();
+
+  if (MaybeFailRead(current_read_call, upload_data_sink)) {
+    failed_ = true;
+    return;
+  }
+
+  read_pending_ = true;
+  started_ = true;
+
+  bool final_chunk = (chunked_ && next_read_ == reads_.size() - 1);
+  EXPECT_TRUE(next_read_ < reads_.size()) << "Too many reads: " << next_read_;
+  const auto& read = reads_[next_read_];
+  EXPECT_TRUE(read.size() < Cronet_Buffer_GetSize(buffer))
+      << "Read buffer smaller than expected.";
+  memcpy(Cronet_Buffer_GetData(buffer), read.data(), read.size());
+  ++next_read_;
+
+  auto complete_closure = base::BindOnce(
+      [](TestUploadDataProvider* upload_data_provider,
+         Cronet_UploadDataSink* upload_data_sink, uint64_t bytes_read,
+         bool final_chunk) {
+        upload_data_provider->read_pending_ = false;
+        Cronet_UploadDataSink_OnReadSucceeded(upload_data_sink, bytes_read,
+                                              final_chunk);
+      },
+      this, upload_data_sink, read.size(), final_chunk);
+
+  if (success_callback_mode_ == SYNC) {
+    std::move(complete_closure).Run();
+  } else {
+    PostTaskToExecutor(std::move(complete_closure));
+  }
+}
+
+void TestUploadDataProvider::Rewind(Cronet_UploadDataSinkPtr upload_data_sink) {
+  ++num_rewind_calls_;
+  EXPECT_TRUE(!closed_.IsSet()) << "Data Provider is closed";
+  AssertIdle();
+
+  if (MaybeFailRewind(upload_data_sink)) {
+    failed_ = true;
+    return;
+  }
+
+  // Should never try and rewind when rewinding does nothing.
+  EXPECT_TRUE(next_read_ != 0) << "Unexpected rewind when already at beginning";
+
+  rewind_pending_ = true;
+  next_read_ = 0;
+
+  auto complete_closure = base::BindOnce(
+      [](TestUploadDataProvider* upload_data_provider,
+         Cronet_UploadDataSink* upload_data_sink) {
+        upload_data_provider->rewind_pending_ = false;
+        Cronet_UploadDataSink_OnRewindSucceeded(upload_data_sink);
+      },
+      this, upload_data_sink);
+
+  if (success_callback_mode_ == SYNC) {
+    std::move(complete_closure).Run();
+  } else {
+    PostTaskToExecutor(std::move(complete_closure));
+  }
+}
+
+void TestUploadDataProvider::PostTaskToExecutor(base::OnceClosure task) {
+  EXPECT_TRUE(executor_);
+  // |runnable| is passed to executor, which destroys it after execution.
+  Cronet_Executor_Execute(executor_,
+                          TestRunnable::CreateRunnable(std::move(task)));
+}
+
+void TestUploadDataProvider::AssertIdle() const {
+  EXPECT_TRUE(!read_pending_) << "Unexpected operation during read";
+  EXPECT_TRUE(!rewind_pending_) << "Unexpected operation during rewind";
+  EXPECT_TRUE(!failed_) << "Unexpected operation after failure";
+}
+
+bool TestUploadDataProvider::MaybeFailRead(
+    int read_index,
+    Cronet_UploadDataSinkPtr upload_data_sink) {
+  if (read_index != read_fail_index_)
+    return false;
+  if (read_fail_mode_ == NONE)
+    return false;
+
+  if (read_fail_mode_ == CALLBACK_SYNC) {
+    Cronet_UploadDataSink_OnReadError(upload_data_sink, "Sync read failure");
+    return true;
+  }
+  EXPECT_EQ(read_fail_mode_, CALLBACK_ASYNC);
+
+  PostTaskToExecutor(base::BindOnce(
+      [](Cronet_UploadDataSink* upload_data_sink) {
+        Cronet_UploadDataSink_OnReadError(upload_data_sink,
+                                          "Async read failure");
+      },
+      upload_data_sink));
+  return true;
+}
+
+bool TestUploadDataProvider::MaybeFailRewind(
+    Cronet_UploadDataSinkPtr upload_data_sink) {
+  if (rewind_fail_mode_ == NONE)
+    return false;
+
+  if (rewind_fail_mode_ == CALLBACK_SYNC) {
+    Cronet_UploadDataSink_OnRewindError(upload_data_sink,
+                                        "Sync rewind failure");
+    return true;
+  }
+  EXPECT_EQ(rewind_fail_mode_, CALLBACK_ASYNC);
+
+  PostTaskToExecutor(base::BindOnce(
+      [](Cronet_UploadDataSink* upload_data_sink) {
+        Cronet_UploadDataSink_OnRewindError(upload_data_sink,
+                                            "Async rewind failure");
+      },
+      upload_data_sink));
+  return true;
+}
+
+void TestUploadDataProvider::Close() {
+  EXPECT_TRUE(!closed_.IsSet()) << "Closed twice";
+  closed_.Set();
+  awaiting_close_.Signal();
+}
+
+void TestUploadDataProvider::AssertClosed() {
+  awaiting_close_.TimedWait(base::TimeDelta::FromMilliseconds(5000));
+  EXPECT_TRUE(closed_.IsSet()) << "Was not closed";
+}
+
+/* static */
+TestUploadDataProvider* TestUploadDataProvider::GetThis(
+    Cronet_UploadDataProviderPtr self) {
+  return static_cast<TestUploadDataProvider*>(
+      Cronet_UploadDataProvider_GetClientContext(self));
+}
+
+/* static */
+int64_t TestUploadDataProvider::GetLength(Cronet_UploadDataProviderPtr self) {
+  return GetThis(self)->GetLength();
+}
+
+/* static */
+void TestUploadDataProvider::Read(Cronet_UploadDataProviderPtr self,
+                                  Cronet_UploadDataSinkPtr upload_data_sink,
+                                  Cronet_BufferPtr buffer) {
+  return GetThis(self)->Read(upload_data_sink, buffer);
+}
+
+/* static */
+void TestUploadDataProvider::Rewind(Cronet_UploadDataProviderPtr self,
+                                    Cronet_UploadDataSinkPtr upload_data_sink) {
+  return GetThis(self)->Rewind(upload_data_sink);
+}
+
+/* static */
+void TestUploadDataProvider::Close(Cronet_UploadDataProviderPtr self) {
+  return GetThis(self)->Close();
+}
+
+}  // namespace test
+}  // namespace cronet

components/cronet/native/test/test_upload_data_provider.h

+// Copyright 2018 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 COMPONENTS_CRONET_NATIVE_TEST_TEST_UPLOAD_DATA_PROVIDER_H_
+#define COMPONENTS_CRONET_NATIVE_TEST_TEST_UPLOAD_DATA_PROVIDER_H_
+
+#include <memory>
+#include <string>
+#include <utility>
+#include <vector>
+
+#include "cronet_c.h"
+
+#include "base/bind.h"
+#include "base/macros.h"
+#include "base/synchronization/atomic_flag.h"
+#include "base/synchronization/lock.h"
+#include "base/synchronization/waitable_event.h"
+#include "base/threading/thread.h"
+#include "testing/gtest/include/gtest/gtest.h"
+
+namespace cronet {
+// Various test utility functions for testing Cronet.
+namespace test {
+
+/**
+ * An UploadDataProvider implementation used in tests.
+ */
+class TestUploadDataProvider {
+ public:
+  // Indicates whether all success callbacks are synchronous or asynchronous.
+  // Doesn't apply to errors.
+  enum SuccessCallbackMode { SYNC, ASYNC };
+
+  // Indicates whether failures should invoke callbacks synchronously, or
+  // invoke callback asynchronously.
+  enum FailMode { NONE, CALLBACK_SYNC, CALLBACK_ASYNC };
+
+  TestUploadDataProvider(SuccessCallbackMode success_callback_mode,
+                         Cronet_ExecutorPtr executor);
+
+  ~TestUploadDataProvider();
+
+  Cronet_UploadDataProviderPtr CreateUploadDataProvider();
+
+  // Adds the result to be returned by a successful read request.  The
+  // returned bytes must all fit within the read buffer provided by Cronet.
+  // After a rewind, if there is one, all reads will be repeated.
+  void AddRead(std::string read);
+
+  void SetReadFailure(int read_fail_index, FailMode read_fail_mode);
+
+  void SetRewindFailure(FailMode rewind_fail_mode);
+
+  void set_bad_length(int64_t bad_length) { bad_length_ = bad_length; }
+
+  void set_chunked(bool chunked) { chunked_ = chunked; }
+
+  int num_read_calls() const { return num_read_calls_; }
+
+  int num_rewind_calls() const { return num_rewind_calls_; }
+
+  /**
+   * Returns the cumulative length of all data added by calls to addRead.
+   */
+  int64_t GetLength() const;
+
+  int64_t GetUploadedLength() const;
+
+  void Read(Cronet_UploadDataSinkPtr upload_data_sink, Cronet_BufferPtr buffer);
+
+  void Rewind(Cronet_UploadDataSinkPtr upload_data_sink);
+
+  void AssertClosed();
+
+ private:
+  void PostTaskToExecutor(base::OnceClosure task);
+
+  void AssertIdle() const;
+
+  bool MaybeFailRead(int read_index, Cronet_UploadDataSinkPtr upload_data_sink);
+
+  bool MaybeFailRewind(Cronet_UploadDataSinkPtr upload_data_sink);
+
+  void Close();
+
+  // Implementation of Cronet_UploadDataProvider methods.
+  static TestUploadDataProvider* GetThis(Cronet_UploadDataProviderPtr self);
+
+  static int64_t GetLength(Cronet_UploadDataProviderPtr self);
+  static void Read(Cronet_UploadDataProviderPtr self,
+                   Cronet_UploadDataSinkPtr upload_data_sink,
+                   Cronet_BufferPtr buffer);
+  static void Rewind(Cronet_UploadDataProviderPtr self,
+                     Cronet_UploadDataSinkPtr upload_data_sink);
+  static void Close(Cronet_UploadDataProviderPtr self);
+
+  std::vector<std::string> reads_;
+  const SuccessCallbackMode success_callback_mode_ = SYNC;
+  const Cronet_ExecutorPtr executor_;
+
+  bool chunked_ = false;
+
+  // Index of read to fail on.
+  int read_fail_index_ = -1;
+  // Indicates how to fail on a read.
+  FailMode read_fail_mode_ = NONE;
+  FailMode rewind_fail_mode_ = NONE;
+  // Report bad length if not set to -1.
+  int64_t bad_length_ = -1;
+
+  int num_read_calls_ = 0;
+  int num_rewind_calls_ = 0;
+
+  size_t next_read_ = 0;
+  bool started_ = false;
+  bool read_pending_ = false;
+  bool rewind_pending_ = false;
+  // Used to ensure there are no read/rewind requests after a failure.
+  bool failed_ = false;
+
+  base::AtomicFlag closed_;
+  base::WaitableEvent awaiting_close_;
+};
+
+}  // namespace test
+}  // namespace cronet
+
+#endif  // COMPONENTS_CRONET_NATIVE_TEST_TEST_UPLOAD_DATA_PROVIDER_H_

components/cronet/native/test/test_url_request_callback.cc

@@ -68,15 +68,21 @@ TestUrlRequestCallback::~TestUrlRequestCallback() {
 }
 
 Cronet_ExecutorPtr TestUrlRequestCallback::GetExecutor(bool direct) {
-  CHECK(!executor_);
+  if (executor_) {
+    CHECK(direct == allow_direct_executor_);
+    return executor_;
+  }
   allow_direct_executor_ = direct;
-  if (direct)
-    return Cronet_Executor_CreateWith(TestUrlRequestCallback::ExecuteDirect);
+  if (direct) {
+    executor_ =
+        Cronet_Executor_CreateWith(TestUrlRequestCallback::ExecuteDirect);
+  } else {
     executor_thread_ =
         std::make_unique<base::Thread>("TestUrlRequestCallback executor");
     executor_thread_->Start();
     executor_ = Cronet_Executor_CreateWith(TestUrlRequestCallback::Execute);
     Cronet_Executor_SetClientContext(executor_, this);
+  }
   return executor_;
 }
 
@@ -186,6 +192,8 @@ void TestUrlRequestCallback::OnFailed(Cronet_UrlRequestPtr request,
   if (info)
     response_info_ = std::make_unique<UrlResponseInfo>(info);
   last_error_ = error;
+  last_error_code_ = Cronet_Error_error_code_get(error);
+  last_error_message_ = Cronet_Error_message_get(error);
   SignalDone();
   MaybeCancelOrPause(request);
 }

components/cronet/native/test/test_url_request_callback.h

@@ -72,6 +72,10 @@ class TestUrlRequestCallback {
   std::unique_ptr<UrlResponseInfo> response_info_;
   // Owned by UrlRequest, only valid until UrlRequest is destroyed.
   Cronet_ErrorPtr last_error_ = nullptr;
+  // Values copied from |last_error_| valid after UrlRequest is destroyed.
+  Cronet_Error_ERROR_CODE last_error_code_ =
+      Cronet_Error_ERROR_CODE_ERROR_OTHER;
+  std::string last_error_message_;
 
   ResponseStep response_step_ = NOTHING;
 

components/cronet/native/test/url_request_test.cc

@@ -11,6 +11,7 @@
 #include "base/message_loop/message_loop.h"
 #include "base/run_loop.h"
 #include "base/synchronization/waitable_event.h"
+#include "components/cronet/native/test/test_upload_data_provider.h"
 #include "components/cronet/native/test/test_url_request_callback.h"
 #include "components/cronet/native/test/test_util.h"
 #include "components/cronet/test/test_server.h"
@@ -19,6 +20,7 @@
 #include "testing/gtest/include/gtest/gtest.h"
 #include "url/gurl.h"
 
+using cronet::test::TestUploadDataProvider;
 using cronet::test::TestUrlRequestCallback;
 
 namespace {
@@ -34,11 +36,30 @@ class UrlRequestTest : public ::testing::Test {
 
   std::unique_ptr<TestUrlRequestCallback> StartAndWaitForComplete(
       const std::string& url,
-      std::unique_ptr<TestUrlRequestCallback> test_callback) {
+      std::unique_ptr<TestUrlRequestCallback> test_callback,
+      const std::string& http_method,
+      TestUploadDataProvider* test_upload_data_provider) {
     Cronet_EnginePtr engine = cronet::test::CreateTestEngine(0);
     Cronet_UrlRequestPtr request = Cronet_UrlRequest_Create();
     Cronet_UrlRequestParamsPtr request_params =
         Cronet_UrlRequestParams_Create();
+    Cronet_UrlRequestParams_http_method_set(request_params,
+                                            http_method.c_str());
+    Cronet_UploadDataProviderPtr upload_data_provider = nullptr;
+
+    // Add upload data provider and set content type required for upload.
+    if (test_upload_data_provider != nullptr) {
+      upload_data_provider =
+          test_upload_data_provider->CreateUploadDataProvider();
+      Cronet_UrlRequestParams_upload_data_provider_set(request_params,
+                                                       upload_data_provider);
+      Cronet_HttpHeaderPtr header = Cronet_HttpHeader_Create();
+      Cronet_HttpHeader_name_set(header, "Content-Type");
+      Cronet_HttpHeader_value_set(header, "Useless/string");
+      Cronet_UrlRequestParams_request_headers_add(request_params, header);
+      Cronet_HttpHeader_Destroy(header);
+    }
+
     // Executor provided by the application is owned by |test_callback|.
     Cronet_ExecutorPtr executor = test_callback->GetExecutor(false);
     // Callback provided by the application.
@@ -55,7 +76,8 @@ class UrlRequestTest : public ::testing::Test {
     test_callback->ShutdownExecutor();
     EXPECT_TRUE(test_callback->IsDone());
     EXPECT_TRUE(Cronet_UrlRequest_IsDone(request));
-
+    if (upload_data_provider != nullptr)
+      Cronet_UploadDataProvider_Destroy(upload_data_provider);
     Cronet_UrlRequestParams_Destroy(request_params);
     Cronet_UrlRequest_Destroy(request);
     Cronet_UrlRequestCallback_Destroy(callback);
@@ -63,6 +85,14 @@ class UrlRequestTest : public ::testing::Test {
     return test_callback;
   }
 
+  std::unique_ptr<TestUrlRequestCallback> StartAndWaitForComplete(
+      const std::string& url,
+      std::unique_ptr<TestUrlRequestCallback> test_callback) {
+    return StartAndWaitForComplete(url, std::move(test_callback),
+                                   /* http_method =  */ std::string(),
+                                   /* upload_data_provider =  */ nullptr);
+  }
+
   std::unique_ptr<TestUrlRequestCallback> StartAndWaitForComplete(
       const std::string& url) {
     return StartAndWaitForComplete(url,
@@ -249,6 +279,345 @@ TEST_F(UrlRequestTest, SimpleGet) {
   ExpectResponseInfoEquals(expected_response_info, *callback->response_info_);
 }
 
+TEST_F(UrlRequestTest, UploadEmptyBodySync) {
+  const std::string url = cronet::TestServer::GetEchoRequestBodyURL();
+  TestUploadDataProvider data_provider(TestUploadDataProvider::SYNC,
+                                       /* executor = */ nullptr);
+  auto callback = std::make_unique<TestUrlRequestCallback>();
+  callback = StartAndWaitForComplete(url, std::move(callback), std::string(),
+                                     &data_provider);
+  data_provider.AssertClosed();
+  EXPECT_EQ(0, data_provider.GetUploadedLength());
+  EXPECT_EQ(0, data_provider.num_read_calls());
+  EXPECT_EQ(0, data_provider.num_rewind_calls());
+  EXPECT_EQ(200, callback->response_info_->http_status_code);
+  EXPECT_EQ("", callback->response_as_string_);
+}
+
+TEST_F(UrlRequestTest, UploadSync) {
+  const std::string url = cronet::TestServer::GetEchoRequestBodyURL();
+  TestUploadDataProvider data_provider(TestUploadDataProvider::SYNC,
+                                       /* executor = */ nullptr);
+  data_provider.AddRead("Test");
+  auto callback = std::make_unique<TestUrlRequestCallback>();
+  callback = StartAndWaitForComplete(url, std::move(callback), std::string(),
+                                     &data_provider);
+  data_provider.AssertClosed();
+  EXPECT_EQ(4, data_provider.GetUploadedLength());
+  EXPECT_EQ(1, data_provider.num_read_calls());
+  EXPECT_EQ(0, data_provider.num_rewind_calls());
+  EXPECT_EQ(200, callback->response_info_->http_status_code);
+  EXPECT_EQ("Test", callback->response_as_string_);
+}
+
+TEST_F(UrlRequestTest, UploadMultiplePiecesSync) {
+  const std::string url = cronet::TestServer::GetEchoRequestBodyURL();
+  auto callback = std::make_unique<TestUrlRequestCallback>();
+  TestUploadDataProvider data_provider(TestUploadDataProvider::SYNC,
+                                       callback->GetExecutor(false));
+  data_provider.AddRead("Y");
+  data_provider.AddRead("et ");
+  data_provider.AddRead("another ");
+  data_provider.AddRead("test");
+  callback = StartAndWaitForComplete(url, std::move(callback), std::string(),
+                                     &data_provider);
+  data_provider.AssertClosed();
+  EXPECT_EQ(16, data_provider.GetUploadedLength());
+  EXPECT_EQ(4, data_provider.num_read_calls());
+  EXPECT_EQ(0, data_provider.num_rewind_calls());
+  EXPECT_EQ(200, callback->response_info_->http_status_code);
+  EXPECT_EQ("Yet another test", callback->response_as_string_);
+}
+
+TEST_F(UrlRequestTest, UploadMultiplePiecesAsync) {
+  const std::string url = cronet::TestServer::GetEchoRequestBodyURL();
+  auto callback = std::make_unique<TestUrlRequestCallback>();
+  TestUploadDataProvider data_provider(TestUploadDataProvider::ASYNC,
+                                       callback->GetExecutor(false));
+  data_provider.AddRead("Y");
+  data_provider.AddRead("et ");
+  data_provider.AddRead("another ");
+  data_provider.AddRead("test");
+  callback = StartAndWaitForComplete(url, std::move(callback), std::string(),
+                                     &data_provider);
+  data_provider.AssertClosed();
+  EXPECT_EQ(16, data_provider.GetUploadedLength());
+  EXPECT_EQ(4, data_provider.num_read_calls());
+  EXPECT_EQ(0, data_provider.num_rewind_calls());
+  EXPECT_EQ(200, callback->response_info_->http_status_code);
+  EXPECT_EQ("Yet another test", callback->response_as_string_);
+}
+
+TEST_F(UrlRequestTest, UploadChangesDefaultMethod) {
+  const std::string url = cronet::TestServer::GetEchoMethodURL();
+  TestUploadDataProvider upload_data_provider(TestUploadDataProvider::SYNC,
+                                              /* executor = */ nullptr);
+  upload_data_provider.AddRead("Test");
+  auto callback = std::make_unique<TestUrlRequestCallback>();
+
+  callback = StartAndWaitForComplete(url, std::move(callback), std::string(),
+                                     &upload_data_provider);
+  EXPECT_EQ(200, callback->response_info_->http_status_code);
+  // Setting upload provider should change method to 'POST'.
+  EXPECT_EQ("POST", callback->response_as_string_);
+}
+
+TEST_F(UrlRequestTest, UploadWithSetMethod) {
+  const std::string url = cronet::TestServer::GetEchoMethodURL();
+  TestUploadDataProvider upload_data_provider(TestUploadDataProvider::SYNC,
+                                              /* executor = */ nullptr);
+  upload_data_provider.AddRead("Test");
+  auto callback = std::make_unique<TestUrlRequestCallback>();
+
+  callback = StartAndWaitForComplete(url, std::move(callback),
+                                     std::string("PUT"), &upload_data_provider);
+  EXPECT_EQ(200, callback->response_info_->http_status_code);
+  // Setting upload provider should change method to 'POST'.
+  EXPECT_EQ("PUT", callback->response_as_string_);
+}
+
+// TODO(mef): Direct Executor is causing lock re-entrancy when callback invoked
+// on the network thread calls Cronet_UrlRequest_IsDone() on the same thread.
+TEST_F(UrlRequestTest, DISABLED_UploadWithDirectExecutor) {
+  const std::string url = cronet::TestServer::GetEchoRequestBodyURL();
+  auto callback = std::make_unique<TestUrlRequestCallback>();
+  TestUploadDataProvider data_provider(TestUploadDataProvider::SYNC,
+                                       callback->GetExecutor(true));
+  data_provider.AddRead("Test");
+  callback = StartAndWaitForComplete(url, std::move(callback), std::string(),
+                                     &data_provider);
+  data_provider.AssertClosed();
+  EXPECT_EQ(4, data_provider.GetUploadedLength());
+  EXPECT_EQ(1, data_provider.num_read_calls());
+  EXPECT_EQ(0, data_provider.num_rewind_calls());
+  EXPECT_EQ(200, callback->response_info_->http_status_code);
+  EXPECT_EQ("Test", callback->response_as_string_);
+}
+
+TEST_F(UrlRequestTest, UploadRedirectSync) {
+  const std::string url = cronet::TestServer::GetRedirectToEchoBodyURL();
+  TestUploadDataProvider data_provider(TestUploadDataProvider::SYNC,
+                                       /* executor = */ nullptr);
+  data_provider.AddRead("Test");
+  auto callback = std::make_unique<TestUrlRequestCallback>();
+  callback = StartAndWaitForComplete(url, std::move(callback), std::string(),
+                                     &data_provider);
+  data_provider.AssertClosed();
+  EXPECT_EQ(4, data_provider.GetUploadedLength());
+  EXPECT_EQ(2, data_provider.num_read_calls());
+  EXPECT_EQ(1, data_provider.num_rewind_calls());
+  EXPECT_EQ(200, callback->response_info_->http_status_code);
+  EXPECT_EQ("Test", callback->response_as_string_);
+}
+
+TEST_F(UrlRequestTest, UploadRedirectAsync) {
+  auto callback = std::make_unique<TestUrlRequestCallback>();
+  const std::string url = cronet::TestServer::GetRedirectToEchoBodyURL();
+  TestUploadDataProvider data_provider(TestUploadDataProvider::ASYNC,
+                                       callback->GetExecutor(false));
+  data_provider.AddRead("Test");
+  callback = StartAndWaitForComplete(url, std::move(callback), std::string(),
+                                     &data_provider);
+  data_provider.AssertClosed();
+  EXPECT_EQ(4, data_provider.GetUploadedLength());
+  EXPECT_EQ(2, data_provider.num_read_calls());
+  EXPECT_EQ(1, data_provider.num_rewind_calls());
+  EXPECT_EQ(200, callback->response_info_->http_status_code);
+  EXPECT_EQ("Test", callback->response_as_string_);
+}
+
+TEST_F(UrlRequestTest, UploadWithBadLength) {
+  auto callback = std::make_unique<TestUrlRequestCallback>();
+  const std::string url = cronet::TestServer::GetEchoRequestBodyURL();
+  TestUploadDataProvider data_provider(TestUploadDataProvider::SYNC,
+                                       callback->GetExecutor(false));
+  data_provider.set_bad_length(1ll);
+  data_provider.AddRead("12");
+  callback = StartAndWaitForComplete(url, std::move(callback), std::string(),
+                                     &data_provider);
+  data_provider.AssertClosed();
+  EXPECT_EQ(2, data_provider.GetUploadedLength());
+  EXPECT_EQ(1, data_provider.num_read_calls());
+  EXPECT_EQ(0, data_provider.num_rewind_calls());
+  EXPECT_EQ(nullptr, callback->response_info_);
+  EXPECT_NE(nullptr, callback->last_error_);
+  EXPECT_EQ(Cronet_Error_ERROR_CODE_ERROR_CALLBACK, callback->last_error_code_);
+  EXPECT_EQ(0ul, callback->last_error_message_.find(
+                     "Failure from UploadDataProvider"));
+  EXPECT_NE(std::string::npos,
+            callback->last_error_message_.find(
+                "Read upload data length 2 exceeds expected length 1"));
+}
+
+TEST_F(UrlRequestTest, UploadWithBadLengthBufferAligned) {
+  auto callback = std::make_unique<TestUrlRequestCallback>();
+  const std::string url = cronet::TestServer::GetEchoRequestBodyURL();
+  TestUploadDataProvider data_provider(TestUploadDataProvider::SYNC,
+                                       callback->GetExecutor(false));
+  data_provider.set_bad_length(8191ll);
+  // Add 8192 bytes to read.
+  for (int i = 0; i < 512; ++i)
+    data_provider.AddRead("0123456789abcdef");
+
+  callback = StartAndWaitForComplete(url, std::move(callback), std::string(),
+                                     &data_provider);
+  data_provider.AssertClosed();
+  EXPECT_EQ(8192, data_provider.GetUploadedLength());
+  EXPECT_EQ(512, data_provider.num_read_calls());
+  EXPECT_EQ(0, data_provider.num_rewind_calls());
+  EXPECT_EQ(nullptr, callback->response_info_);
+  EXPECT_NE(nullptr, callback->last_error_);
+  EXPECT_EQ(Cronet_Error_ERROR_CODE_ERROR_CALLBACK, callback->last_error_code_);
+  EXPECT_EQ(0ul, callback->last_error_message_.find(
+                     "Failure from UploadDataProvider"));
+  EXPECT_NE(std::string::npos,
+            callback->last_error_message_.find(
+                "Read upload data length 8192 exceeds expected length 8191"));
+}
+
+TEST_F(UrlRequestTest, UploadReadFailSync) {
+  auto callback = std::make_unique<TestUrlRequestCallback>();
+  const std::string url = cronet::TestServer::GetEchoRequestBodyURL();
+  TestUploadDataProvider data_provider(TestUploadDataProvider::SYNC,
+                                       callback->GetExecutor(false));
+  data_provider.SetReadFailure(0, TestUploadDataProvider::CALLBACK_SYNC);
+  data_provider.AddRead("Test");
+  callback = StartAndWaitForComplete(url, std::move(callback), std::string(),
+                                     &data_provider);
+  data_provider.AssertClosed();
+  EXPECT_EQ(4, data_provider.GetUploadedLength());
+  EXPECT_EQ(1, data_provider.num_read_calls());
+  EXPECT_EQ(0, data_provider.num_rewind_calls());
+  EXPECT_EQ(nullptr, callback->response_info_);
+  EXPECT_NE(nullptr, callback->last_error_);
+  EXPECT_EQ(Cronet_Error_ERROR_CODE_ERROR_CALLBACK, callback->last_error_code_);
+  EXPECT_EQ(0ul, callback->last_error_message_.find(
+                     "Failure from UploadDataProvider"));
+  EXPECT_NE(std::string::npos,
+            callback->last_error_message_.find("Sync read failure"));
+}
+
+TEST_F(UrlRequestTest, UploadReadFailAsync) {
+  auto callback = std::make_unique<TestUrlRequestCallback>();
+  const std::string url = cronet::TestServer::GetEchoRequestBodyURL();
+  TestUploadDataProvider data_provider(TestUploadDataProvider::SYNC,
+                                       callback->GetExecutor(false));
+  data_provider.SetReadFailure(0, TestUploadDataProvider::CALLBACK_ASYNC);
+  data_provider.AddRead("Test");
+  callback = StartAndWaitForComplete(url, std::move(callback), std::string(),
+                                     &data_provider);
+  data_provider.AssertClosed();
+  EXPECT_EQ(4, data_provider.GetUploadedLength());
+  EXPECT_EQ(1, data_provider.num_read_calls());
+  EXPECT_EQ(0, data_provider.num_rewind_calls());
+  EXPECT_EQ(nullptr, callback->response_info_);
+  EXPECT_NE(nullptr, callback->last_error_);
+  EXPECT_EQ(Cronet_Error_ERROR_CODE_ERROR_CALLBACK, callback->last_error_code_);
+  EXPECT_EQ(0ul, callback->last_error_message_.find(
+                     "Failure from UploadDataProvider"));
+  EXPECT_NE(std::string::npos,
+            callback->last_error_message_.find("Async read failure"));
+}
+
+TEST_F(UrlRequestTest, UploadRewindFailSync) {
+  auto callback = std::make_unique<TestUrlRequestCallback>();
+  const std::string url = cronet::TestServer::GetRedirectToEchoBodyURL();
+  TestUploadDataProvider data_provider(TestUploadDataProvider::SYNC,
+                                       callback->GetExecutor(false));
+  data_provider.SetRewindFailure(TestUploadDataProvider::CALLBACK_SYNC);
+  data_provider.AddRead("Test");
+  callback = StartAndWaitForComplete(url, std::move(callback), std::string(),
+                                     &data_provider);
+  data_provider.AssertClosed();
+  EXPECT_EQ(4, data_provider.GetUploadedLength());
+  EXPECT_EQ(1, data_provider.num_read_calls());
+  EXPECT_EQ(1, data_provider.num_rewind_calls());
+  EXPECT_NE(nullptr, callback->last_error_);
+  EXPECT_EQ(Cronet_Error_ERROR_CODE_ERROR_CALLBACK, callback->last_error_code_);
+  EXPECT_EQ(0ul, callback->last_error_message_.find(
+                     "Failure from UploadDataProvider"));
+  EXPECT_NE(std::string::npos,
+            callback->last_error_message_.find("Sync rewind failure"));
+}
+
+TEST_F(UrlRequestTest, UploadRewindFailAsync) {
+  auto callback = std::make_unique<TestUrlRequestCallback>();
+  const std::string url = cronet::TestServer::GetRedirectToEchoBodyURL();
+  TestUploadDataProvider data_provider(TestUploadDataProvider::SYNC,
+                                       callback->GetExecutor(false));
+  data_provider.SetRewindFailure(TestUploadDataProvider::CALLBACK_ASYNC);
+  data_provider.AddRead("Test");
+  callback = StartAndWaitForComplete(url, std::move(callback), std::string(),
+                                     &data_provider);
+  data_provider.AssertClosed();
+  EXPECT_EQ(4, data_provider.GetUploadedLength());
+  EXPECT_EQ(1, data_provider.num_read_calls());
+  EXPECT_EQ(1, data_provider.num_rewind_calls());
+  EXPECT_NE(nullptr, callback->last_error_);
+  EXPECT_EQ(Cronet_Error_ERROR_CODE_ERROR_CALLBACK, callback->last_error_code_);
+  EXPECT_EQ(0ul, callback->last_error_message_.find(
+                     "Failure from UploadDataProvider"));
+  EXPECT_NE(std::string::npos,
+            callback->last_error_message_.find("Async rewind failure"));
+}
+
+TEST_F(UrlRequestTest, UploadChunked) {
+  auto callback = std::make_unique<TestUrlRequestCallback>();
+  const std::string url = cronet::TestServer::GetEchoRequestBodyURL();
+  TestUploadDataProvider data_provider(TestUploadDataProvider::SYNC,
+                                       callback->GetExecutor(false));
+  data_provider.AddRead("Test Hello");
+  data_provider.set_chunked(true);
+  EXPECT_EQ(-1, data_provider.GetLength());
+  callback = StartAndWaitForComplete(url, std::move(callback), std::string(),
+                                     &data_provider);
+  data_provider.AssertClosed();
+  EXPECT_EQ(-1, data_provider.GetUploadedLength());
+  EXPECT_EQ(1, data_provider.num_read_calls());
+  EXPECT_EQ(200, callback->response_info_->http_status_code);
+  EXPECT_EQ("Test Hello", callback->response_as_string_);
+}
+
+TEST_F(UrlRequestTest, UploadChunkedLastReadZeroLengthBody) {
+  auto callback = std::make_unique<TestUrlRequestCallback>();
+  const std::string url = cronet::TestServer::GetEchoRequestBodyURL();
+  TestUploadDataProvider data_provider(TestUploadDataProvider::SYNC,
+                                       callback->GetExecutor(false));
+  // Add 3 reads. The last read has a 0-length body.
+  data_provider.AddRead("hello there");
+  data_provider.AddRead("!");
+  data_provider.AddRead("");
+  data_provider.set_chunked(true);
+  EXPECT_EQ(-1, data_provider.GetLength());
+  callback = StartAndWaitForComplete(url, std::move(callback), std::string(),
+                                     &data_provider);
+  data_provider.AssertClosed();
+  EXPECT_EQ(-1, data_provider.GetUploadedLength());
+  // 2 read call for the first two data chunks, and 1 for final chunk.
+  EXPECT_EQ(3, data_provider.num_read_calls());
+  EXPECT_EQ(200, callback->response_info_->http_status_code);
+  EXPECT_EQ("hello there!", callback->response_as_string_);
+}
+
+// Test where an upload fails without ever initializing the
+// UploadDataStream, because it can't connect to the server.
+TEST_F(UrlRequestTest, UploadFailsWithoutInitializingStream) {
+  // The port for PTP will always refuse a TCP connection
+  const std::string url = "http://127.0.0.1:319";
+  TestUploadDataProvider data_provider(TestUploadDataProvider::SYNC,
+                                       /* executor = */ nullptr);
+  data_provider.AddRead("Test");
+  auto callback = std::make_unique<TestUrlRequestCallback>();
+  callback = StartAndWaitForComplete(url, std::move(callback), std::string(),
+                                     &data_provider);
+  data_provider.AssertClosed();
+  EXPECT_EQ(0, data_provider.num_read_calls());
+  EXPECT_EQ(0, data_provider.num_rewind_calls());
+  EXPECT_EQ(nullptr, callback->response_info_);
+  EXPECT_EQ("", callback->response_as_string_);
+  EXPECT_TRUE(callback->on_error_called_);
+}
+
 TEST_F(UrlRequestTest, SimpleRequest) {
   Cronet_EnginePtr engine = cronet::test::CreateTestEngine(0);
   Cronet_UrlRequestPtr request = Cronet_UrlRequest_Create();
@@ -373,21 +742,21 @@ TEST_F(UrlRequestTest, FailedRequestHostNotFound) {
   test_callback.WaitForDone();
   EXPECT_TRUE(test_callback.IsDone());
   EXPECT_TRUE(test_callback.on_error_called_);
-  ASSERT_FALSE(test_callback.on_canceled_called_);
+  EXPECT_FALSE(test_callback.on_canceled_called_);
 
   EXPECT_TRUE(test_callback.response_as_string_.empty());
-  ASSERT_EQ(nullptr, test_callback.response_info_);
-  ASSERT_NE(nullptr, test_callback.last_error_);
+  EXPECT_EQ(nullptr, test_callback.response_info_);
+  EXPECT_NE(nullptr, test_callback.last_error_);
 
-  ASSERT_EQ(Cronet_Error_ERROR_CODE_ERROR_HOSTNAME_NOT_RESOLVED,
+  EXPECT_EQ(Cronet_Error_ERROR_CODE_ERROR_HOSTNAME_NOT_RESOLVED,
             Cronet_Error_error_code_get(test_callback.last_error_));
-  ASSERT_FALSE(
+  EXPECT_FALSE(
       Cronet_Error_immediately_retryable_get(test_callback.last_error_));
-  ASSERT_STREQ("net::ERR_NAME_NOT_RESOLVED",
+  EXPECT_STREQ("net::ERR_NAME_NOT_RESOLVED",
                Cronet_Error_message_get(test_callback.last_error_));
-  ASSERT_EQ(-105,
+  EXPECT_EQ(-105,
             Cronet_Error_internal_error_code_get(test_callback.last_error_));
-  ASSERT_EQ(
+  EXPECT_EQ(
       0, Cronet_Error_quic_detailed_error_code_get(test_callback.last_error_));
 
   Cronet_UrlRequestParams_Destroy(request_params);

components/cronet/native/upload_data_sink.cc

+// Copyright 2018 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/native/upload_data_sink.h"
+
+#include <inttypes.h>
+#include <utility>
+
+#include "base/bind.h"
+#include "base/logging.h"
+#include "base/macros.h"
+#include "base/strings/strcat.h"
+#include "base/strings/stringprintf.h"
+#include "components/cronet/cronet_upload_data_stream.h"
+#include "components/cronet/native/engine.h"
+#include "components/cronet/native/generated/cronet.idl_impl_struct.h"
+#include "components/cronet/native/include/cronet_c.h"
+#include "components/cronet/native/io_buffer_with_cronet_buffer.h"
+#include "components/cronet/native/runnables.h"
+#include "components/cronet/native/url_request.h"
+#include "net/base/io_buffer.h"
+
+namespace cronet {
+
+// This class is called by Cronet's network stack as an implementation of
+// CronetUploadDataStream::Delegate, and forwards the calls along to
+// Cronet_UploadDataSinkImpl on the embedder's executor.
+// This class is always called on the network thread and is destroyed in
+// OnUploadDataStreamDestroyed() callback.
+class Cronet_UploadDataSinkImpl::NetworkTasks
+    : public CronetUploadDataStream::Delegate {
+ public:
+  NetworkTasks(Cronet_UploadDataSinkImpl* upload_data_sink,
+               Cronet_Executor* upload_data_provider_executor);
+  ~NetworkTasks() override;
+
+ private:
+  // CronetUploadDataStream::Delegate implementation:
+  void InitializeOnNetworkThread(
+      base::WeakPtr<CronetUploadDataStream> upload_data_stream) override;
+  void Read(net::IOBuffer* buffer, int buf_len) override;
+  void Rewind() override;
+  void OnUploadDataStreamDestroyed() override;
+
+  // Post |task| to client executor.
+  void PostTaskToExecutor(base::OnceClosure task);
+
+  // The upload data sink that is owned by url request and always accessed on
+  // the client thread. It always outlives |this| callback.
+  Cronet_UploadDataSinkImpl* const upload_data_sink_ = nullptr;
+
+  // Executor for provider callback, used, but not owned, by |this|. Always
+  // outlives |this| callback.
+  Cronet_ExecutorPtr const upload_data_provider_executor_ = nullptr;
+
+  THREAD_CHECKER(network_thread_checker_);
+  DISALLOW_COPY_AND_ASSIGN(NetworkTasks);
+};
+
+Cronet_UploadDataSinkImpl::NetworkTasks::NetworkTasks(
+    Cronet_UploadDataSinkImpl* upload_data_sink,
+    Cronet_Executor* upload_data_provider_executor)
+    : upload_data_sink_(upload_data_sink),
+      upload_data_provider_executor_(upload_data_provider_executor) {
+  DETACH_FROM_THREAD(network_thread_checker_);
+}
+
+Cronet_UploadDataSinkImpl::NetworkTasks::~NetworkTasks() = default;
+
+Cronet_UploadDataSinkImpl::Cronet_UploadDataSinkImpl(
+    Cronet_UrlRequestImpl* url_request,
+    Cronet_UploadDataProvider* upload_data_provider,
+    Cronet_Executor* upload_data_provider_executor)
+    : url_request_(url_request),
+      upload_data_provider_executor_(upload_data_provider_executor),
+      upload_data_provider_(upload_data_provider) {}
+
+Cronet_UploadDataSinkImpl::~Cronet_UploadDataSinkImpl() = default;
+
+bool Cronet_UploadDataSinkImpl::InitRequest(CronetURLRequest* request) {
+  int64_t length = upload_data_provider_->GetLength();
+  if (length == -1) {
+    is_chunked_ = true;
+  } else {
+    CHECK_GE(length, 0);
+    length_ = static_cast<uint64_t>(length);
+    remaining_length_ = length_;
+  }
+
+  request->SetUpload(std::make_unique<CronetUploadDataStream>(
+      new NetworkTasks(this, upload_data_provider_executor_), length));
+  return true;
+}
+
+void Cronet_UploadDataSinkImpl::OnReadSucceeded(uint64_t bytes_read,
+                                                bool final_chunk) {
+  {
+    base::AutoLock lock(lock_);
+    CheckState(READ);
+    in_which_user_callback_ = NOT_IN_CALLBACK;
+    if (!upload_data_provider_)
+      return;
+  }
+  if (url_request_->IsDone())
+    return;
+  if (close_when_not_in_callback_) {
+    PostCloseToExecutor();
+    return;
+  }
+  CHECK(bytes_read > 0 || (final_chunk && bytes_read == 0));
+  // Bytes read exceeds buffer length.
+  CHECK_LT(static_cast<size_t>(bytes_read), buffer_->io_buffer_len());
+  if (!is_chunked_) {
+    // Only chunked upload can have the final chunk.
+    CHECK(!final_chunk);
+    // Read upload data length exceeds specified length.
+    if (bytes_read > remaining_length_) {
+      PostCloseToExecutor();
+      std::string error_message =
+          base::StringPrintf("Read upload data length %" PRIu64
+                             " exceeds expected length %" PRIu64,
+                             length_ - remaining_length_ + bytes_read, length_);
+      url_request_->OnUploadDataProviderError(error_message.c_str());
+      return;
+    }
+    remaining_length_ -= bytes_read;
+  }
+  network_task_runner_->PostTask(
+      FROM_HERE, base::BindOnce(&CronetUploadDataStream::OnReadSuccess,
+                                upload_data_stream_, bytes_read, final_chunk));
+}
+
+void Cronet_UploadDataSinkImpl::OnReadError(Cronet_String error_message) {
+  {
+    base::AutoLock lock(lock_);
+    CheckState(READ);
+    in_which_user_callback_ = NOT_IN_CALLBACK;
+    if (!upload_data_provider_)
+      return;
+  }
+  if (url_request_->IsDone())
+    return;
+  PostCloseToExecutor();
+  url_request_->OnUploadDataProviderError(error_message);
+}
+
+void Cronet_UploadDataSinkImpl::OnRewindSucceeded() {
+  {
+    base::AutoLock lock(lock_);
+    CheckState(REWIND);
+    in_which_user_callback_ = NOT_IN_CALLBACK;
+    if (!upload_data_provider_)
+      return;
+  }
+  remaining_length_ = length_;
+  if (url_request_->IsDone())
+    return;
+  if (close_when_not_in_callback_) {
+    PostCloseToExecutor();
+    return;
+  }
+  network_task_runner_->PostTask(
+      FROM_HERE, base::BindOnce(&CronetUploadDataStream::OnRewindSuccess,
+                                upload_data_stream_));
+}
+
+void Cronet_UploadDataSinkImpl::OnRewindError(Cronet_String error_message) {
+  {
+    base::AutoLock lock(lock_);
+    CheckState(REWIND);
+    in_which_user_callback_ = NOT_IN_CALLBACK;
+    if (!upload_data_provider_)
+      return;
+  }
+  if (url_request_->IsDone())
+    return;
+  PostCloseToExecutor();
+  url_request_->OnUploadDataProviderError(error_message);
+}
+
+void Cronet_UploadDataSinkImpl::InitializeUploadDataStream(
+    base::WeakPtr<CronetUploadDataStream> upload_data_stream,
+    scoped_refptr<base::SingleThreadTaskRunner> network_task_runner) {
+  DCHECK(!upload_data_stream_);
+  DCHECK(!network_task_runner_.get());
+  upload_data_stream_ = upload_data_stream;
+  network_task_runner_ = network_task_runner;
+}
+
+void Cronet_UploadDataSinkImpl::PostCloseToExecutor() {
+  Cronet_RunnablePtr runnable = new cronet::OnceClosureRunnable(base::BindOnce(
+      &Cronet_UploadDataSinkImpl::Close, base::Unretained(this)));
+  // |runnable| is passed to executor, which destroys it after execution.
+  Cronet_Executor_Execute(upload_data_provider_executor_, runnable);
+}
+
+void Cronet_UploadDataSinkImpl::Read(net::IOBuffer* buffer, int buf_len) {
+  if (url_request_->IsDone())
+    return;
+  Cronet_UploadDataProviderPtr upload_data_provider = nullptr;
+  {
+    base::AutoLock lock(lock_);
+    if (!upload_data_provider_)
+      return;
+    CheckState(NOT_IN_CALLBACK);
+    in_which_user_callback_ = READ;
+    upload_data_provider = upload_data_provider_;
+  }
+  buffer_ = std::make_unique<Cronet_BufferWithIOBuffer>(buffer, buf_len);
+  Cronet_UploadDataProvider_Read(upload_data_provider, this,
+                                 buffer_->cronet_buffer());
+}
+
+void Cronet_UploadDataSinkImpl::Rewind() {
+  if (url_request_->IsDone())
+    return;
+  Cronet_UploadDataProviderPtr upload_data_provider = nullptr;
+  {
+    base::AutoLock lock(lock_);
+    if (!upload_data_provider_)
+      return;
+    CheckState(NOT_IN_CALLBACK);
+    in_which_user_callback_ = REWIND;
+    upload_data_provider = upload_data_provider_;
+  }
+  Cronet_UploadDataProvider_Rewind(upload_data_provider, this);
+}
+
+void Cronet_UploadDataSinkImpl::Close() {
+  Cronet_UploadDataProviderPtr upload_data_provider = nullptr;
+  {
+    base::AutoLock lock(lock_);
+    // If |upload_data_provider_| is already closed from OnResponseStarted(),
+    // don't close it again from OnError() or OnCanceled().
+    if (!upload_data_provider_)
+      return;
+    if (in_which_user_callback_ != NOT_IN_CALLBACK) {
+      // If currently in the callback, then wait until return from callback
+      // before closing.
+      close_when_not_in_callback_ = true;
+      return;
+    }
+    upload_data_provider = upload_data_provider_;
+    upload_data_provider_ = nullptr;
+  }
+  Cronet_UploadDataProvider_Close(upload_data_provider);
+}
+
+void Cronet_UploadDataSinkImpl::CheckState(UserCallback expected_state) {
+  lock_.AssertAcquired();
+  CHECK(in_which_user_callback_ == expected_state);
+}
+
+void Cronet_UploadDataSinkImpl::NetworkTasks::InitializeOnNetworkThread(
+    base::WeakPtr<CronetUploadDataStream> upload_data_stream) {
+  DCHECK_CALLED_ON_VALID_THREAD(network_thread_checker_);
+  PostTaskToExecutor(
+      base::BindOnce(&Cronet_UploadDataSinkImpl::InitializeUploadDataStream,
+                     base::Unretained(upload_data_sink_), upload_data_stream,
+                     base::ThreadTaskRunnerHandle::Get()));
+}
+
+void Cronet_UploadDataSinkImpl::NetworkTasks::Read(net::IOBuffer* buffer,
+                                                   int buf_len) {
+  DCHECK_CALLED_ON_VALID_THREAD(network_thread_checker_);
+  PostTaskToExecutor(base::BindOnce(&Cronet_UploadDataSinkImpl::Read,
+                                    base::Unretained(upload_data_sink_),
+                                    base::Passed(&buffer), buf_len));
+}
+
+void Cronet_UploadDataSinkImpl::NetworkTasks::Rewind() {
+  DCHECK_CALLED_ON_VALID_THREAD(network_thread_checker_);
+  PostTaskToExecutor(base::BindOnce(&Cronet_UploadDataSinkImpl::Rewind,
+                                    base::Unretained(upload_data_sink_)));
+}
+
+void Cronet_UploadDataSinkImpl::NetworkTasks::OnUploadDataStreamDestroyed() {
+  DCHECK_CALLED_ON_VALID_THREAD(network_thread_checker_);
+  delete this;
+}
+
+void Cronet_UploadDataSinkImpl::NetworkTasks::PostTaskToExecutor(
+    base::OnceClosure task) {
+  Cronet_RunnablePtr runnable =
+      new cronet::OnceClosureRunnable(std::move(task));
+  // |runnable| is passed to executor, which destroys it after execution.
+  Cronet_Executor_Execute(upload_data_provider_executor_, runnable);
+}
+
+};  // namespace cronet

components/cronet/native/upload_data_sink.h

+// Copyright 2018 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 COMPONENTS_CRONET_NATIVE_UPLOAD_DATA_SINK_H_
+#define COMPONENTS_CRONET_NATIVE_UPLOAD_DATA_SINK_H_
+
+#include <memory>
+#include <string>
+
+#include "base/macros.h"
+#include "base/synchronization/lock.h"
+#include "base/synchronization/waitable_event.h"
+#include "components/cronet/cronet_upload_data_stream.h"
+#include "components/cronet/cronet_url_request.h"
+#include "components/cronet/cronet_url_request_context.h"
+#include "components/cronet/native/generated/cronet.idl_impl_interface.h"
+
+namespace cronet {
+
+class Cronet_UrlRequestImpl;
+class Cronet_BufferWithIOBuffer;
+
+// Implementation of Cronet_UploadDataSink that uses CronetUploadDataStream.
+// Always accessed on client executor.
+class Cronet_UploadDataSinkImpl : public Cronet_UploadDataSink {
+ public:
+  Cronet_UploadDataSinkImpl(Cronet_UrlRequestImpl* url_request,
+                            Cronet_UploadDataProvider* upload_data_provider,
+                            Cronet_Executor* upload_data_provider_executor);
+  ~Cronet_UploadDataSinkImpl() override;
+
+  // Initialize length and attach upload to request. Called on client thread.
+  bool InitRequest(CronetURLRequest* request);
+
+  // Mark stream as closed and post |Close()| callback to consumer.
+  void PostCloseToExecutor();
+
+ private:
+  class NetworkTasks;
+  enum UserCallback { READ, REWIND, GET_LENGTH, NOT_IN_CALLBACK };
+
+  // Cronet_UploadDataSink
+  void OnReadSucceeded(uint64_t bytes_read, bool final_chunk) override;
+  void OnReadError(Cronet_String error_message) override;
+  void OnRewindSucceeded() override;
+  void OnRewindError(Cronet_String error_message) override;
+
+  // CronetUploadDataStream::Delegate methods posted from the network thread.
+  void InitializeUploadDataStream(
+      base::WeakPtr<CronetUploadDataStream> upload_data_stream,
+      scoped_refptr<base::SingleThreadTaskRunner> network_task_runner);
+  void Read(net::IOBuffer* buffer, int buf_len);
+  void Rewind();
+  void Close();
+
+  void CheckState(UserCallback expected_state);
+
+  // Cronet objects not owned by |this| and accessed on client thread.
+
+  // The request, which owns |this|.
+  Cronet_UrlRequestImpl* const url_request_ = nullptr;
+  // Executor for provider callback, used, but not owned, by |this|. Always
+  // outlives |this| callback.
+  Cronet_ExecutorPtr const upload_data_provider_executor_ = nullptr;
+
+  // These are initialized in InitializeUploadDataStream(), so are safe to
+  // access during client callbacks, which all happen after initialization.
+  scoped_refptr<base::SingleThreadTaskRunner> network_task_runner_;
+  base::WeakPtr<CronetUploadDataStream> upload_data_stream_;
+
+  bool is_chunked_ = false;
+  uint64_t length_ = 0;
+  uint64_t remaining_length_ = 0;
+
+  // Synchronize access to |buffer_| and other objects below from different
+  // threads.
+  base::Lock lock_;
+  // Data provider callback interface, used, but not owned, by |this|.
+  // Set to nullptr when data provider is closed.
+  Cronet_UploadDataProviderPtr upload_data_provider_ = nullptr;
+
+  UserCallback in_which_user_callback_ = NOT_IN_CALLBACK;
+  // Close data provider once it returns from the callback.
+  bool close_when_not_in_callback_ = false;
+  // Keeps the net::IOBuffer and Cronet ByteBuffer alive until the next Read().
+  std::unique_ptr<Cronet_BufferWithIOBuffer> buffer_;
+
+  DISALLOW_COPY_AND_ASSIGN(Cronet_UploadDataSinkImpl);
+};
+
+}  // namespace cronet
+
+#endif  // COMPONENTS_CRONET_NATIVE_UPLOAD_DATA_SINK_H_

components/cronet/native/url_request.cc

@@ -5,14 +5,17 @@
 #include "components/cronet/native/url_request.h"
 
 #include <utility>
+#include <vector>
 
 #include "base/logging.h"
 #include "base/macros.h"
+#include "components/cronet/cronet_upload_data_stream.h"
 #include "components/cronet/native/engine.h"
 #include "components/cronet/native/generated/cronet.idl_impl_struct.h"
 #include "components/cronet/native/include/cronet_c.h"
 #include "components/cronet/native/io_buffer_with_cronet_buffer.h"
 #include "components/cronet/native/runnables.h"
+#include "components/cronet/native/upload_data_sink.h"
 #include "net/base/io_buffer.h"
 
 namespace {
@@ -130,16 +133,15 @@ std::unique_ptr<Cronet_Error> CreateCronet_Error(
 
 namespace cronet {
 
-// Callback is owned by CronetURLRequest. It is constructed on client thread,
-// but invoked and deleted on the network thread.
-class Cronet_UrlRequestImpl::Callback : public CronetURLRequest::Callback {
+// NetworkTasks is owned by CronetURLRequest. It is constructed on client
+// thread, but invoked and deleted on the network thread.
+class Cronet_UrlRequestImpl::NetworkTasks : public CronetURLRequest::Callback {
  public:
-  Callback(const std::string& url,
-           Cronet_UrlRequestImpl* url_request,
-           Cronet_UrlRequestCallbackPtr callback,
-           Cronet_ExecutorPtr executor);
-  ~Callback() override = default;
-  // CronetURLRequest::Callback implementations:
+  NetworkTasks(const std::string& url, Cronet_UrlRequestImpl* url_request);
+  ~NetworkTasks() override = default;
+
+ private:
+  // CronetURLRequest::Callback implementation:
   void OnReceivedRedirect(const std::string& new_location,
                           int http_status_code,
                           const std::string& http_status_text,
@@ -183,16 +185,8 @@ class Cronet_UrlRequestImpl::Callback : public CronetURLRequest::Callback {
                           int64_t sent_bytes_count,
                           int64_t received_bytes_count) override;
 
- private:
-  void PostTaskToExecutor(base::OnceClosure task);
-
   // The UrlRequest which owns context that owns the callback.
-  Cronet_UrlRequestImpl* url_request_ = nullptr;
-
-  // Application callback interface, used, but not owned, by |this|.
-  Cronet_UrlRequestCallbackPtr callback_ = nullptr;
-  // Executor for application callback, used, but not owned, by |this|.
-  Cronet_ExecutorPtr executor_ = nullptr;
+  Cronet_UrlRequestImpl* const url_request_ = nullptr;
 
   // URL chain contains the URL currently being requested, and
   // all URLs previously requested. New URLs are added before
@@ -201,16 +195,21 @@ class Cronet_UrlRequestImpl::Callback : public CronetURLRequest::Callback {
 
   // All methods except constructor are invoked on the network thread.
   THREAD_CHECKER(network_thread_checker_);
-  DISALLOW_COPY_AND_ASSIGN(Callback);
+  DISALLOW_COPY_AND_ASSIGN(NetworkTasks);
 };
 
 Cronet_UrlRequestImpl::Cronet_UrlRequestImpl() = default;
 
 Cronet_UrlRequestImpl::~Cronet_UrlRequestImpl() {
   base::AutoLock lock(lock_);
-  // Request may already be destroyed if it hasn't started or got canceled.
-  if (request_)
-    request_->Destroy(false);
+  // Only request that has never started is allowed to exist at this point.
+  // The app must wait for OnSucceeded / OnFailed / OnCanceled  callback before
+  // destroying |this|.
+  if (request_) {
+    CHECK(!started_);
+    DestroyRequestUnlessDoneLocked(
+        Cronet_RequestFinishedInfo_FINISHED_REASON_SUCCEEDED);
+  }
 }
 
 Cronet_RESULT Cronet_UrlRequestImpl::InitWithParams(
@@ -238,9 +237,12 @@ Cronet_RESULT Cronet_UrlRequestImpl::InitWithParams(
         Cronet_RESULT_ILLEGAL_STATE_REQUEST_ALREADY_INITIALIZED);
   }
 
+  callback_ = callback;
+  executor_ = executor;
+
   request_ = new CronetURLRequest(
       engine_->cronet_url_request_context(),
-      std::make_unique<Callback>(url, this, callback, executor), GURL(url),
+      std::make_unique<NetworkTasks>(url, this), GURL(url),
       ConvertRequestPriority(params->priority), params->disable_cache,
       true /* params->disableConnectionMigration */,
       false /* params->enableMetrics */,
@@ -248,6 +250,17 @@ Cronet_RESULT Cronet_UrlRequestImpl::InitWithParams(
       false /* traffic_stats_tag_set */, 0 /* traffic_stats_tag */,
       false /* traffic_stats_uid_set */, 0 /* traffic_stats_uid */);
 
+  if (params->upload_data_provider) {
+    upload_data_sink_ = std::make_unique<Cronet_UploadDataSinkImpl>(
+        this, params->upload_data_provider,
+        params->upload_data_provider_executor
+            ? params->upload_data_provider_executor
+            : executor);
+    if (!upload_data_sink_->InitRequest(request_))
+      return engine_->CheckResult(Cronet_RESULT_NULL_POINTER_CALLBACK);
+    request_->SetHttpMethod("POST");
+  }
+
   if (!params->http_method.empty() &&
       !request_->SetHttpMethod(params->http_method)) {
     return engine_->CheckResult(
@@ -358,22 +371,84 @@ void Cronet_UrlRequestImpl::GetStatus(
   NOTIMPLEMENTED();
 }
 
-Cronet_UrlRequestImpl::Callback::Callback(const std::string& url,
-                                          Cronet_UrlRequestImpl* url_request,
-                                          Cronet_UrlRequestCallbackPtr callback,
-                                          Cronet_ExecutorPtr executor)
-    : url_request_(url_request),
-      callback_(callback),
-      executor_(executor),
-      url_chain_({url}) {
+void Cronet_UrlRequestImpl::OnUploadDataProviderError(
+    const std::string& error_message) {
+  base::AutoLock lock(lock_);
+  // If |error_| is not nullptr, that means that another network error is
+  // already reported.
+  if (error_)
+    return;
+  error_ = CreateCronet_Error(
+      0, 0, "Failure from UploadDataProvider: " + error_message);
+  error_->error_code = Cronet_Error_ERROR_CODE_ERROR_CALLBACK;
+  // Invoke Cronet_UrlRequestCallback_OnFailed on client executor.
+  PostTaskToExecutor(base::BindOnce(
+      &Cronet_UrlRequestImpl::InvokeCallbackOnFailed, base::Unretained(this)));
+}
+
+void Cronet_UrlRequestImpl::PostTaskToExecutor(base::OnceClosure task) {
+  Cronet_RunnablePtr runnable =
+      new cronet::OnceClosureRunnable(std::move(task));
+  // |runnable| is passed to executor, which destroys it after execution.
+  Cronet_Executor_Execute(executor_, runnable);
+}
+
+void Cronet_UrlRequestImpl::InvokeCallbackOnRedirectReceived() {
+  if (IsDone())
+    return;
+  Cronet_UrlRequestCallback_OnRedirectReceived(
+      callback_, this, response_info_.get(),
+      response_info_->url_chain.front().c_str());
+}
+
+void Cronet_UrlRequestImpl::InvokeCallbackOnResponseStarted() {
+  if (IsDone())
+    return;
+  Cronet_UrlRequestCallback_OnResponseStarted(callback_, this,
+                                              response_info_.get());
+}
+
+void Cronet_UrlRequestImpl::InvokeCallbackOnReadCompleted(
+    std::unique_ptr<Cronet_Buffer> cronet_buffer,
+    int bytes_read) {
+  if (IsDone())
+    return;
+  Cronet_UrlRequestCallback_OnReadCompleted(
+      callback_, this, response_info_.get(), cronet_buffer.release(),
+      bytes_read);
+}
+
+void Cronet_UrlRequestImpl::InvokeCallbackOnSucceeded() {
+  if (DestroyRequestUnlessDone(
+          Cronet_RequestFinishedInfo_FINISHED_REASON_SUCCEEDED)) {
+    return;
+  }
+  Cronet_UrlRequestCallback_OnSucceeded(callback_, this, response_info_.get());
+}
+
+void Cronet_UrlRequestImpl::InvokeCallbackOnFailed() {
+  if (DestroyRequestUnlessDone(
+          Cronet_RequestFinishedInfo_FINISHED_REASON_FAILED)) {
+    return;
+  }
+  Cronet_UrlRequestCallback_OnFailed(callback_, this, response_info_.get(),
+                                     error_.get());
+}
+
+void Cronet_UrlRequestImpl::InvokeCallbackOnCanceled() {
+  Cronet_UrlRequestCallback_OnCanceled(callback_, this, response_info_.get());
+}
+
+Cronet_UrlRequestImpl::NetworkTasks::NetworkTasks(
+    const std::string& url,
+    Cronet_UrlRequestImpl* url_request)
+    : url_request_(url_request), url_chain_({url}) {
   DETACH_FROM_THREAD(network_thread_checker_);
   DCHECK(url_request);
-  DCHECK(callback);
-  DCHECK(executor);
 }
 
-// CronetURLRequest::Callback implementations:
-void Cronet_UrlRequestImpl::Callback::OnReceivedRedirect(
+// CronetURLRequest::NetworkTasks implementations:
+void Cronet_UrlRequestImpl::NetworkTasks::OnReceivedRedirect(
     const std::string& new_location,
     int http_status_code,
     const std::string& http_status_text,
@@ -391,20 +466,13 @@ void Cronet_UrlRequestImpl::Callback::OnReceivedRedirect(
   // Have to do this after creating responseInfo.
   url_chain_.push_back(new_location);
 
-  // Invoke Cronet_UrlRequestCallback_OnRedrectReceived using OnceClosure.
-  PostTaskToExecutor(base::BindOnce(
-      [](Cronet_UrlRequestCallback* callback,
-         Cronet_UrlRequestImpl* url_request) {
-        if (url_request->IsDone())
-          return;
-        Cronet_UrlRequestCallback_OnRedirectReceived(
-            callback, url_request, url_request->response_info_.get(),
-            url_request->response_info_->url_chain.front().c_str());
-      },
-      callback_, url_request_));
+  // Invoke Cronet_UrlRequestCallback_OnRedirectReceived on client executor.
+  url_request_->PostTaskToExecutor(
+      base::BindOnce(&Cronet_UrlRequestImpl::InvokeCallbackOnRedirectReceived,
+                     base::Unretained(url_request_)));
 }
 
-void Cronet_UrlRequestImpl::Callback::OnResponseStarted(
+void Cronet_UrlRequestImpl::NetworkTasks::OnResponseStarted(
     int http_status_code,
     const std::string& http_status_text,
     const net::HttpResponseHeaders* headers,
@@ -418,20 +486,16 @@ void Cronet_UrlRequestImpl::Callback::OnResponseStarted(
   url_request_->response_info_ = CreateCronet_UrlResponseInfo(
       url_chain_, http_status_code, http_status_text, headers, was_cached,
       negotiated_protocol, proxy_server, received_byte_count);
+  if (url_request_->upload_data_sink_)
+    url_request_->upload_data_sink_->PostCloseToExecutor();
 
-  // Invoke Cronet_UrlRequestCallback_OnResponseStarted using OnceClosure.
-  PostTaskToExecutor(base::BindOnce(
-      [](Cronet_UrlRequestCallback* callback,
-         Cronet_UrlRequestImpl* url_request) {
-        if (url_request->IsDone())
-          return;
-        Cronet_UrlRequestCallback_OnResponseStarted(
-            callback, url_request, url_request->response_info_.get());
-      },
-      callback_, url_request_));
+  // Invoke Cronet_UrlRequestCallback_OnResponseStarted on client executor.
+  url_request_->PostTaskToExecutor(
+      base::BindOnce(&Cronet_UrlRequestImpl::InvokeCallbackOnResponseStarted,
+                     base::Unretained(url_request_)));
 }
 
-void Cronet_UrlRequestImpl::Callback::OnReadCompleted(
+void Cronet_UrlRequestImpl::NetworkTasks::OnReadCompleted(
     scoped_refptr<net::IOBuffer> buffer,
     int bytes_read,
     int64_t received_byte_count) {
@@ -443,41 +507,26 @@ void Cronet_UrlRequestImpl::Callback::OnReadCompleted(
   url_request_->waiting_on_read_ = true;
   url_request_->response_info_->received_byte_count = received_byte_count;
 
-  // Invoke Cronet_UrlRequestCallback_OnReadCompleted using OnceClosure.
-  PostTaskToExecutor(base::BindOnce(
-      [](Cronet_UrlRequestCallback* callback,
-         Cronet_UrlRequestImpl* url_request,
-         std::unique_ptr<Cronet_Buffer> cronet_buffer, int bytes_read) {
-        if (url_request->IsDone())
-          return;
-
-        Cronet_UrlRequestCallback_OnReadCompleted(
-            callback, url_request, url_request->response_info_.get(),
-            cronet_buffer.release(), bytes_read);
-      },
-      callback_, url_request_, std::move(cronet_buffer), bytes_read));
+  // Invoke Cronet_UrlRequestCallback_OnReadCompleted on client executor.
+  url_request_->PostTaskToExecutor(base::BindOnce(
+      &Cronet_UrlRequestImpl::InvokeCallbackOnReadCompleted,
+      base::Unretained(url_request_), std::move(cronet_buffer), bytes_read));
 }
 
-void Cronet_UrlRequestImpl::Callback::OnSucceeded(int64_t received_byte_count) {
+void Cronet_UrlRequestImpl::NetworkTasks::OnSucceeded(
+    int64_t received_byte_count) {
   DCHECK_CALLED_ON_VALID_THREAD(network_thread_checker_);
   base::AutoLock lock(url_request_->lock_);
   url_request_->response_info_->received_byte_count = received_byte_count;
 
-  // Invoke Cronet_UrlRequestCallback_OnSucceeded using OnceClosure.
-  PostTaskToExecutor(base::BindOnce(
-      [](Cronet_UrlRequestCallback* callback,
-         Cronet_UrlRequestImpl* url_request) {
-        if (url_request->DestroyRequestUnlessDone(
-                Cronet_RequestFinishedInfo_FINISHED_REASON_SUCCEEDED)) {
-          return;
-        }
-        Cronet_UrlRequestCallback_OnSucceeded(
-            callback, url_request, url_request->response_info_.get());
-      },
-      callback_, url_request_));
+  // Invoke Cronet_UrlRequestCallback_OnSucceeded on client executor.
+  url_request_->PostTaskToExecutor(
+      base::BindOnce(&Cronet_UrlRequestImpl::InvokeCallbackOnSucceeded,
+                     base::Unretained(url_request_)));
 }
 
-void Cronet_UrlRequestImpl::Callback::OnError(int net_error,
+void Cronet_UrlRequestImpl::NetworkTasks::OnError(
+    int net_error,
     int quic_error,
     const std::string& error_string,
     int64_t received_byte_count) {
@@ -485,42 +534,35 @@ void Cronet_UrlRequestImpl::Callback::OnError(int net_error,
   base::AutoLock lock(url_request_->lock_);
   if (url_request_->response_info_)
     url_request_->response_info_->received_byte_count = received_byte_count;
-
+  if (url_request_->upload_data_sink_)
+    url_request_->upload_data_sink_->PostCloseToExecutor();
   url_request_->error_ =
       CreateCronet_Error(net_error, quic_error, error_string);
 
-  // Invoke Cronet_UrlRequestCallback_OnFailed using OnceClosure.
-  PostTaskToExecutor(base::BindOnce(
-      [](Cronet_UrlRequestCallback* callback,
-         Cronet_UrlRequestImpl* url_request) {
-        if (url_request->DestroyRequestUnlessDone(
-                Cronet_RequestFinishedInfo_FINISHED_REASON_FAILED)) {
-          return;
-        }
-        Cronet_UrlRequestCallback_OnFailed(callback, url_request,
-                                           url_request->response_info_.get(),
-                                           url_request->error_.get());
-      },
-      callback_, url_request_));
+  // Invoke Cronet_UrlRequestCallback_OnFailed on client executor.
+  url_request_->PostTaskToExecutor(
+      base::BindOnce(&Cronet_UrlRequestImpl::InvokeCallbackOnFailed,
+                     base::Unretained(url_request_)));
 }
 
-void Cronet_UrlRequestImpl::Callback::OnCanceled() {
+void Cronet_UrlRequestImpl::NetworkTasks::OnCanceled() {
   DCHECK_CALLED_ON_VALID_THREAD(network_thread_checker_);
-  PostTaskToExecutor(base::BindOnce(
-      [](Cronet_UrlRequestCallback* callback,
-         Cronet_UrlRequestImpl* url_request) {
-        Cronet_UrlRequestCallback_OnCanceled(callback, url_request,
-                                             url_request->response_info_.get());
-      },
-      callback_, url_request_));
+  base::AutoLock lock(url_request_->lock_);
+  if (url_request_->upload_data_sink_)
+    url_request_->upload_data_sink_->PostCloseToExecutor();
+
+  // Invoke Cronet_UrlRequestCallback_OnCanceled on client executor.
+  url_request_->PostTaskToExecutor(
+      base::BindOnce(&Cronet_UrlRequestImpl::InvokeCallbackOnCanceled,
+                     base::Unretained(url_request_)));
 }
 
-void Cronet_UrlRequestImpl::Callback::OnDestroyed() {
+void Cronet_UrlRequestImpl::NetworkTasks::OnDestroyed() {
   DCHECK_CALLED_ON_VALID_THREAD(network_thread_checker_);
   DCHECK(url_request_);
 }
 
-void Cronet_UrlRequestImpl::Callback::OnMetricsCollected(
+void Cronet_UrlRequestImpl::NetworkTasks::OnMetricsCollected(
     const base::Time& request_start_time,
     const base::TimeTicks& request_start,
     const base::TimeTicks& dns_start,
@@ -541,14 +583,6 @@ void Cronet_UrlRequestImpl::Callback::OnMetricsCollected(
   DCHECK_CALLED_ON_VALID_THREAD(network_thread_checker_);
 }
 
-void Cronet_UrlRequestImpl::Callback::PostTaskToExecutor(
-    base::OnceClosure task) {
-  Cronet_RunnablePtr runnable =
-      new cronet::OnceClosureRunnable(std::move(task));
-  // |runnable| is passed to executor, which destroys it after execution.
-  Cronet_Executor_Execute(executor_, runnable);
-}
-
 };  // namespace cronet
 
 CRONET_EXPORT Cronet_UrlRequestPtr Cronet_UrlRequest_Create() {

components/cronet/native/url_request.h

@@ -10,7 +10,6 @@
 
 #include "base/macros.h"
 #include "base/synchronization/lock.h"
-#include "base/synchronization/waitable_event.h"
 #include "components/cronet/cronet_url_request.h"
 #include "components/cronet/cronet_url_request_context.h"
 #include "components/cronet/native/generated/cronet.idl_impl_interface.h"
@@ -18,6 +17,7 @@
 namespace cronet {
 
 class Cronet_EngineImpl;
+class Cronet_UploadDataSinkImpl;
 
 // Implementation of Cronet_UrlRequest that uses CronetURLRequestContext.
 class Cronet_UrlRequestImpl : public Cronet_UrlRequest {
@@ -38,8 +38,12 @@ class Cronet_UrlRequestImpl : public Cronet_UrlRequest {
   bool IsDone() override;
   void GetStatus(Cronet_UrlRequestStatusListenerPtr listener) override;
 
+  // Upload data provider has reported error while reading or rewinding
+  // so request must fail.
+  void OnUploadDataProviderError(const std::string& error_message);
+
  private:
-  class Callback;
+  class NetworkTasks;
 
   // Return |true| if request has started and is now done.
   // Must be called under |lock_| held.
@@ -57,6 +61,19 @@ class Cronet_UrlRequestImpl : public Cronet_UrlRequest {
   bool DestroyRequestUnlessDoneLocked(
       Cronet_RequestFinishedInfo_FINISHED_REASON finished_reason);
 
+  // Helper method to post |task| to the |executor_|.
+  void PostTaskToExecutor(base::OnceClosure task);
+
+  // Helper methods to invoke application |callback_|.
+  void InvokeCallbackOnRedirectReceived();
+  void InvokeCallbackOnResponseStarted();
+  void InvokeCallbackOnReadCompleted(
+      std::unique_ptr<Cronet_Buffer> cronet_buffer,
+      int bytes_read);
+  void InvokeCallbackOnSucceeded();
+  void InvokeCallbackOnFailed();
+  void InvokeCallbackOnCanceled();
+
   // Synchronize access to |request_| and other objects below from different
   // threads.
   base::Lock lock_;
@@ -72,6 +89,14 @@ class Cronet_UrlRequestImpl : public Cronet_UrlRequest {
   // The error reported by request. May be nullptr if no error has occurred.
   std::unique_ptr<Cronet_Error> error_;
 
+  // The upload data stream if specified.
+  std::unique_ptr<Cronet_UploadDataSinkImpl> upload_data_sink_;
+
+  // Application callback interface, used, but not owned, by |this|.
+  Cronet_UrlRequestCallbackPtr callback_ = nullptr;
+  // Executor for application callback, used, but not owned, by |this|.
+  Cronet_ExecutorPtr executor_ = nullptr;
+
   // Cronet Engine used to run network operations. Not owned, accessed from
   // client thread. Must outlive this request.
   Cronet_EngineImpl* engine_;