[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..c00823c2185d697f5bb59d10ba582f24f45b08e4
--- /dev/null
+++ b/components/cronet/android/test/upload_data_stream_handler.cc
@@ -0,0 +1,294 @@
+// Copyright 2014 The Chromium Authors. All rights reserved.

[mmenke, 2015/01/22 20:19:07]

2015, for all of these files (No idea if you should keep the 2014 on the files I
handed off to you or not)

[xunjieli, 2015/01/23 16:39:48]

Done. 2015 it is! :)

+// 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;
+
+class UploadDataStreamHandler {

[mmenke, 2015/01/22 20:19:07]

Should have a description of this class.

[xunjieli, 2015/01/23 16:39:48]

Done.

+ public:
+  UploadDataStreamHandler(CronetUploadDataStreamAdapter* adapter,
+                          JNIEnv* env,
+                          jobject jupload_data_stream_handler);
+
+  ~UploadDataStreamHandler();
+
+  void Destroy();

[mmenke, 2015/01/22 20:19:08]

Suggest separating this from the other methods with a blank line, and giving it
its own comment.

[xunjieli, 2015/01/23 16:39:48]

Done.

+  void Init();

[mmenke, 2015/01/22 20:19:07]

Suggest a comment before this block of methods, and one before the second block,
telling when/where they're called...Or may even want to comment them
individually.

[xunjieli, 2015/01/23 16:39:48]

Done.

+  void Read();
+  void Reset();
+  scoped_refptr<base::SingleThreadTaskRunner> GetTaskRunner() const;

[mmenke, 2015/01/22 20:19:07]

I don't think this is needed - can just access the task runner directly
everywhere.

[xunjieli, 2015/01/23 16:39:48]

Done.

+  void CheckInitCallbackNotInvoked();
+  void CheckReadCallbackNotInvoked();

[mmenke, 2015/01/22 20:19:07]

Declaration order should match definition order.

[xunjieli, 2015/01/23 16:39:48]

Done.

+
+ private:
+  void OnInitComplete(int res);
+  void OnReadComplete(int res);
+  void InitOnNetworkThread();
+  void ReadOnNetworkThread();
+  void ResetOnNetworkThread();
+  void CheckInitCallbackNotInvokedOnNetworkThread();
+  void CheckReadCallbackNotInvokedOnNetworkThread();
+  std::string Data() const;
+
+  bool init_callback_invoked_;
+  bool read_callback_invoked_;
+  int bytes_read_;

[mmenke, 2015/01/22 20:19:08]

Should include descriptions of these (When set to true/false, in particular, and
where they live.  Same goes for the other member variables)

[xunjieli, 2015/01/23 16:39:48]

Done.

[xunjieli, 2015/01/23 16:39:48]

Done.

+
+  base::Thread* network_thread_;

[mmenke, 2015/01/22 20:19:07]

Should mention lifetime / creation / destruction of this object, in particular.

[xunjieli, 2015/01/23 16:39:49]

Done.

+  scoped_ptr<CronetUploadDataStreamAdapter> adapter_;
+  scoped_refptr<net::IOBufferWithSize> read_buffer_;
+  base::android::ScopedJavaGlobalRef<jobject> jupload_data_stream_handler_;
+
+  DISALLOW_COPY_AND_ASSIGN(UploadDataStreamHandler);
+};
+
+}  // namespace

[mmenke, 2015/01/22 20:19:08]

This namespace should extend down to the last method of UploadDataStreamHandler.

[xunjieli, 2015/01/23 16:39:48]

Done.

