Upstream changes to enable chunked uploads.

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

Index: components/cronet/android/java/src/org/chromium/net/ChromiumUrlRequest.java
diff --git a/components/cronet/android/java/src/org/chromium/net/ChromiumUrlRequest.java b/components/cronet/android/java/src/org/chromium/net/ChromiumUrlRequest.java
index 91268c3f072eca2d7e7b89799d18215a4d76dd3b..ae22787434fd208d2e991a069ec1d5bad61e1204 100644
--- a/components/cronet/android/java/src/org/chromium/net/ChromiumUrlRequest.java
+++ b/components/cronet/android/java/src/org/chromium/net/ChromiumUrlRequest.java
@@ -22,6 +22,7 @@ import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
+import java.util.concurrent.Semaphore;
 
 /**
  * Network request using the native http stack implementation.
@@ -41,6 +42,7 @@ public class ChromiumUrlRequest implements HttpUrlRequest {
     private String mUploadContentType;
     private String mMethod;
     private byte[] mUploadData;
+    private Semaphore mAppendChunkSemaphore;
     private ReadableByteChannel mUploadChannel;
     private WritableByteChannel mOutputChannel;
     private IOException mSinkException;
@@ -220,6 +222,7 @@ public class ChromiumUrlRequest implements HttpUrlRequest {
             mUploadContentType = contentType;
             mUploadData = data;
             mUploadChannel = null;
+            mAppendChunkSemaphore = null;
         }
     }
 
@@ -240,9 +243,28 @@ public class ChromiumUrlRequest implements HttpUrlRequest {
             mUploadChannel = channel;
             mUploadContentLength = contentLength;
             mUploadData = null;
+            mAppendChunkSemaphore = null;
         }
     }
 
+   /**
+     * Creates writable byte channel for chunked uploads as part of a POST or
+     * PUT request.
+     *
+     * @param contentType MIME type of the upload content.
+     */
+    public WritableByteChannel createChunkedUploadChannel(String contentType) {
+        synchronized (mLock) {
+            validateNotStarted();
+            mUploadContentType = contentType;
+            mUploadChannel = null;
+            mUploadContentLength = 0;
+            mUploadData = null;
+            mAppendChunkSemaphore = new Semaphore(0);
+        }
+        return new UploadChannel();
+    }
+
     /**
      * Sets HTTP method for upload request. Only PUT or POST are allowed.
      */
@@ -303,6 +325,9 @@ public class ChromiumUrlRequest implements HttpUrlRequest {
             } else if (mUploadChannel != null) {
                 nativeSetUploadChannel(mUrlRequestAdapter, mUploadContentType,
                                        mUploadContentLength);
+            } else if (mUploadContentType != null) {
+                nativeEnableChunkedUpload(mUrlRequestAdapter,
+                                          mUploadContentType);
             }
 
             nativeStart(mUrlRequestAdapter);
@@ -378,6 +403,40 @@ public class ChromiumUrlRequest implements HttpUrlRequest {
         cancel();
     }
 
+     /**
+       * Invokes {@link #nativeAppendChunk(long, ByteBuffer, int, boolean)} and
+       * waits for it to notify completion.
+       * @param chunk The data. It's position must be zero.
+       * @param isLastChunk Whether chunk is the last one.
+       */
+    void appendChunkBlocking(ByteBuffer chunk, boolean isLastChunk)
+            throws IOException {
+        if (chunk.position() != 0) {
+            throw new IllegalArgumentException("The position must be zero.");
+        }
+        synchronized (mLock) {
+            if (mUrlRequestAdapter == 0) {
+                throw new IOException("Native adapter is destroyed.");
+            }
+            nativeAppendChunkToUpload(mUrlRequestAdapter, chunk, chunk.limit(),
+                                      false);
+            // Wait for the data to be actually consumed.
+            try {
+                mAppendChunkSemaphore.acquire();
+            } catch (InterruptedException e) {
+                // We were interrupted before the data was uploaded. Recovering
+                // from this state is complicated so we cancel the upload
+                // operation and fail.
+                Thread.currentThread().interrupt();
+
+                // TODO(miloslav): Not sure why do we set mSinkException here.
+                mSinkException = new IOException("Upload interrupted", e);
+                cancel();
+                throw mSinkException;
+            }
+        }
+    }
+
     /**
      * A callback invoked when the response has been fully consumed.
      */
