[Cronet] Upload support for async APIs

components/cronet/android/test/test_upload_data_stream_handler.cc

Index: components/cronet/android/test/test_upload_data_stream_handler.cc
diff --git a/components/cronet/android/test/test_upload_data_stream_handler.cc b/components/cronet/android/test/test_upload_data_stream_handler.cc
new file mode 100644
index 0000000000000000000000000000000000000000..786caae1ffb514b2fdee6699690a7f9c36f07415
--- /dev/null
+++ b/components/cronet/android/test/test_upload_data_stream_handler.cc
@@ -0,0 +1,186 @@
+// 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 "test_upload_data_stream_handler.h"
+
+#include <string>
+
+#include "base/android/jni_android.h"
+#include "base/android/jni_string.h"
+#include "base/bind.h"
+#include "base/message_loop/message_loop_proxy.h"
+#include "jni/TestUploadDataStreamHandler_jni.h"
+#include "net/base/net_errors.h"
+
+namespace cronet {
+
+static const size_t kReadBufferSize = 32768;
+
+TestUploadDataStreamHandler::TestUploadDataStreamHandler(
+    scoped_ptr<net::UploadDataStream> upload_data_stream,
+    JNIEnv* env,
+    jobject jtest_upload_data_stream_handler)
+    : init_callback_invoked_(false),
+      read_callback_invoked_(false),
+      bytes_read_(0),
+      network_thread_(new base::Thread("network")) {
+  upload_data_stream_ = upload_data_stream.Pass();
+  base::Thread::Options options;
+  options.message_loop_type = base::MessageLoop::TYPE_IO;
+  network_thread_->StartWithOptions(options);
+  jtest_upload_data_stream_handler_.Reset(env,
+                                          jtest_upload_data_stream_handler);
+}
+
+TestUploadDataStreamHandler::~TestUploadDataStreamHandler() {
+}
+
+void TestUploadDataStreamHandler::Destroy(JNIEnv* env, jobject jcaller) {
+  DCHECK(!network_thread_->task_runner()->BelongsToCurrentThread());
+  // Stick network_thread_ in a local, so |this| may be destroyed from the
+  // network thread before the network thread is destroyed.
+  scoped_ptr<base::Thread> network_thread = network_thread_.Pass();
+  network_thread->task_runner()->DeleteSoon(FROM_HERE, this);
+  // Deleting thread stops it after all tasks are completed.
+  network_thread.reset();
+}
+
+void TestUploadDataStreamHandler::OnInitCompleted(int res) {
+  DCHECK(network_thread_->task_runner()->BelongsToCurrentThread());
+  init_callback_invoked_ = true;
+  JNIEnv* env = base::android::AttachCurrentThread();
+  cronet::Java_TestUploadDataStreamHandler_onInitCompleted(
+      env, jtest_upload_data_stream_handler_.obj(), res);
+}
+
+void TestUploadDataStreamHandler::OnReadCompleted(int res) {
+  DCHECK(network_thread_->task_runner()->BelongsToCurrentThread());
+  read_callback_invoked_ = true;
+  bytes_read_ = res;
+  NotifyJavaReadCompleted();
+}
+
+void TestUploadDataStreamHandler::Init(JNIEnv* env, jobject jcaller) {
+  DCHECK(!network_thread_->task_runner()->BelongsToCurrentThread());
+  network_thread_->task_runner()->PostTask(
+      FROM_HERE, base::Bind(&TestUploadDataStreamHandler::InitOnNetworkThread,
+                            base::Unretained(this)));
+}
+
+void TestUploadDataStreamHandler::Read(JNIEnv* env, jobject jcaller) {
+  DCHECK(!network_thread_->task_runner()->BelongsToCurrentThread());
+  network_thread_->task_runner()->PostTask(
+      FROM_HERE, base::Bind(&TestUploadDataStreamHandler::ReadOnNetworkThread,
+                            base::Unretained(this)));
+}
+
+void TestUploadDataStreamHandler::Reset(JNIEnv* env, jobject jcaller) {
+  DCHECK(!network_thread_->task_runner()->BelongsToCurrentThread());
+  network_thread_->task_runner()->PostTask(
+      FROM_HERE, base::Bind(&TestUploadDataStreamHandler::ResetOnNetworkThread,
+                            base::Unretained(this)));
+}
+
+void TestUploadDataStreamHandler::CheckInitCallbackNotInvoked(JNIEnv* env,
+                                                              jobject jcaller) {
+  DCHECK(!network_thread_->task_runner()->BelongsToCurrentThread());
+  network_thread_->task_runner()->PostTask(
+      FROM_HERE, base::Bind(&TestUploadDataStreamHandler::
+                                CheckInitCallbackNotInvokedOnNetworkThread,
+                            base::Unretained(this)));
+}
+
+void TestUploadDataStreamHandler::CheckReadCallbackNotInvoked(JNIEnv* env,
+                                                              jobject jcaller) {
+  DCHECK(!network_thread_->task_runner()->BelongsToCurrentThread());
+  network_thread_->task_runner()->PostTask(
+      FROM_HERE, base::Bind(&TestUploadDataStreamHandler::
+                                CheckReadCallbackNotInvokedOnNetworkThread,
+                            base::Unretained(this)));
+}
+
+void TestUploadDataStreamHandler::InitOnNetworkThread() {
+  DCHECK(network_thread_->task_runner()->BelongsToCurrentThread());
+  init_callback_invoked_ = false;
+  read_buffer_ = nullptr;
+  bytes_read_ = 0;
+  int res = upload_data_stream_->Init(base::Bind(
+      &TestUploadDataStreamHandler::OnInitCompleted, base::Unretained(this)));
+  JNIEnv* env = base::android::AttachCurrentThread();
+  cronet::Java_TestUploadDataStreamHandler_onInitCalled(
+      env, jtest_upload_data_stream_handler_.obj(), res);
+
+  if (res == net::OK) {
+    cronet::Java_TestUploadDataStreamHandler_onInitCompleted(
+        env, jtest_upload_data_stream_handler_.obj(), res);
+  }
+}
+
+void TestUploadDataStreamHandler::ReadOnNetworkThread() {
+  DCHECK(network_thread_->task_runner()->BelongsToCurrentThread());
+  read_callback_invoked_ = false;
+  if (!read_buffer_.get())
+    read_buffer_ = new net::IOBufferWithSize(kReadBufferSize);
+
+  int bytes_read = upload_data_stream_->Read(
+      read_buffer_.get(), kReadBufferSize,
+      base::Bind(&TestUploadDataStreamHandler::OnReadCompleted,
+                 base::Unretained(this)));
+  if (bytes_read == net::OK) {
+    bytes_read_ = bytes_read;
+    NotifyJavaReadCompleted();
+  }
+}
+
+void TestUploadDataStreamHandler::ResetOnNetworkThread() {
+  DCHECK(network_thread_->task_runner()->BelongsToCurrentThread());
+  read_buffer_ = nullptr;
+  bytes_read_ = 0;
+  upload_data_stream_->Reset();
+  JNIEnv* env = base::android::AttachCurrentThread();
+  cronet::Java_TestUploadDataStreamHandler_onResetCompleted(
+      env, jtest_upload_data_stream_handler_.obj());
+}
+
+void TestUploadDataStreamHandler::CheckInitCallbackNotInvokedOnNetworkThread() {
+  DCHECK(network_thread_->task_runner()->BelongsToCurrentThread());
+  JNIEnv* env = base::android::AttachCurrentThread();
+  cronet::Java_TestUploadDataStreamHandler_onCheckInitCallbackNotInvoked(
+      env, jtest_upload_data_stream_handler_.obj(), !init_callback_invoked_);
+}
+
+void TestUploadDataStreamHandler::CheckReadCallbackNotInvokedOnNetworkThread() {
+  DCHECK(network_thread_->task_runner()->BelongsToCurrentThread());
+  JNIEnv* env = base::android::AttachCurrentThread();
+  cronet::Java_TestUploadDataStreamHandler_onCheckReadCallbackNotInvoked(
+      env, jtest_upload_data_stream_handler_.obj(), !read_callback_invoked_);
+}
+
+void TestUploadDataStreamHandler::NotifyJavaReadCompleted() {
+  DCHECK(network_thread_->task_runner()->BelongsToCurrentThread());
+  JNIEnv* env = base::android::AttachCurrentThread();
+  std::string data_read = "";
+  if (read_buffer_.get() && bytes_read_ > 0)
+    data_read = std::string(read_buffer_->data(), bytes_read_);
+  cronet::Java_TestUploadDataStreamHandler_onReadCompleted(
+      env, jtest_upload_data_stream_handler_.obj(), bytes_read_,
+      base::android::ConvertUTF8ToJavaString(env, data_read).Release());
+}
+
+static jlong CreateTestUploadDataStreamHandler(
+    JNIEnv* env,
+    jobject jtest_upload_data_stream_handler,
+    jlong jupload_data_stream) {
+  scoped_ptr<net::UploadDataStream> upload_data_stream(
+      reinterpret_cast<net::UploadDataStream*>(jupload_data_stream));
+  TestUploadDataStreamHandler* handler = new TestUploadDataStreamHandler(
+      upload_data_stream.Pass(), env, jtest_upload_data_stream_handler);
+  return reinterpret_cast<jlong>(handler);
+}
+
+bool TestUploadDataStreamHandlerRegisterJni(JNIEnv* env) {
+  return RegisterNativesImpl(env);
+}
+
+}  // namespace cronet