+
+UploadDataStreamHandler::UploadDataStreamHandler(
+    CronetUploadDataStreamAdapter* adapter,
+    JNIEnv* env,
+    jobject jupload_data_stream_handler)
+    : init_callback_invoked_(false),
+      read_callback_invoked_(false),
+      bytes_read_(0),
+      adapter_(adapter) {
+  network_thread_ = new base::Thread("network");

[mmenke, 2015/01/22 20:19:07]

nit:  Think this can be in the initializer list.

[xunjieli, 2015/01/23 16:39:48]

Done.

+  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(!GetTaskRunner()->BelongsToCurrentThread());
+  // Stick network_thread_ in a local, as |this| may be destroyed from the
+  // network thread before delete network_thread is called.

[mmenke, 2015/01/22 20:19:07]

nit:  "as" -> "so", "before delete network_thread is called." -> "before the
network thread is destroyed."

[xunjieli, 2015/01/23 16:39:48]

Done.

+  base::Thread* network_thread = network_thread_;

[mmenke, 2015/01/22 20:19:07]

Ownership is a bit weird, but I suggest making both of these scoped_ptrs, and
using Pass here.

[xunjieli, 2015/01/23 16:39:48]

Done. Copied from other code from elsewhere. Why do we need to delete |this| on
network thread? 

[mmenke, 2015/01/23 21:31:41]

Weak pointer factories must be destroyed on the thread the weak pointer is bound
to.  The handler no longer has a weak pointer itself, but the
CronetUploadDataStream[Adapter?] has one, for callbacks from the Java side.

In general, it's safest to only use objects from a single thread, and destroy
them on it, too.  We could even create an object owned by Java, and have it have
own the network thread and have a pointer to the handle, and then never touch
the network thread (Other than passing around its pointer and creating it) on
the Java thread, if we wanted to be even more strict about that.  Doesn't seem
worth the effort, in this case, but it's a common pattern.

+  GetTaskRunner()->DeleteSoon(FROM_HERE, this);
+  // Deleting thread stops it after all tasks are completed.
+  delete network_thread;
+}
+
+void UploadDataStreamHandler::OnInitComplete(int res) {
+  DCHECK(GetTaskRunner()->BelongsToCurrentThread());
+  init_callback_invoked_ = true;
+  JNIEnv* env = base::android::AttachCurrentThread();
+  cronet::Java_UploadDataStreamHandler_onInitCompleted(
+      env, jupload_data_stream_handler_.obj(), res);
+}
+
+void UploadDataStreamHandler::OnReadComplete(int res) {
+  DCHECK(GetTaskRunner()->BelongsToCurrentThread());
+  read_callback_invoked_ = true;
+  bytes_read_ = res;
+  JNIEnv* env = base::android::AttachCurrentThread();
+  cronet::Java_UploadDataStreamHandler_onReadCompleted(
+      env, jupload_data_stream_handler_.obj(), res,
+      base::android::ConvertUTF8ToJavaString(env, this->Data()).Release());

[mmenke, 2015/01/22 20:19:07]

This callback is idental to the one in ReadOnNetworkThread.  Suggest a common
subroutine (Also suggest getting rid of the Data() method and just inlining the
code to get it)

[xunjieli, 2015/01/23 16:39:48]

Done.

+}
+
+void UploadDataStreamHandler::Init() {
+  DCHECK(!GetTaskRunner()->BelongsToCurrentThread());
+  GetTaskRunner()->PostTask(
+      FROM_HERE, base::Bind(&UploadDataStreamHandler::InitOnNetworkThread,
+                            base::Unretained(this)));
+}
+
+void UploadDataStreamHandler::InitOnNetworkThread() {
+  DCHECK(GetTaskRunner()->BelongsToCurrentThread());
+  init_callback_invoked_ = false;
+  read_buffer_ = NULL;

[mmenke, 2015/01/22 20:19:08]

nit:  nullptr

[xunjieli, 2015/01/23 16:39:48]

Done.

+  bytes_read_ = 0;
+  int res = adapter_->Init(base::Bind(&UploadDataStreamHandler::OnInitComplete,
+                                      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::Read() {
+  DCHECK(!GetTaskRunner()->BelongsToCurrentThread());
+  GetTaskRunner()->PostTask(
+      FROM_HERE, base::Bind(&UploadDataStreamHandler::ReadOnNetworkThread,
+                            base::Unretained(this)));
+}
+
+void UploadDataStreamHandler::ReadOnNetworkThread() {
+  DCHECK(GetTaskRunner()->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::OnReadComplete,
+                                base::Unretained(this)));
+  if (bytes_read == net::OK) {
+    JNIEnv* env = base::android::AttachCurrentThread();
+    cronet::Java_UploadDataStreamHandler_onReadCompleted(
+        env, jupload_data_stream_handler_.obj(), bytes_read,
+        base::android::ConvertUTF8ToJavaString(env, this->Data()).Release());
+  }
+}
+
+void UploadDataStreamHandler::Reset() {
+  DCHECK(!GetTaskRunner()->BelongsToCurrentThread());
+  GetTaskRunner()->PostTask(
+      FROM_HERE, base::Bind(&UploadDataStreamHandler::ResetOnNetworkThread,
+                            base::Unretained(this)));
+}
+
+void UploadDataStreamHandler::ResetOnNetworkThread() {
+  DCHECK(GetTaskRunner()->BelongsToCurrentThread());
+  read_buffer_ = NULL;
+  bytes_read_ = 0;
+  adapter_->Reset();
+  JNIEnv* env = base::android::AttachCurrentThread();
+  cronet::Java_UploadDataStreamHandler_onResetCompleted(
+      env, jupload_data_stream_handler_.obj());
+}
+
+void UploadDataStreamHandler::CheckInitCallbackNotInvoked() {
+  DCHECK(!GetTaskRunner()->BelongsToCurrentThread());
+  GetTaskRunner()->PostTask(
+      FROM_HERE,
+      base::Bind(
+          &UploadDataStreamHandler::CheckInitCallbackNotInvokedOnNetworkThread,
+          base::Unretained(this)));
+}
+
+void UploadDataStreamHandler::CheckInitCallbackNotInvokedOnNetworkThread() {
+  DCHECK(GetTaskRunner()->BelongsToCurrentThread());
+  JNIEnv* env = base::android::AttachCurrentThread();
+  cronet::Java_UploadDataStreamHandler_onCheckInitCallbackNotInvoked(
+      env, jupload_data_stream_handler_.obj(), init_callback_invoked_ == false);
+}
+
+void UploadDataStreamHandler::CheckReadCallbackNotInvoked() {
+  DCHECK(!GetTaskRunner()->BelongsToCurrentThread());
+  GetTaskRunner()->PostTask(
+      FROM_HERE,
+      base::Bind(
+          &UploadDataStreamHandler::CheckReadCallbackNotInvokedOnNetworkThread,
+          base::Unretained(this)));
+}
+
+void UploadDataStreamHandler::CheckReadCallbackNotInvokedOnNetworkThread() {
+  DCHECK(GetTaskRunner()->BelongsToCurrentThread());
+  JNIEnv* env = base::android::AttachCurrentThread();
+  cronet::Java_UploadDataStreamHandler_onCheckReadCallbackNotInvoked(
+      env, jupload_data_stream_handler_.obj(), read_callback_invoked_ == false);
+}
+
+std::string UploadDataStreamHandler::Data() const {
+  DCHECK(GetTaskRunner()->BelongsToCurrentThread());
+  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 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 void Destroy(JNIEnv* env,

[mmenke, 2015/01/22 20:19:07]

This should probably be after CreateUploadDataStreamHandler.

[xunjieli, 2015/01/23 16:39:48]

Done.

+                    jclass jcaller,
+                    jlong jupload_data_stream_handler) {
+  UploadDataStreamHandler* handler =
+      reinterpret_cast<UploadDataStreamHandler*>(jupload_data_stream_handler);
+  DCHECK(handler);
+  handler->Destroy();
+}
+
+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