[Cronet] Upload support for async APIs

components/cronet/android/cronet_upload_data_stream.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 "components/cronet/android/cronet_upload_data_stream.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/CronetUploadDataStream_jni.h"
+#include "net/base/io_buffer.h"
+#include "net/base/upload_data_stream.h"
+
+using base::android::ConvertUTF8ToJavaString;
+
+namespace cronet {
+
+namespace {
+
+// The Delegate holds onto a reference to the IOBuffer that is currently being
+// written to in Java, so may not be deleted until any read operation in Java
+// has completed.
+//
+// The Delegate is owned by the Java CronetUploadDataStream, and also owns a
+// reference to it.  The Delegate is only destroyed after the URLRequest
+// destroys the adapter and the CronetUploadDataStream has no operation pending,
+// at which point it also releases its reference to the CronetUploadDataStream.
+//
+// Failures don't go through the delegate, but directly to the Java request
+// object, since normally reads aren't allowed to fail during an upload.
+class CronetUploadDataStreamDelegate
+    : public CronetUploadDataStreamAdapter::Delegate {
+ public:
+  CronetUploadDataStreamDelegate(JNIEnv* env, jobject jupload_data_stream);
+  ~CronetUploadDataStreamDelegate() override {}
+
+  // CronetUploadDataStreamAdapter::Delegate implementation.  Called on network
+  // thread.
+  void InitializeOnNetworkThread(
+      base::WeakPtr<CronetUploadDataStreamAdapter> adapter) override;
+  void Read(net::IOBuffer* buffer, int buf_len) override;
+  void Rewind() override;
+  void OnAdapterDestroyed() override;
+
+  // Callbacks from Java, called on some Java thread.
+  void OnReadSucceeded(int position, int limit, bool final_chunk);
+  void OnRewindSucceeded();
+
+ private:
+  // Initialized on construction, effectively constant.
+  base::android::ScopedJavaGlobalRef<jobject> jupload_data_stream_;
+
+  // These are initialized in InitializeOnNetworkThread, so are safe to access
+  // during Java callbacks, which all happen after initialization.
+  scoped_refptr<base::MessageLoopProxy> network_message_loop_;
+  base::WeakPtr<CronetUploadDataStreamAdapter> adapter_;
+
+  // Used to keep the read buffer alive until the callback from Java has been
+  // received.
+  scoped_refptr<net::IOBuffer> buffer_;
+
+  DISALLOW_COPY_AND_ASSIGN(CronetUploadDataStreamDelegate);
+};
+
+CronetUploadDataStreamDelegate::CronetUploadDataStreamDelegate(
+    JNIEnv* env,
+    jobject jupload_data_stream) {
+  jupload_data_stream_.Reset(env, jupload_data_stream);
+}
+
+void CronetUploadDataStreamDelegate::InitializeOnNetworkThread(
+    base::WeakPtr<CronetUploadDataStreamAdapter> adapter) {
+  DCHECK(!adapter_);
+  DCHECK(!network_message_loop_.get());
+
+  adapter_ = adapter;
+  network_message_loop_ = base::MessageLoopProxy::current();
+  DCHECK(network_message_loop_);
+}
+
+void CronetUploadDataStreamDelegate::Read(net::IOBuffer* buffer, int buf_len) {
+  DCHECK(adapter_);
+  DCHECK(network_message_loop_);
+  DCHECK(network_message_loop_->BelongsToCurrentThread());
+  DCHECK_GT(buf_len, 0);
+  DCHECK(!buffer_.get());
+  buffer_ = buffer;
+
+  // TODO(mmenke):  Consider preserving the java buffer across reads, when the
+  // IOBuffer's data pointer and its length are unchanged.
+  JNIEnv* env = base::android::AttachCurrentThread();
+  base::android::ScopedJavaLocalRef<jobject> java_buffer(
+      env, env->NewDirectByteBuffer(buffer->data(), buf_len));
+  Java_CronetUploadDataStream_readData(env, jupload_data_stream_.obj(),
+                                       java_buffer.obj());
+}
+
+void CronetUploadDataStreamDelegate::Rewind() {
+  DCHECK(adapter_);
+  DCHECK(network_message_loop_->BelongsToCurrentThread());
+
+  JNIEnv* env = base::android::AttachCurrentThread();
+  Java_CronetUploadDataStream_rewind(env, jupload_data_stream_.obj());
+}
+
+void CronetUploadDataStreamDelegate::OnAdapterDestroyed() {
+  DCHECK(adapter_);
+  DCHECK(network_message_loop_->BelongsToCurrentThread());
+
+  JNIEnv* env = base::android::AttachCurrentThread();
+  Java_CronetUploadDataStream_onAdapterDestroyed(env,
+                                                 jupload_data_stream_.obj());
+}
+
+void CronetUploadDataStreamDelegate::OnReadSucceeded(int position,
+                                                     int limit,
+                                                     bool final_chunk) {
+  DCHECK(!network_message_loop_->BelongsToCurrentThread());
+  DCHECK_LE(position, limit);
+  DCHECK(position < limit || final_chunk);
+
+  int bytes_read = position;
+
+  buffer_ = nullptr;
+  DCHECK(network_message_loop_->PostTask(
+      FROM_HERE, base::Bind(&CronetUploadDataStreamAdapter::OnReadSuccess,
+                            adapter_, bytes_read, final_chunk)));
+}
+
+void CronetUploadDataStreamDelegate::OnRewindSucceeded() {
+  DCHECK(!network_message_loop_->BelongsToCurrentThread());
+
+  network_message_loop_->PostTask(
+      FROM_HERE,
+      base::Bind(&CronetUploadDataStreamAdapter::OnRewindSuccess, adapter_));
+}
+
+}  // namespace
+
+// Explicitly register static JNI functions.
+bool CronetUploadDataStreamRegisterJni(JNIEnv* env) {
+  return RegisterNativesImpl(env);
+}
+
+static jlong AttachUploadDataToRequest(JNIEnv* env,
+                                       jobject jupload_data_stream,
+                                       jlong jcronet_url_request_adapter,
+                                       jlong jlength) {
+  CronetURLRequestAdapter* request_adapter =
+      reinterpret_cast<CronetURLRequestAdapter*>(jcronet_url_request_adapter);
+  DCHECK(request_adapter != nullptr);
+
+  CronetUploadDataStreamDelegate* delegate =
+      new CronetUploadDataStreamDelegate(env, jupload_data_stream);
+
+  scoped_ptr<CronetUploadDataStreamAdapter> upload_adapter(
+      new CronetUploadDataStreamAdapter(delegate, jlength));
+
+  request_adapter->SetUpload(upload_adapter.Pass());
+
+  return reinterpret_cast<jlong>(delegate);
+}
+
+static jlong CreateDelegateForTesting(JNIEnv* env,
+                                      jobject jupload_data_stream) {
+  CronetUploadDataStreamDelegate* delegate =
+      new CronetUploadDataStreamDelegate(env, jupload_data_stream);
+  return reinterpret_cast<jlong>(delegate);
+}
+
+static jlong CreateAdapterForTesting(JNIEnv* env,
+                                     jobject jupload_data_stream,
+                                     jlong jlength,
+                                     jlong jdelegate) {
+  CronetUploadDataStreamDelegate* delegate =
+      reinterpret_cast<CronetUploadDataStreamDelegate*>(jdelegate);
+  CronetUploadDataStreamAdapter* upload_adapter =
+      new CronetUploadDataStreamAdapter(delegate, jlength);
+  return reinterpret_cast<jlong>(upload_adapter);
+}
+
+static void OnReadSucceeded(JNIEnv* env,
+                            jobject jcaller,
+                            jlong jupload_data_stream_delegate,
+                            jint position,
+                            jint limit,
+                            jboolean final_chunk) {
+  CronetUploadDataStreamDelegate* delegate =
+      reinterpret_cast<CronetUploadDataStreamDelegate*>(
+          jupload_data_stream_delegate);
+  DCHECK(delegate != nullptr);
+
+  delegate->OnReadSucceeded(position, limit, final_chunk);
+}
+
+static void OnRewindSucceeded(JNIEnv* env,
+                              jobject jcaller,
+                              jlong jupload_data_stream_delegate) {
+  CronetUploadDataStreamDelegate* delegate =
+      reinterpret_cast<CronetUploadDataStreamDelegate*>(
+          jupload_data_stream_delegate);
+  DCHECK(delegate != nullptr);
+
+  delegate->OnRewindSucceeded();
+}
+
+static void DestroyDelegate(JNIEnv* env,
+                            jclass jcronet_url_request_adapter,
+                            jlong jupload_data_stream_delegate) {
+  CronetUploadDataStreamDelegate* delegate =
+      reinterpret_cast<CronetUploadDataStreamDelegate*>(
+          jupload_data_stream_delegate);
+  DCHECK(delegate != nullptr);
+  delete delegate;
+}
+
+}  // namespace cronet