[Cronet] Fix BidirectionalStream.flush()

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

Index: components/cronet/android/java/src/org/chromium/net/CronetBidirectionalStream.java
diff --git a/components/cronet/android/java/src/org/chromium/net/CronetBidirectionalStream.java b/components/cronet/android/java/src/org/chromium/net/CronetBidirectionalStream.java
index 5e943043e1d490592c137c0a7cccb463a0ff1b4c..06773b2b0dbd60b6c0e671544408807a78494e02 100644
--- a/components/cronet/android/java/src/org/chromium/net/CronetBidirectionalStream.java
+++ b/components/cronet/android/java/src/org/chromium/net/CronetBidirectionalStream.java
@@ -343,6 +343,14 @@ class CronetBidirectionalStream extends BidirectionalStream {
             // 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];
@@ -357,7 +365,8 @@ class CronetBidirectionalStream extends BidirectionalStream {
         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;
@@ -365,6 +374,34 @@ class CronetBidirectionalStream extends BidirectionalStream {
         }
     }
 
+    /**
+     * 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.
@@ -515,7 +552,7 @@ class CronetBidirectionalStream extends BidirectionalStream {
             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++) {
@@ -527,7 +564,9 @@ class CronetBidirectionalStream extends BidirectionalStream {
             }
             // 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));
         }
     }