[Cronet] Implement getOutputStream in CronetHttpURLConnection

components/cronet/android/java/src/org/chromium/net/urlconnection/CronetOutputStream.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.urlconnection;
+
+import org.chromium.base.VisibleForTesting;
+import org.chromium.net.UploadDataProvider;
+import org.chromium.net.UploadDataSink;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.nio.ByteBuffer;
+
+/**
+ * An implementation of {@link java.io.OutputStream} to send data to a server,
+ * when {@link CronetHttpURLConnection#setFixedLengthStreamingMode} is used.
+ * This implementation does not buffer the entire request body in memory.
+ * Note that {@link #write} should only be called from the thread on which the
+ * {@link #mConnection} is created.
+ */
+final class CronetOutputStream extends OutputStream implements UploadDataProvider {
+    // CronetOutputStream buffers up to this value and wait for UploadDataStream
+    // to consume the data. This field is non-final, so it can be changed for tests.
+    // TODO(xunjieli): figure out whether this default value should be changed.
+    @VisibleForTesting

[mmenke, 2015/03/13 14:51:12]

What does this do?  I don't see any of our tests using the value below, and
since it's private, I assume it can't be accessed.

[xunjieli, 2015/03/13 19:26:40]

The docs on this flag says "Annotation used to mark code that has wider
visibility or present for testing code", I put it here to make it clear why this
static field isn't final. There is a public setter for this variable. This is to
test the "byteBuffer.remaining() < mBuffer.position()" case below (unit test is
in CronetOutputStreamTest.java). 

+    private static int sDefaultBufferLength = 2048;

[mmenke, 2015/03/13 14:51:12]

This should probably be 32k

[xunjieli, 2015/03/13 19:26:40]

But 32k is larger than all the read buffers used in QUIC(14520), SPDY(2852) and
normal stream(16384). Why do we use 32k? Should we use 16384 (1 << 14) instead?

[mmenke, 2015/03/25 18:34:57]

You're right.  We generally read from URLRequests with a 32k buffer, but use a
smaller buffer for read data to be uploaded.  That seems a little weird.  Wonder
if it's worth testing upload buffer sizes, C++-side.

+    private final CronetHttpURLConnection mConnection;
+    private final long mContentLength;
+    private final ByteBuffer mBuffer;
+    private long mBytesReceived;
+
+    /**
+     * Packaged protected constructor.
+     * @param connection The CronetHttpURLConnection object.
+     * @param contentLength The content length of the request body. Non-zero for
+     *            non-chunked upload.
+     */
+    CronetOutputStream(CronetHttpURLConnection connection, long contentLength) {
+        if (connection == null) {
+            throw new NullPointerException();
+        }
+        if (contentLength < 0) {
+            throw new IllegalArgumentException(
+                    "Content length must be larger than 0 for non-chunked upload.");
+        }
+        mContentLength = contentLength;
+        int bufferSize = (int) Math.min(mContentLength, sDefaultBufferLength);
+        mBuffer = ByteBuffer.allocate(bufferSize);
+        mConnection = connection;
+        mBytesReceived = 0;
+    }
+
+    @Override
+    public void write(int oneByte) throws IOException {
+        while (mBuffer.position() == mBuffer.limit()) {
+            // Wait until buffer is consumed.
+            mConnection.waitForRead();
+        }
+        mBuffer.put((byte) oneByte);
+        mBytesReceived++;
+        if (mBytesReceived == mContentLength) {
+            // Entire post data has been received. Now wait for network stack to
+            // read it.
+            mConnection.waitForRead();
+        }
+    }
+
+    @Override
+    public void write(byte[] buffer, int offset, int count) throws IOException {
+        if (buffer.length - offset < count || offset < 0 || count < 0) {
+            throw new IndexOutOfBoundsException();
+        }
+        if (count == 0) {
+            return;
+        }
+        int toSend = count;
+        while (toSend > 0) {
+            if (mBuffer.position() == mBuffer.limit()) {
+                // Wait until buffer is consumed.
+                mConnection.waitForRead();
+            }
+            int sent = Math.min(toSend, mBuffer.limit() - mBuffer.position());
+            mBuffer.put(buffer, offset + count - toSend, sent);
+            toSend -= sent;
+        }
+        mBytesReceived += count;
+        if (mBytesReceived == mContentLength) {
+            // Entire post data has been received. Now wait for network stack to
+            // read it.
+            mConnection.waitForRead();
+        }
+    }
+
+    // Below are UploadDataProvider implementations. Only intended to be used
+    // within Cronet.
+
+    @Override
+    public long getLength() {
+        return mContentLength;
+    }
+
+    @Override
+    public void read(final UploadDataSink uploadDataSink, final ByteBuffer byteBuffer) {
+        Runnable readTask = new Runnable() {
+            @Override
+            public void run() {
+                if (byteBuffer.remaining() < mBuffer.position()) {
+                    // byteBuffer does not have enough capacity, so only put a portion
+                    // of mBuffer in it.
+                    mBuffer.position(byteBuffer.remaining());
+                    byteBuffer.put(mBuffer.array(), 0, byteBuffer.remaining());
+                    // Move remaining buffer to the head of the buffer for use in the
+                    // next read call.
+                    mBuffer.compact();
+                    uploadDataSink.onReadSucceeded(false);
+                } else {
+                    // byteBuffer has enough capacity to hold content in mBuffer.
+                    mBuffer.flip();
+                    byteBuffer.put(mBuffer);
+                    // Reuse this buffer.
+                    mBuffer.clear();
+                    uploadDataSink.onReadSucceeded(false);
+                }
+            }
+        };
+        if (mBuffer.position() == 0) {
+            // Exit the message loop, so consumer can write more data.
+            if (mBytesReceived < mContentLength) {
+                mConnection.postponeRead(readTask);
+            }
+        } else {
+            readTask.run();
+        }
+    }
+
+    @Override
+    public void rewind(UploadDataSink uploadDataSink) {
+        uploadDataSink.onRewindError(null);
+    }
+
+    /**
+     * Sets the default buffer length for use in tests.
+     */
+    @VisibleForTesting
+    static void setDefaultBufferLengthForTesting(int length) {
+        sDefaultBufferLength = length;
+    }
+}