Upstream changes to enable chunked uploads.

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

 // 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.
 
 package org.chromium.net;
 
 import android.util.Log;
 
 import org.apache.http.conn.ConnectTimeoutException;
 import org.chromium.base.CalledByNative;
 import org.chromium.base.JNINamespace;
 
 import java.io.IOException;
 import java.net.MalformedURLException;
 import java.net.URL;
 import java.net.UnknownHostException;
 import java.nio.ByteBuffer;
 import java.nio.channels.ReadableByteChannel;
 import java.nio.channels.WritableByteChannel;
 import java.util.ArrayList;
 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.
  */
 @JNINamespace("cronet")
 public class ChromiumUrlRequest implements HttpUrlRequest {
     /**
      * Native adapter object, owned by UrlRequest.
      */
     private long mUrlRequestAdapter;
     private final ChromiumUrlRequestContext mRequestContext;
     private final String mUrl;
     private final int mPriority;
     private final Map<String, String> mHeaders;
     private final WritableByteChannel mSink;
     private Map<String, String> mAdditionalHeaders;
     private String mUploadContentType;
     private String mMethod;
     private byte[] mUploadData;
+    private Semaphore mAppendChunkSemaphore;
     private ReadableByteChannel mUploadChannel;
     private WritableByteChannel mOutputChannel;
     private IOException mSinkException;
     private volatile boolean mStarted;
     private volatile boolean mCanceled;
     private volatile boolean mRecycled;
     private volatile boolean mFinished;
     private boolean mHeadersAvailable;
     private String mContentType;
     private long mUploadContentLength;
@@ skipping 159 matching lines @@
      * @param contentType MIME type of the upload content or null if this is not
      *            an upload.
      * @param data The content that needs to be uploaded.
      */
     public void setUploadData(String contentType, byte[] data) {
         synchronized (mLock) {
             validateNotStarted();
             mUploadContentType = contentType;
             mUploadData = data;
             mUploadChannel = null;
+            mAppendChunkSemaphore = null;
         }
     }
 
     /**
      * Sets a readable byte channel to upload as part of a POST or PUT request.
      *
      * @param contentType MIME type of the upload content or null if this is not
      *            an upload request.
      * @param channel The channel to read to read upload data from if this is an
      *            upload request.
      * @param contentLength The length of data to upload.
      */
     public void setUploadChannel(String contentType,
             ReadableByteChannel channel, long contentLength) {
         synchronized (mLock) {
             validateNotStarted();
             mUploadContentType = contentType;
             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.
      */
     public void setHttpMethod(String method) {
         validateNotStarted();
         if (!("PUT".equals(method) || "POST".equals(method))) {
             throw new IllegalArgumentException("Only PUT or POST are allowed.");
         }
         mMethod = method;
     }
@@ skipping 40 matching lines @@
                             entry.getValue());
                 }
             }
 
             if (mUploadData != null && mUploadData.length > 0) {
                 nativeSetUploadData(mUrlRequestAdapter, mUploadContentType,
                                     mUploadData);
             } else if (mUploadChannel != null) {
                 nativeSetUploadChannel(mUrlRequestAdapter, mUploadContentType,
                                        mUploadContentLength);
+            } else if (mUploadContentType != null) {
+                nativeEnableChunkedUpload(mUrlRequestAdapter,
+                                          mUploadContentType);
             }
 
             nativeStart(mUrlRequestAdapter);
           }
     }
 
     public void cancel() {
         synchronized (mLock) {
             if (mCanceled) {
                 return;
@@ skipping 55 matching lines @@
             default:
                 return ChromiumUrlRequestPriority.MEDIUM;
         }
     }
 
     private void onContentLengthOverLimit() {
         mContentLengthOverLimit = true;
         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.
      */
     private void onRequestComplete() {
         mListener.onRequestComplete(this);
     }
 
     private void validateNotRecycled() {
         if (mRecycled) {
             throw new IllegalStateException("Accessing recycled request");
         }
     }
 
     private void validateNotStarted() {
         if (mStarted) {
             throw new IllegalStateException("Request already started");
         }
     }
 
     private void validateHeadersAvailable() {
         if (!mHeadersAvailable) {
             throw new IllegalStateException("Response headers not available");
         }
     }
 
+    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.
 
     /**
      * If @CalledByNative method throws an exception, request gets cancelled
      * and exception could be retrieved from using getException().
      */
     private void onCalledByNativeException(Exception e) {
         mSinkException = new IOException(
                 "CalledByNative method has thrown an exception", e);
         Log.e(ChromiumUrlRequestContext.LOG_TAG,
                 "Exception in CalledByNative method", e);
         try {
             cancel();
         } catch (Exception cancel_exception) {
             Log.e(ChromiumUrlRequestContext.LOG_TAG,
                     "Exception trying to cancel request", cancel_exception);
         }
     }
 
     /**
+     * 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
     private void onResponseStarted() {
         try {
             mContentType = nativeGetContentType(mUrlRequestAdapter);
             mContentLength = nativeGetContentLength(mUrlRequestAdapter);
             mHeadersAvailable = true;
 
             if (mContentLengthLimit > 0 &&
@@ skipping 162 matching lines @@
             String value);
 
     private native void nativeSetMethod(long urlRequestAdapter, String method);
 
     private native void nativeSetUploadData(long urlRequestAdapter,
             String contentType, byte[] content);
 
     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);
 
     private native void nativeDestroyRequestAdapter(long urlRequestAdapter);
 
     private native int nativeGetErrorCode(long urlRequestAdapter);
 
     private native int nativeGetHttpStatusCode(long urlRequestAdapter);
 
     private native String nativeGetErrorString(long urlRequestAdapter);
 
     private native String nativeGetContentType(long urlRequestAdapter);
 
     private native long nativeGetContentLength(long urlRequestAdapter);
 
     private native String nativeGetHeader(long urlRequestAdapter, String name);
 
     private native void nativeGetAllHeaders(long urlRequestAdapter,
             ResponseHeadersMap headers);
 
     // Explicit class to work around JNI-generator generics confusion.
     private class ResponseHeadersMap extends HashMap<String, List<String>> {
     }
 }