[Cronet] Fix BidirectionalStream.flush()

components/cronet/android/java/src/org/chromium/net/CronetBidirectionalStream.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.Log;
 import org.chromium.base.VisibleForTesting;
 import org.chromium.base.annotations.CalledByNative;
 import org.chromium.base.annotations.JNINamespace;
@@ skipping 325 matching lines @@
         if (!mPendingData.isEmpty()) {
             mFlushData.addAll(mPendingData);
             mPendingData.clear();
         }
 
         if (mWriteState == State.WRITING) {
             // If there is a write already pending, wait until onWritevCompleted is
             // called before pushing data to the native stack.
             return;
         }
+        sendFlushDataLocked();
+    }
+
+    // Helper method to send buffers in mFlushData. Caller needs to acquire
+    // mNativeStreamLock and make sure mWriteState is WAITING_FOR_FLUSH and
+    // mFlushData queue isn't empty.
+    @SuppressWarnings("GuardedByChecker")
+    private void sendFlushDataLocked() {
         assert mWriteState == State.WAITING_FOR_FLUSH;
         int size = mFlushData.size();
         ByteBuffer[] buffers = new ByteBuffer[size];
         int[] positions = new int[size];
         int[] limits = new int[size];
         for (int i = 0; i < size; i++) {
             ByteBuffer buffer = mFlushData.poll();
             buffers[i] = buffer;
             positions[i] = buffer.position();
             limits[i] = buffer.limit();
         }
         assert mFlushData.isEmpty();
         assert buffers.length >= 1;
         mWriteState = State.WRITING;
-        if (!nativeWritevData(mNativeStream, buffers, positions, limits, mEndOfStreamWritten)) {
+        if (!nativeWritevData(mNativeStream, buffers, positions, limits,
+                    mEndOfStreamWritten && mPendingData.isEmpty())) {
             // Still waiting on flush. This is just to have consistent
             // behavior with the other error cases.
             mWriteState = State.WAITING_FOR_FLUSH;
             throw new IllegalArgumentException("Unable to call native writev.");
         }
     }
 
+    /**
+     * Returns a read-only copy of {@code mPendingData} for testing.
+     */
+    @VisibleForTesting
+    public List<ByteBuffer> getPendingDataForTesting() {
+        synchronized (mNativeStreamLock) {
+            List<ByteBuffer> pendingData = new LinkedList<ByteBuffer>();
+            for (ByteBuffer buffer : mPendingData) {
+                pendingData.add(buffer.asReadOnlyBuffer());
+            }
+            return pendingData;
+        }
+    }
+
+    /**
+     * Returns a read-only copy of {@code mFlushData} for testing.
+     */
+    @VisibleForTesting
+    public List<ByteBuffer> getFlushDataForTesting() {
+        synchronized (mNativeStreamLock) {
+            List<ByteBuffer> flushData = new LinkedList<ByteBuffer>();
+            for (ByteBuffer buffer : mFlushData) {
+                flushData.add(buffer.asReadOnlyBuffer());
+            }
+            return flushData;
+        }
+    }
+
     @Override
     public void ping(PingCallback callback, Executor executor) {
         // TODO(mef): May be last thing to be implemented on Android.
         throw new UnsupportedOperationException("ping is not supported yet.");
     }
 
     @Override
     public void windowUpdate(int windowSizeIncrement) {
         // TODO(mef): Understand the needs and semantics of this method.
         throw new UnsupportedOperationException("windowUpdate is not supported yet.");
@@ skipping 130 matching lines @@
     @SuppressWarnings("unused")
     @CalledByNative
     private void onWritevCompleted(final ByteBuffer[] byteBuffers, int[] initialPositions,
             int[] initialLimits, boolean endOfStream) {
         assert byteBuffers.length == initialPositions.length;
         assert byteBuffers.length == initialLimits.length;
         synchronized (mNativeStreamLock) {
             mWriteState = State.WAITING_FOR_FLUSH;
             // Flush if there is anything in the flush queue mFlushData.
             if (!mFlushData.isEmpty()) {
-                flushLocked();
+                sendFlushDataLocked();
             }
         }
         for (int i = 0; i < byteBuffers.length; i++) {
             ByteBuffer buffer = byteBuffers[i];
             if (buffer.position() != initialPositions[i] || buffer.limit() != initialLimits[i]) {
                 failWithException(
                         new CronetException("ByteBuffer modified externally during write", null));
                 return;
             }
             // Current implementation always writes the complete buffer.
             buffer.position(buffer.limit());
-            postTaskToExecutor(new OnWriteCompletedRunnable(buffer, endOfStream));
+            postTaskToExecutor(new OnWriteCompletedRunnable(buffer,
+                    // Only set endOfStream flag if this buffer is the last in byteBuffers.
+                    endOfStream && i == byteBuffers.length - 1));
         }
     }
 
     @SuppressWarnings("unused")
     @CalledByNative
     private void onResponseTrailersReceived(String[] trailers) {
         final UrlResponseInfo.HeaderBlock trailersBlock =
                 new UrlResponseInfo.HeaderBlock(headersListFromStrings(trailers));
         postTaskToExecutor(new Runnable() {
             public void run() {
@@ skipping 185 matching lines @@
     private native boolean nativeReadData(
             long nativePtr, ByteBuffer byteBuffer, int position, int limit);
 
     @NativeClassQualifiedName("CronetBidirectionalStreamAdapter")
     private native boolean nativeWritevData(long nativePtr, ByteBuffer[] buffers, int[] positions,
             int[] limits, boolean endOfStream);
 
     @NativeClassQualifiedName("CronetBidirectionalStreamAdapter")
     private native void nativeDestroy(long nativePtr, boolean sendOnCanceled);
 }