[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/android/scoped_java_ref.h"
+#include "base/bind.h"
+#include "base/macros.h"
+#include "base/memory/ref_counted.h"
+#include "base/message_loop/message_loop_proxy.h"
+#include "base/threading/thread.h"
+#include "components/cronet/android/cronet_upload_data_stream_adapter.h"
+#include "jni/UploadDataStreamHandler_jni.h"
+#include "net/base/io_buffer.h"
+#include "net/base/net_errors.h"
+#include "net/base/upload_data_stream.h"
+
+namespace cronet {
+
+namespace {
+
+static const size_t kReadBufferSize = 32768;
+
+/**
+ * This class sets up the JNI for Java tests to invoke methods on the native
+ * CronetUploadDataStreamAdapter.
+ */
+class UploadDataStreamHandler {
+ public:
+  UploadDataStreamHandler(CronetUploadDataStreamAdapter* adapter,
+                          JNIEnv* env,
+                          jobject jupload_data_stream_handler);
+
+  ~UploadDataStreamHandler();
+
+  // Destroys |network_thread_| created by this class.
+  void Destroy();
+
+  // Posts a task to |network_thread_| to call the corresponding method of
+  // CronetUploadDataStreamAdapter on |adapter_|.
+
+  void Init();
+  void Read();
+  void Reset();
+
+  // Posts a task to |network_thread_| to check whether init complete callback
+  // has been invoked by net::UploadDataStream, and notifies the Java side of
+  // the result.
+  void CheckInitCallbackNotInvoked();
+  // Posts a task to |network_thread_| to check whether read complete callback
+  // has been invoked by net::UploadDataStream, and notifies the Java side of
+  // the result.
+  void CheckReadCallbackNotInvoked();
+
+ private:
+  // Complete callbacks that are passed to the |adapter_|.
+  void OnInitCompleted(int res);
+  void OnReadCompleted(int res);
+
+  // Helper methods that run corresponding task on |network_thread_|.
+
+  void InitOnNetworkThread();
+  void ReadOnNetworkThread();
+  void ResetOnNetworkThread();
+  void CheckInitCallbackNotInvokedOnNetworkThread();
+  void CheckReadCallbackNotInvokedOnNetworkThread();
+
+  // Notify the Java UploadDataStreamHandler that read has completed.
+  void NotifyJavaReadCompleted();
+
+  // True if |OnInitCompleted| callback has been invoked. It is set to false
+  // when init or reset is called again. Created on a Java thread, but are only
+  // accessed from |network_thread_|.
+  bool init_callback_invoked_;
+  // True if |OnReadCompleted| callback has been invoked. It is set to false
+  // when init or reset is called again. Created on a Java thread, but are only
+  // accessed from |network_thread_|.
+  bool read_callback_invoked_;
+  // Indicates the number of bytes read. It is reset to 0 when init, reset, or
+  // read is called again. Created on a Java thread, but are only accessed from
+  // |network_thread_|.
+  int bytes_read_;
+
+  // Created and destroyed on the same Java thread. This is where methods of
+  // CronetUploadDataStreamAdapter run on.
+  scoped_ptr<base::Thread> network_thread_;
+  // Created on a Java thread. Accessed only on |network_thread_|.
+  scoped_ptr<CronetUploadDataStreamAdapter> adapter_;
+  // Created and accessed only on |network_thread_|.
+  scoped_refptr<net::IOBufferWithSize> read_buffer_;
+  // A Java reference pointer for calling methods on the Java
+  // UploadDataStreamHandler object. Initialized during construction.
+  base::android::ScopedJavaGlobalRef<jobject> jupload_data_stream_handler_;
+
+  DISALLOW_COPY_AND_ASSIGN(UploadDataStreamHandler);
+};
+
+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() {
+  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() {
+  DCHECK(!network_thread_->task_runner()->BelongsToCurrentThread());
+  network_thread_->task_runner()->PostTask(
+      FROM_HERE, base::Bind(&UploadDataStreamHandler::InitOnNetworkThread,
+                            base::Unretained(this)));
+}
+
+void UploadDataStreamHandler::Read() {
+  DCHECK(!network_thread_->task_runner()->BelongsToCurrentThread());
+  network_thread_->task_runner()->PostTask(
+      FROM_HERE, base::Bind(&UploadDataStreamHandler::ReadOnNetworkThread,
+                            base::Unretained(this)));
+}
+
+void UploadDataStreamHandler::Reset() {
+  DCHECK(!network_thread_->task_runner()->BelongsToCurrentThread());
+  network_thread_->task_runner()->PostTask(
+      FROM_HERE, base::Bind(&UploadDataStreamHandler::ResetOnNetworkThread,
+                            base::Unretained(this)));
+}
+
+void UploadDataStreamHandler::CheckInitCallbackNotInvoked() {
+  DCHECK(!network_thread_->task_runner()->BelongsToCurrentThread());
+  network_thread_->task_runner()->PostTask(
+      FROM_HERE,
+      base::Bind(
+          &UploadDataStreamHandler::CheckInitCallbackNotInvokedOnNetworkThread,
+          base::Unretained(this)));
+}
+
+void UploadDataStreamHandler::CheckReadCallbackNotInvoked() {
+  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_ == false);
+}
+
+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_ == false);
+}
+
+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());
+}
+
+}  // namespace
+
+static void Init(JNIEnv* env,
+                 jclass jcaller,
+                 jlong jupload_data_stream_handler) {
+  UploadDataStreamHandler* handler =
+      reinterpret_cast<UploadDataStreamHandler*>(jupload_data_stream_handler);
+  DCHECK(handler);
+  handler->Init();
+}
+
+static void Read(JNIEnv* env,
+                 jclass jcaller,
+                 jlong jupload_data_stream_handler) {
+  UploadDataStreamHandler* handler =
+      reinterpret_cast<UploadDataStreamHandler*>(jupload_data_stream_handler);
+  DCHECK(handler);
+  handler->Read();
+}
+
+static void Reset(JNIEnv* env,
+                  jclass jcaller,
+                  jlong jupload_data_stream_handler) {
+  UploadDataStreamHandler* handler =
+      reinterpret_cast<UploadDataStreamHandler*>(jupload_data_stream_handler);
+  DCHECK(handler);
+  handler->Reset();
+}
+
+static void CheckInitCallbackNotInvoked(JNIEnv* env,
+                                        jclass jcaller,
+                                        jlong jupload_data_stream_handler) {
+  UploadDataStreamHandler* handler =
+      reinterpret_cast<UploadDataStreamHandler*>(jupload_data_stream_handler);
+  DCHECK(handler);
+  handler->CheckInitCallbackNotInvoked();
+}
+
+static void CheckReadCallbackNotInvoked(JNIEnv* env,
+                                        jclass jcaller,
+                                        jlong jupload_data_stream_handler) {
+  UploadDataStreamHandler* handler =
+      reinterpret_cast<UploadDataStreamHandler*>(jupload_data_stream_handler);
+  DCHECK(handler);
+  handler->CheckReadCallbackNotInvoked();
+}
+
+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);
+}
+
+static void Destroy(JNIEnv* env,
+                    jclass jcaller,
+                    jlong jupload_data_stream_handler) {
+  UploadDataStreamHandler* handler =
+      reinterpret_cast<UploadDataStreamHandler*>(jupload_data_stream_handler);
+  DCHECK(handler);
+  handler->Destroy();
+}
+
+bool UploadDataStreamHandlerRegisterJni(JNIEnv* env) {
+  return RegisterNativesImpl(env);
+}
+
+}  // namespace cronet