[Cronet] Upload support for async APIs

components/cronet/android/java/src/org/chromium/net/CronetUploadDataStream.java

+// 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.
+
+package org.chromium.net;
+
+import org.chromium.base.CalledByNative;
+import org.chromium.base.JNINamespace;
+
+import java.nio.ByteBuffer;
+import java.util.concurrent.Executor;
+
+/**
+ * Pass an upload body to a UrlRequest using an UploadDataProvider.
+ */
+@JNINamespace("cronet")
+public class CronetUploadDataStream implements UploadDataSink {
+    // These are never changed, once a request starts.
+    private final Executor mExecutor;
+    private final UploadDataProvider mDataProvider;
+    private final long mLength;
+    private CronetUrlRequest mRequest;
+
+    // Reusable read task, to reduce redundant memory allocation.
+    private final Runnable mReadTask = new Runnable() {
+        @Override
+        public void run() {
+            synchronized (mLock) {
+                if (mReading || mRewinding || mByteBuffer == null
+                        || mUploadDataStreamDelegate == 0) {
+                    throw new IllegalStateException(
+                            "Unexpected readData call.");
+                }
+                mReading = true;
+            }
+
+            try {
+                mDataProvider.read(CronetUploadDataStream.this, mByteBuffer);
+            } catch (Exception exception) {
+                onError(exception);
+            }
+        }
+    };
+
+    private ByteBuffer mByteBuffer = null;
+
+    // Lock that protects all subsequent variables. The delegate has to be
+    // protected to ensure safe shutdown, mReading and mRewinding are protected
+    // to robustly detect getting read/rewind results more often than expected.
+    private final Object mLock = new Object();
+
+    // Native adapter object, owned by the CronetUploadDataStream. It's only

[xunjieli, 2015/02/04 18:50:30]

The name is confusing. Can I refer to mUploadDataStreamDelegate as the delegate
(rather than the adapter) in the comment above?
and rename destroyAdapter() to destroyDelegate() ? WDYT?

+    // deleted after the native UploadDataStream object is destroyed. All
+    // access to the adapter is synchronized, for safe usage and cleanup.
+    private long mUploadDataStreamDelegate;
+
+    private boolean mReading = false;
+    private boolean mRewinding = false;
+    private boolean mDestroyAdapterPostponed = false;
+
+    public CronetUploadDataStream(UploadDataProvider dataProvider,
+            Executor executor) {
+        mExecutor = executor;
+        mDataProvider = dataProvider;
+        mLength = mDataProvider.getLength();
+    }
+
+    @SuppressWarnings("unused")
+    @CalledByNative
+    void readData(ByteBuffer byteBuffer) {
+        mByteBuffer = byteBuffer;
+        try {
+            mExecutor.execute(mReadTask);
+        } catch (Exception exception) {
+            onError(exception);
+        }
+    }
+
+    // TODO(mmenke): Consider implementing a cancel method.
+    // currently wait for any pending read to complete.
+
+    @SuppressWarnings("unused")
+    @CalledByNative
+    void rewind() {
+        Runnable task = new Runnable() {
+            @Override
+            public void run() {
+                synchronized (mLock) {
+                    if (mReading || mRewinding
+                            || mUploadDataStreamDelegate == 0) {
+                        throw new IllegalStateException(
+                                "Unexpected rewind call.");
+                    }
+                    mRewinding = true;
+                    try {
+                        mDataProvider.rewind(CronetUploadDataStream.this);
+                    } catch (Exception exception) {
+                        onError(exception);
+                    }
+                }
+            }
+        };
+
+        mExecutor.execute(task);
+    }
+
+    @SuppressWarnings("unused")
+    @CalledByNative
+    void onAdapterDestroyed() {
+        Runnable task = new Runnable() {
+            @Override
+            public void run() {
+                destroyAdapter();
+            }
+        };
+
+        mExecutor.execute(task);
+    }
+
+    private void onError(Exception exception) {
+        synchronized (mLock) {

[mmenke, 2015/02/04 16:46:00]

Maybe throw an exception if !mReading and !mRewinding?

[xunjieli, 2015/02/04 18:50:30]

Done.

+            mReading = false;
+            mRewinding = false;
+            mByteBuffer = null;
+            destroyAdapterIfPostponed();
+        }
+
+        // Just fail the request - simpler to fail directly, and
+        // UploadDataStream only supports failing during initialization, not
+        // while reading.
+        mRequest.onUploadException(exception);

[mmenke, 2015/02/04 16:46:00]

This should be safe, even if we just deleted the adapter, because in that case,
the request has already been cancelled...  Maybe a comment along those lines?

[xunjieli, 2015/02/04 18:50:29]

Done.

+    }
+
+    @Override
+    public void onReadSucceeded(boolean lastChunk) {
+        synchronized (mLock) {
+            if (!mReading) {
+                throw new IllegalStateException("Non-existent read succeeded.");
+            }
+            if (lastChunk && mLength >= 0) {
+                throw new IllegalArgumentException(
+                        "Non-chunked upload can't have last chunk");
+            }
+            int position = mByteBuffer.position();
+            int limit = mByteBuffer.limit();
+
+            mByteBuffer = null;
+            mReading = false;
+
+            destroyAdapterIfPostponed();
+            // Request may been canceled already.
+            if (mUploadDataStreamDelegate == 0) {
+                return;
+            }
+            nativeOnReadSucceeded(mUploadDataStreamDelegate, position, limit,
+                    lastChunk);
+        }
+    }
+
+    public void onReadError(Exception exception) {
+        synchronized (mLock) {
+            if (!mReading) {
+                throw new IllegalStateException("Non-existent read failed.");
+            }
+            // Request may been canceled already.
+            if (mUploadDataStreamDelegate == 0) {
+                return;
+            }
+            onError(exception);
+        }
+    }
+
+    public void onRewindSucceeded() {
+        synchronized (mLock) {
+            if (!mRewinding) {
+                throw new IllegalStateException(
+                        "Non-existent rewind succeeded.");
+            }
+            mRewinding = false;
+            // Request may been canceled already.
+            if (mUploadDataStreamDelegate == 0) {
+                return;
+            }
+            nativeOnRewindSucceeded(mUploadDataStreamDelegate);
+        }
+    }
+
+    public void onRewindError(Exception exception) {
+        synchronized (mLock) {
+            if (!mRewinding) {
+                throw new IllegalStateException("Non-existent rewind failed.");
+            }
+            // Request may been canceled already.
+            if (mUploadDataStreamDelegate == 0) {
+                return;
+            }
+            onError(exception);
+        }
+    }
+
+    /**
+     * The adapter is owned by the CronetUploadDataStream, so it can be
+     * destroyed safely when there is no pending read; however, destruction is
+     * initiated by the destruction of the native UploadDataStream.
+     */
+    private void destroyAdapter() {
+        synchronized (mLock) {
+            if (mReading) {
+                // Wait for the read to complete before destroy the adapter.
+                mDestroyAdapterPostponed = true;

[mmenke, 2015/02/04 16:46:00]

Should probably have a test for this - using your test fixture, start an upload,
wait for the read to be pending, and then during the read, delete the C++
object.  Then make the read complete.

The test is racy, in that the read could complete either before or after
destroyAdapter() is called.  It should pass either way, but we're really only
interested in the latter case.  I don't see a way around the race,
unfortunately, without digging into the guts of the UploadDataStream.

+                return;
+            }
+            if (mUploadDataStreamDelegate == 0) {
+                return;
+            }
+            nativeDestroyDelegate(mUploadDataStreamDelegate);
+            mUploadDataStreamDelegate = 0;
+        }
+    }
+
+    /**
+     * Destroy the native adapter if the destruction is postponed due to a
+     * pending read, which has since completed. Caller needs to be on executor
+     * thread and acquire mLock.
+     */
+    private void destroyAdapterIfPostponed() {
+        if (mReading) {
+            throw new IllegalStateException(
+                    "Method should not be called when read has not completed.");
+        }
+        if (mDestroyAdapterPostponed) {
+            destroyAdapter();
+        }
+    }
+
+    /**
+     * Creates native objects and attaches them to the underlying request
+     * adapter object.
+     * TODO(mmenke):  If more types of native upload streams are needed, create
+     * an interface with just this method, to minimize CronetURLRequest's
+     * dependencies on each upload stream type.
+     */
+    void attachToRequest(CronetUrlRequest request, long requestAdapter) {
+        if (mLength < 0) {
+            // TODO(mmenke):  Add tests and remove this line.
+            throw new IllegalArgumentException(
+                    "Chunked uploads not supported.");
+        }
+        mRequest = request;
+        mUploadDataStreamDelegate =
+                nativeAttachUploadDataToRequest(requestAdapter, mLength);
+    }
+
+    /**
+     * Creates a native UploadDataStreamDelegate and UploadDataStreamAdapter
+     * for testing.
+     */
+    public long createAdapter() {
+        mUploadDataStreamDelegate = nativeCreateDelegateForTesting();
+        return nativeCreateAdapterForTesting(mLength, mUploadDataStreamDelegate);
+    }
+
+    // Native methods are implemented in upload_data_stream_adapter.cc.
+
+    private native long nativeAttachUploadDataToRequest(long urlRequestAdapter,
+            long length);
+
+    private native long nativeCreateDelegateForTesting();
+
+    private native long nativeCreateAdapterForTesting(long length,
+            long delegate);
+
+    private native void nativeOnReadSucceeded(long uploadDataStreamDelegate,
+            int position, int limit, boolean finalChunk);
+
+    private native void nativeOnRewindSucceeded(long uploadDataStreamDelegate);
+
+    private static native void nativeDestroyDelegate(
+            long uploadDataStreamDelegate);
+}