@@ -403,6 +462,68 @@ public class ChromiumUrlRequest implements HttpUrlRequest {
         }
     }
 
+    class UploadChannel implements WritableByteChannel {
+        private static final int UPLOAD_BYTE_BUFFER_SIZE = 32768;
+
+        private boolean mOpen = true;
+        // Native wants a direct buffer.
+        private final ByteBuffer mBuffer =
+                ByteBuffer.allocateDirect(UPLOAD_BYTE_BUFFER_SIZE);
+
+        @Override
+        public synchronized boolean isOpen() {
+            return mOpen;
+        }
+
+        @Override
+        public synchronized void close() throws IOException {
+            Log.d(ChromiumUrlRequestContext.LOG_TAG,
+                    "UploadChannel.close() url=" + getUrl());
+            if (!mOpen) {
+                return;
+            }
+
+            mOpen = false;
+            mBuffer.clear();
+
+            // NOOP If the native peer has been destroyed.
+            try {
+                Log.d(ChromiumUrlRequestContext.LOG_TAG,
+                        "UploadChannel.close(): final chunk.");
+                appendChunkBlocking(mBuffer, true);
+            } catch (IOException e) {
+                Log.w(ChromiumUrlRequestContext.LOG_TAG,
+                        "Ignoring exception during closing.", e);
+            }
+            Log.d(ChromiumUrlRequestContext.LOG_TAG,
+                    "UploadChannel.close() done.");
+        }
+
+        @Override
+        public synchronized int write(ByteBuffer sourceBuffer)
+                throws IOException {
+            Log.d(ChromiumUrlRequestContext.LOG_TAG, "UploadChannel.write(" +
+                    sourceBuffer.remaining() + " bytes) url=" + getUrl());
+            int written = 0;
+            while (sourceBuffer.hasRemaining()) {
+                mBuffer.clear();
+                int oldLimit = sourceBuffer.limit();
+                if (sourceBuffer.remaining() > mBuffer.remaining()) {
+                    sourceBuffer.limit(sourceBuffer.position()
+                            + mBuffer.remaining());
+                }
+                mBuffer.put(sourceBuffer);
+                mBuffer.flip();
+                written += mBuffer.limit();
+                appendChunkBlocking(mBuffer, false);
+                sourceBuffer.limit(oldLimit);
+            }
+            Log.d(ChromiumUrlRequestContext.LOG_TAG,
+                    "UploadChannel.write() returning");
+            return written;
+        }
+    }
+
     // Private methods called by native library.
 
     /**
@@ -423,6 +544,14 @@ public class ChromiumUrlRequest implements HttpUrlRequest {
     }
 
     /**
+     * A callback invoked when appending a chunk to the request has completed.
+     */
+    @CalledByNative
+    protected void onAppendChunkCompleted() {
+        mAppendChunkSemaphore.release();
+    }
+
+    /**
      * A callback invoked when the first chunk of the response has arrived.
      */
     @CalledByNative
@@ -605,6 +734,12 @@ public class ChromiumUrlRequest implements HttpUrlRequest {
     private native void nativeSetUploadChannel(long urlRequestAdapter,
             String contentType, long contentLength);
 
+    private native void nativeEnableChunkedUpload(long urlRequestAdapter,
+            String contentType);
+
+    private native void nativeAppendChunkToUpload(long urlRequestAdapter,
+            ByteBuffer chunk, int chunkSize, boolean isLastChunk);
+
     private native void nativeStart(long urlRequestAdapter);
 
     private native void nativeCancel(long urlRequestAdapter);