[Cronet] Upload support for async APIs

components/cronet/android/test/upload_data_stream_handler.cc

Index: components/cronet/android/test/upload_data_stream_handler.cc
diff --git a/components/cronet/android/test/upload_data_stream_handler.cc b/components/cronet/android/test/upload_data_stream_handler.cc
new file mode 100644
index 0000000000000000000000000000000000000000..509ecf7a8bcdae71d31ce05339ada1ee6eaba37c
--- /dev/null
+++ b/components/cronet/android/test/upload_data_stream_handler.cc
@@ -0,0 +1,270 @@
+// Copyright 2014 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/callback_helpers.h"
+#include "base/location.h"
+#include "base/logging.h"
+#include "base/macros.h"
+#include "base/memory/ref_counted.h"
+#include "base/memory/weak_ptr.h"
+#include "base/message_loop/message_loop_proxy.h"
+#include "components/cronet/android/cronet_upload_data_stream_adapter.h"
+#include "components/cronet/android/cronet_url_request_adapter.h"
+#include "jni/UploadDataStreamHandler_jni.h"
+#include "net/base/io_buffer.h"
+#include "net/base/net_errors.h"
+#include "net/base/test_completion_callback.h"
+#include "net/base/upload_data_stream.h"
+
+namespace cronet {
+
+static const size_t kReadBufferSize = 32768;
+
+bool UploadDataStreamHandlerRegisterJni(JNIEnv* env) {

[mmenke, 2015/01/21 18:48:58]

Declaration order should match definition order - this should be at the bottom
of the file.

[xunjieli, 2015/01/21 23:49:24]

Done.

+  return RegisterNativesImpl(env);
+};
+
+class UploadDataStreamHandlerDelegate
+    : public UploadDataStreamHandler::Delegate {
+ public:
+  UploadDataStreamHandlerDelegate(JNIEnv* env,
+                                  jobject jupload_data_stream_handler) {
+    jupload_data_stream_handler_.Reset(env, jupload_data_stream_handler);
+  }
+
+  ~UploadDataStreamHandlerDelegate() override {}
+
+  void OnInitCalled(int res) {
+    JNIEnv* env = base::android::AttachCurrentThread();
+    cronet::Java_UploadDataStreamHandler_onInitCalled(
+        env, jupload_data_stream_handler_.obj(), res);
+  }
+
+  void OnInitCompleted(int res) {
+    JNIEnv* env = base::android::AttachCurrentThread();
+    cronet::Java_UploadDataStreamHandler_onInitCompleted(
+        env, jupload_data_stream_handler_.obj(), res);
+  }
+
+  void OnReadCompleted(int res) {
+    JNIEnv* env = base::android::AttachCurrentThread();
+    cronet::Java_UploadDataStreamHandler_onReadCompleted(
+        env, jupload_data_stream_handler_.obj(), res);
+  }
+
+  void OnResetCompleted() {
+    JNIEnv* env = base::android::AttachCurrentThread();
+    cronet::Java_UploadDataStreamHandler_onResetCompleted(
+        env, jupload_data_stream_handler_.obj());
+  }
+
+  void OnCheckIfInitCallbackInvoked(bool res) {
+    JNIEnv* env = base::android::AttachCurrentThread();
+    cronet::Java_UploadDataStreamHandler_onCheckIfInitCallbackInvoked(
+        env, jupload_data_stream_handler_.obj(), res);
+  }
+
+  void OnCheckIfReadCallbackInvoked(bool res) {
+    JNIEnv* env = base::android::AttachCurrentThread();
+    cronet::Java_UploadDataStreamHandler_onCheckIfReadCallbackInvoked(
+        env, jupload_data_stream_handler_.obj(), res);
+  }
+
+ private:
+  // Initialized on construction, effectively constant.
+  base::android::ScopedJavaGlobalRef<jobject> jupload_data_stream_handler_;
+
+  DISALLOW_COPY_AND_ASSIGN(UploadDataStreamHandlerDelegate);
+};
+
+UploadDataStreamHandler::UploadDataStreamHandler(
+    CronetUploadDataStreamAdapter* adapter,
+    JNIEnv* env,
+    jobject jupload_data_stream_handler)
+    : init_complete_(false),
+      read_complete_(false),
+      bytes_read_(-1),
+      adapter_(adapter),
+      weak_factory_(this) {
+  network_thread_ = new base::Thread("network");

[mmenke, 2015/01/21 18:48:58]

Problem:  Once we start mucking with weak references on some thread, the weak
factory can only be used (And deleted) on that thread.  But this class also owns
the network thread, and threads can't be deleted from a method called on that
thread.

This makes clean deletion a bit of a problem.  Currently, it looks like we
aren't cleaning things up.  One way to do that is a method that moves
network_thread_ into a local scoped_ptr, posts a task to delete the Handler on
that thread, and then deletes the method.


There are a couple alternatives:  The simplest, and the one I recommend:  Get
rid of the weak factory, and just use base::Unretained(this).  Since the handler
owns the network thread, which is the only place we can dereference the weak
pointers anyways, this should be safe.

[xunjieli, 2015/01/21 23:49:23]

Got it. Thanks for the detailed explanation! 

+  base::Thread::Options options;
+  options.message_loop_type = base::MessageLoop::TYPE_IO;
+  network_thread_->StartWithOptions(options);
+  delegate_.reset(
+      new UploadDataStreamHandlerDelegate(env, jupload_data_stream_handler));
+}
+
+UploadDataStreamHandler::~UploadDataStreamHandler() {
+}
+
+// Called by the data provider.
+// On data provider's executor.

[mmenke, 2015/01/21 18:48:58]

Method level comments should go above the method declarations, not their
definitions.

[xunjieli, 2015/01/21 23:49:23]

Done. Right, haven't cleaned this up. For my own reference mostly, as I often
lose track about which method is called by whom :)

