[Cronet] Upload support for async APIs

components/cronet/android/test/upload_data_stream_handler.cc

+// Copyright 2015 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#include "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/UploadDataStreamHandler_jni.h"
+#include "net/base/net_errors.h"
+
+namespace cronet {
+
+static const size_t kReadBufferSize = 32768;
+
+UploadDataStreamHandler::UploadDataStreamHandler(
+    CronetUploadDataStreamAdapter* adapter,
+    JNIEnv* env,
+    jobject jupload_data_stream_handler)
+    : init_callback_invoked_(false),
+      read_callback_invoked_(false),
+      bytes_read_(0),
+      network_thread_(new base::Thread("network")),
+      adapter_(adapter) {
+  base::Thread::Options options;
+  options.message_loop_type = base::MessageLoop::TYPE_IO;
+  network_thread_->StartWithOptions(options);
+  jupload_data_stream_handler_.Reset(env, jupload_data_stream_handler);
+}
+
+UploadDataStreamHandler::~UploadDataStreamHandler() {
+}
+
+void UploadDataStreamHandler::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 UploadDataStreamHandler::OnInitCompleted(int res) {
+  DCHECK(network_thread_->task_runner()->BelongsToCurrentThread());
+  init_callback_invoked_ = true;
+  JNIEnv* env = base::android::AttachCurrentThread();
+  cronet::Java_UploadDataStreamHandler_onInitCompleted(
+      env, jupload_data_stream_handler_.obj(), res);
+}
+
+void UploadDataStreamHandler::OnReadCompleted(int res) {
+  DCHECK(network_thread_->task_runner()->BelongsToCurrentThread());
+  read_callback_invoked_ = true;
+  bytes_read_ = res;
+  NotifyJavaReadCompleted();
+}
+
+void UploadDataStreamHandler::Init(JNIEnv* env, jobject jcaller) {
+  DCHECK(!network_thread_->task_runner()->BelongsToCurrentThread());
+  network_thread_->task_runner()->PostTask(
+      FROM_HERE, base::Bind(&UploadDataStreamHandler::InitOnNetworkThread,
+                            base::Unretained(this)));
+}
+
+void UploadDataStreamHandler::Read(JNIEnv* env, jobject jcaller) {
+  DCHECK(!network_thread_->task_runner()->BelongsToCurrentThread());
+  network_thread_->task_runner()->PostTask(
+      FROM_HERE, base::Bind(&UploadDataStreamHandler::ReadOnNetworkThread,
+                            base::Unretained(this)));
+}
+
+void UploadDataStreamHandler::Reset(JNIEnv* env, jobject jcaller) {
+  DCHECK(!network_thread_->task_runner()->BelongsToCurrentThread());
+  network_thread_->task_runner()->PostTask(
+      FROM_HERE, base::Bind(&UploadDataStreamHandler::ResetOnNetworkThread,
+                            base::Unretained(this)));
+}
+
+void UploadDataStreamHandler::CheckInitCallbackNotInvoked(JNIEnv* env,
+                                                          jobject jcaller) {
+  DCHECK(!network_thread_->task_runner()->BelongsToCurrentThread());
+  network_thread_->task_runner()->PostTask(
+      FROM_HERE,
+      base::Bind(
+          &UploadDataStreamHandler::CheckInitCallbackNotInvokedOnNetworkThread,
+          base::Unretained(this)));
+}
+
+void UploadDataStreamHandler::CheckReadCallbackNotInvoked(JNIEnv* env,
+                                                          jobject jcaller) {
+  DCHECK(!network_thread_->task_runner()->BelongsToCurrentThread());
+  network_thread_->task_runner()->PostTask(
+      FROM_HERE,
+      base::Bind(
+          &UploadDataStreamHandler::CheckReadCallbackNotInvokedOnNetworkThread,
+          base::Unretained(this)));
+}
+
+void UploadDataStreamHandler::InitOnNetworkThread() {
+  DCHECK(network_thread_->task_runner()->BelongsToCurrentThread());
+  init_callback_invoked_ = false;
+  read_buffer_ = nullptr;
+  bytes_read_ = 0;
+  int res = adapter_->Init(base::Bind(&UploadDataStreamHandler::OnInitCompleted,
+                                      base::Unretained(this)));
+  JNIEnv* env = base::android::AttachCurrentThread();
+  cronet::Java_UploadDataStreamHandler_onInitCalled(
+      env, jupload_data_stream_handler_.obj(), res);
+
+  if (res == net::OK) {
+    cronet::Java_UploadDataStreamHandler_onInitCompleted(
+        env, jupload_data_stream_handler_.obj(), res);
+  }
+}
+
+void UploadDataStreamHandler::ReadOnNetworkThread() {
+  DCHECK(network_thread_->task_runner()->BelongsToCurrentThread());
+  read_callback_invoked_ = false;
+  if (!read_buffer_.get())
+    read_buffer_ = new net::IOBufferWithSize(kReadBufferSize);
+
+  int bytes_read =
+      adapter_->Read(read_buffer_.get(), kReadBufferSize,
+                     base::Bind(&UploadDataStreamHandler::OnReadCompleted,
+                                base::Unretained(this)));
+  if (bytes_read == net::OK) {
+    bytes_read_ = bytes_read;
+    NotifyJavaReadCompleted();
+  }
+}
+
+void UploadDataStreamHandler::ResetOnNetworkThread() {
+  DCHECK(network_thread_->task_runner()->BelongsToCurrentThread());
+  init_callback_invoked_ = false;
+  read_callback_invoked_ = false;
+  read_buffer_ = nullptr;
+  bytes_read_ = 0;
+  adapter_->Reset();
+  JNIEnv* env = base::android::AttachCurrentThread();
+  cronet::Java_UploadDataStreamHandler_onResetCompleted(
+      env, jupload_data_stream_handler_.obj());
+}
+
+void UploadDataStreamHandler::CheckInitCallbackNotInvokedOnNetworkThread() {
+  DCHECK(network_thread_->task_runner()->BelongsToCurrentThread());
+  JNIEnv* env = base::android::AttachCurrentThread();
+  cronet::Java_UploadDataStreamHandler_onCheckInitCallbackNotInvoked(
+      env, jupload_data_stream_handler_.obj(), !init_callback_invoked_);
+}
+
+void UploadDataStreamHandler::CheckReadCallbackNotInvokedOnNetworkThread() {
+  DCHECK(network_thread_->task_runner()->BelongsToCurrentThread());
+  JNIEnv* env = base::android::AttachCurrentThread();
+  cronet::Java_UploadDataStreamHandler_onCheckReadCallbackNotInvoked(
+      env, jupload_data_stream_handler_.obj(), !read_callback_invoked_);
+}
+
+void UploadDataStreamHandler::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_UploadDataStreamHandler_onReadCompleted(
+      env, jupload_data_stream_handler_.obj(), bytes_read_,
+      base::android::ConvertUTF8ToJavaString(env, data_read).Release());
+}
+
+static jlong CreateUploadDataStreamHandler(JNIEnv* env,
+                                           jobject jupload_data_stream_handler,
+                                           jlong jupload_data_stream_adapter) {
+  CronetUploadDataStreamAdapter* adapter =
+      reinterpret_cast<CronetUploadDataStreamAdapter*>(
+          jupload_data_stream_adapter);
+  UploadDataStreamHandler* handler =
+      new UploadDataStreamHandler(adapter, env, jupload_data_stream_handler);
+  return reinterpret_cast<jlong>(handler);
+}
+
+bool UploadDataStreamHandlerRegisterJni(JNIEnv* env) {
+  return RegisterNativesImpl(env);
+}
+
+}  // namespace cronet