+void UploadDataStreamHandler::OnInitComplete(int res) {
+  init_complete_ = true;
+  delegate_->OnInitCompleted(res);
+}
+
+// Called by the data provider.
+// On data provider's executor.
+void UploadDataStreamHandler::OnReadComplete(int res) {
+  read_complete_ = true;
+  bytes_read_ = res;
+  delegate_->OnReadCompleted(res);
+}
+
+void UploadDataStreamHandler::Init() {
+  init_complete_ = false;

[mmenke, 2015/01/21 18:48:58]

This shouldn't be touched on the IO thread.  May make life simpler/safer to make
this purely a network thread class, with the possible exception of grabbing the
network thread and destruction.

[xunjieli, 2015/01/21 23:49:23]

Done. Agreed!

+  GetTaskRunner()->PostTask(
+      FROM_HERE, base::Bind(&UploadDataStreamHandler::InitOnNetworkThread,
+                            weak_factory_.GetWeakPtr()));
+}
+
+void UploadDataStreamHandler::InitOnNetworkThread() {
+  DCHECK(GetTaskRunner()->BelongsToCurrentThread());
+  int res = adapter_->Init(base::Bind(&UploadDataStreamHandler::OnInitComplete,
+                                      weak_factory_.GetWeakPtr()));
+  delegate_->OnInitCalled(res);
+  if (res == net::OK) {
+    delegate_->OnInitCompleted(res);
+  }
+}
+
+void UploadDataStreamHandler::Read() {
+  read_complete_ = false;

[mmenke, 2015/01/21 18:48:58]

This shouldn't be touched on the IO thread

[xunjieli, 2015/01/21 23:49:24]

Done.

+  GetTaskRunner()->PostTask(
+      FROM_HERE, base::Bind(&UploadDataStreamHandler::ReadOnNetworkThread,
+                            weak_factory_.GetWeakPtr()));

[mmenke, 2015/01/21 18:48:58]

The GetWeakPtr calls aren't safe.  According to the WeakPtrFactory docs, once a
weak pointer is dereferenced, the factory and the weak pointers can only be
dereferenced on that thread.  So you need to grab a weak pointer once, and then
just use it for all these calls.

[xunjieli, 2015/01/21 23:49:23]

Ahh I see. That's how it is used. Thanks!

+}
+
+void UploadDataStreamHandler::ReadOnNetworkThread() {
+  DCHECK(GetTaskRunner()->BelongsToCurrentThread());
+  if (!read_buffer_.get())
+    read_buffer_ = new net::IOBufferWithSize(kReadBufferSize);
+
+  int res = adapter_->Read(read_buffer_.get(), kReadBufferSize,
+                           base::Bind(&UploadDataStreamHandler::OnReadComplete,
+                                      weak_factory_.GetWeakPtr()));
+  if (res == net::OK) {
+    delegate_->OnReadCompleted(res);
+  }
+}
+
+void UploadDataStreamHandler::Reset() {
+  GetTaskRunner()->PostTask(
+      FROM_HERE, base::Bind(&UploadDataStreamHandler::ResetOnNetworkThread,
+                            weak_factory_.GetWeakPtr()));
+}
+
+void UploadDataStreamHandler::ResetOnNetworkThread() {
+  DCHECK(GetTaskRunner()->BelongsToCurrentThread());
+  read_buffer_ = NULL;
+  bytes_read_ = 0;
+  adapter_->Reset();
+  delegate_->OnResetCompleted();
+}
+
+void UploadDataStreamHandler::CheckIfInitCallbackInvoked() {
+  GetTaskRunner()->PostTask(
+      FROM_HERE, base::Bind(&UploadDataStreamHandler::CheckIfInitCallbackInvokedOnNetworkThread,
+                            weak_factory_.GetWeakPtr()));
+}
+
+void UploadDataStreamHandler::CheckIfInitCallbackInvokedOnNetworkThread() {

[mmenke, 2015/01/21 18:48:58]

DCHECK(GetTaskRunner()->BelongsToCurrentThread());

[xunjieli, 2015/01/21 23:49:24]

Done.

+  delegate_->OnCheckIfInitCallbackInvoked(init_complete_);
+}
+
+void UploadDataStreamHandler::CheckIfReadCallbackInvoked() {
+  GetTaskRunner()->PostTask(
+      FROM_HERE, base::Bind(&UploadDataStreamHandler::CheckIfReadCallbackInvokedOnNetworkThread,
+                            weak_factory_.GetWeakPtr()));
+}
+
+void UploadDataStreamHandler::CheckIfReadCallbackInvokedOnNetworkThread() {

[mmenke, 2015/01/21 18:48:58]

DCHECK(GetTaskRunner()->BelongsToCurrentThread());

[xunjieli, 2015/01/21 23:49:23]

Done.

+  delegate_->OnCheckIfReadCallbackInvoked(read_complete_);
+}
+
+std::string UploadDataStreamHandler::Data() const {

[mmenke, 2015/01/21 18:48:58]

DCHECK(GetTaskRunner()->BelongsToCurrentThread());

[xunjieli, 2015/01/21 23:49:23]

Done.

+  if (read_buffer_.get() && bytes_read_ > 0)
+    return std::string(read_buffer_->data(), bytes_read_);
+  return "";
+}
+
+scoped_refptr<base::SingleThreadTaskRunner>
+UploadDataStreamHandler::GetTaskRunner() const {
+  return network_thread_->task_runner();
+}
+
+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 jstring GetData(JNIEnv* env,
+                       jclass jcaller,
+                       jlong jupload_data_stream_handler) {
+  UploadDataStreamHandler* handler =
+      reinterpret_cast<UploadDataStreamHandler*>(jupload_data_stream_handler);
+  DCHECK(handler);
+  return base::android::ConvertUTF8ToJavaString(env, handler->Data()).Release();

[mmenke, 2015/01/21 18:48:58]

This doesn't seem threadsafe.    Should probably pass it along in OnReadComplete

[xunjieli, 2015/01/21 23:49:24]

Done. You are right! this is not thread-safe.

+}
+
+static void CheckIfInitCallbackInvoked(JNIEnv* env,
+                                           jclass jcaller,
+                                           jlong jupload_data_stream_handler) {
+  UploadDataStreamHandler* handler =
+      reinterpret_cast<UploadDataStreamHandler*>(jupload_data_stream_handler);
+  DCHECK(handler);
+  handler->CheckIfInitCallbackInvoked();
+}
+
+static void CheckIfReadCallbackInvoked(JNIEnv* env,
+                                           jclass jcaller,
+                                           jlong jupload_data_stream_handler) {
+  UploadDataStreamHandler* handler =
+      reinterpret_cast<UploadDataStreamHandler*>(jupload_data_stream_handler);
+  DCHECK(handler);
+  handler->CheckIfReadCallbackInvoked();
+}
+
+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);
+}
+}  // namespace cronet