Initial implementation of CronetBidirectionalStream.

components/cronet/android/test/javatests/src/org/chromium/net/TestBidirectionalStreamCallback.java

-// Copyright 2014 The Chromium Authors. All rights reserved.
+// 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 android.os.ConditionVariable;
 
 import static junit.framework.Assert.assertEquals;
 import static junit.framework.Assert.assertFalse;
 import static junit.framework.Assert.assertNull;
 import static junit.framework.Assert.assertTrue;
 
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
 import java.util.concurrent.Executor;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ThreadFactory;
 
 /**
  * Callback that tracks information from different callbacks and and has a
- * method to block thread until the request completes on another thread.
- * Allows to cancel, block request or throw an exception from an arbitrary step.
+ * method to block thread until the stream completes on another thread.
+ * Allows to cancel, block stream or throw an exception from an arbitrary step.
  */
-class TestUrlRequestCallback extends UrlRequest.Callback {
-    public ArrayList<UrlResponseInfo> mRedirectResponseInfoList = new ArrayList<UrlResponseInfo>();
-    public ArrayList<String> mRedirectUrlList = new ArrayList<String>();
+class TestBidirectionalStreamCallback extends BidirectionalStream.Callback {
     public UrlResponseInfo mResponseInfo;
-    public UrlRequestException mError;
+    public CronetException mError;
 
     public ResponseStep mResponseStep = ResponseStep.NOTHING;
 
-    public int mRedirectCount = 0;
     public boolean mOnErrorCalled = false;
     public boolean mOnCanceledCalled = false;
 
     public int mHttpResponseDataLength = 0;
     public String mResponseAsString = "";
 
-    // Expect legacy read() API to be used on UrlRequest.
-    // TODO(pauljensen): Remove when all callers of UrlRequest.read() are
-    // transitioned to UrlRequest.readNew();
-    public boolean mLegacyReadByteBufferAdjustment = false;
+    public UrlResponseInfo.HeaderBlock mTrailers;
 
     private static final int READ_BUFFER_SIZE = 32 * 1024;
 
     // When false, the consumer is responsible for all calls into the request

[xunjieli, 2016/01/22 16:12:19]

nit: s/request/stream

[mef, 2016/01/22 17:36:07]

Done.

     // that advance it.
     private boolean mAutoAdvance = true;
 
     // Conditionally fail on certain steps.
     private FailureType mFailureType = FailureType.NONE;
     private ResponseStep mFailureStep = ResponseStep.NOTHING;
 
     // Signals when request is done either successfully or not.

[xunjieli, 2016/01/22 16:12:19]

nit: s/request/stream

[mef, 2016/01/22 17:36:07]

Done.

     private final ConditionVariable mDone = new ConditionVariable();
 
     // Signaled on each step when mAutoAdvance is false.
-    private final ConditionVariable mStepBlock = new ConditionVariable();
+    private final ConditionVariable mReadStepBlock = new ConditionVariable();
+    private final ConditionVariable mWriteStepBlock = new ConditionVariable();
 
     // Executor Service for Cronet callbacks.
     private final ExecutorService mExecutorService =
             Executors.newSingleThreadExecutor(new ExecutorThreadFactory());
     private Thread mExecutorThread;
 
-    // position() of ByteBuffer prior to readNew() call.
+    // position() of ByteBuffer prior to read() call.
     private int mBufferPositionBeforeRead;
 
+    // Data to write.
+    private ArrayList<ByteBuffer> mWriteBuffers = new ArrayList<ByteBuffer>();
+
     private class ExecutorThreadFactory implements ThreadFactory {
         public Thread newThread(Runnable r) {
             mExecutorThread = new Thread(r);
             return mExecutorThread;
         }
     }
 
     public enum ResponseStep {
         NOTHING,
-        ON_RECEIVED_REDIRECT,
+        ON_REQUEST_HEADERS_SENT,
         ON_RESPONSE_STARTED,
         ON_READ_COMPLETED,
+        ON_WRITE_COMPLETED,
+        ON_TRAILERS,
+        ON_CANCELED,
+        ON_FAILED,
         ON_SUCCEEDED
     }
 
     public enum FailureType {
         NONE,
         CANCEL_SYNC,
         CANCEL_ASYNC,
         // Same as above, but continues to advance the request after posting

[xunjieli, 2016/01/22 16:12:19]

nit: s/request/stream

[mef, 2016/01/22 17:36:07]

Done.

         // the cancellation task.
         CANCEL_ASYNC_WITHOUT_PAUSE,
         THROW_SYNC
     }
 
     public void setAutoAdvance(boolean autoAdvance) {
         mAutoAdvance = autoAdvance;
     }
 
     public void setFailure(FailureType failureType, ResponseStep failureStep) {
         mFailureStep = failureStep;
         mFailureType = failureType;
     }
 
     public void blockForDone() {
         mDone.block();
     }
 
-    public void waitForNextStep() {
-        mStepBlock.block();
-        mStepBlock.close();
+    public void waitForNextReadStep() {
+        mReadStepBlock.block();
+        mReadStepBlock.close();
+    }
+
+    public void waitForNextWriteStep() {
+        mWriteStepBlock.block();
+        mWriteStepBlock.close();
     }
 
     public Executor getExecutor() {
         return mExecutorService;
     }
 
     public void shutdownExecutor() {
         mExecutorService.shutdown();
     }
 
-    @Override
-    public void onRedirectReceived(
-            UrlRequest request, UrlResponseInfo info, String newLocationUrl) {
-        assertEquals(mExecutorThread, Thread.currentThread());
-        assertFalse(request.isDone());
-        assertTrue(mResponseStep == ResponseStep.NOTHING
-                || mResponseStep == ResponseStep.ON_RECEIVED_REDIRECT);
-        assertNull(mError);
-
-        mResponseStep = ResponseStep.ON_RECEIVED_REDIRECT;
-        mRedirectUrlList.add(newLocationUrl);
-        mRedirectResponseInfoList.add(info);
-        ++mRedirectCount;
-        if (maybeThrowCancelOrPause(request)) {
-            return;
-        }
-        request.followRedirect();
+    public void addWriteData(byte[] data) {
+        ByteBuffer writeBuffer = ByteBuffer.allocateDirect(data.length);
+        writeBuffer.put(data);
+        writeBuffer.flip();
+        mWriteBuffers.add(writeBuffer);
     }
 
     @Override
-    public void onResponseStarted(UrlRequest request, UrlResponseInfo info) {
+    public void onRequestHeadersSent(BidirectionalStream stream) {
         assertEquals(mExecutorThread, Thread.currentThread());
-        assertFalse(request.isDone());
+        assertFalse(stream.isDone());
+        assertEquals(ResponseStep.NOTHING, mResponseStep);
+        assertNull(mError);
+
+        mResponseStep = ResponseStep.ON_REQUEST_HEADERS_SENT;
+        if (maybeThrowCancelOrPause(stream, mWriteStepBlock)) {
+            return;
+        }
+        startNextWrite(stream);
+    }
+
+    @Override
+    public void onResponseHeadersReceived(BidirectionalStream stream, UrlResponseInfo info) {
+        assertEquals(mExecutorThread, Thread.currentThread());
+        assertFalse(stream.isDone());
         assertTrue(mResponseStep == ResponseStep.NOTHING
-                || mResponseStep == ResponseStep.ON_RECEIVED_REDIRECT);
+                || mResponseStep == ResponseStep.ON_REQUEST_HEADERS_SENT);
         assertNull(mError);
 
         mResponseStep = ResponseStep.ON_RESPONSE_STARTED;
         mResponseInfo = info;
-        if (maybeThrowCancelOrPause(request)) {
+        if (maybeThrowCancelOrPause(stream, mReadStepBlock)) {
             return;
         }
-        startNextRead(request);
+        startNextRead(stream);
     }
 
     @Override
-    public void onReadCompleted(UrlRequest request, UrlResponseInfo info, ByteBuffer byteBuffer) {
+    public void onReadCompleted(
+            BidirectionalStream stream, UrlResponseInfo info, ByteBuffer byteBuffer) {
         assertEquals(mExecutorThread, Thread.currentThread());
-        assertFalse(request.isDone());
+        assertFalse(stream.isDone());
         assertTrue(mResponseStep == ResponseStep.ON_RESPONSE_STARTED
-                || mResponseStep == ResponseStep.ON_READ_COMPLETED);
+                || mResponseStep == ResponseStep.ON_READ_COMPLETED
+                || mResponseStep == ResponseStep.ON_WRITE_COMPLETED);
         assertNull(mError);
 
         mResponseStep = ResponseStep.ON_READ_COMPLETED;
 
         final byte[] lastDataReceivedAsBytes;
-        if (mLegacyReadByteBufferAdjustment) {
-            // Make a slice of ByteBuffer, so can read from it without affecting
-            // position, which allows tests to check the state of the buffer.
-            ByteBuffer slice = byteBuffer.slice();
-            mHttpResponseDataLength += slice.remaining();
-            lastDataReceivedAsBytes = new byte[slice.remaining()];
-            slice.get(lastDataReceivedAsBytes);
-        } else {
-            final int bytesRead = byteBuffer.position() - mBufferPositionBeforeRead;
-            mHttpResponseDataLength += bytesRead;
-            lastDataReceivedAsBytes = new byte[bytesRead];
-            // Rewind |byteBuffer.position()| to pre-read() position.
-            byteBuffer.position(mBufferPositionBeforeRead);
-            // This restores |byteBuffer.position()| to its value on entrance to
-            // this function.
-            byteBuffer.get(lastDataReceivedAsBytes);
-        }
+        final int bytesRead = byteBuffer.position() - mBufferPositionBeforeRead;
+        mHttpResponseDataLength += bytesRead;
+        lastDataReceivedAsBytes = new byte[bytesRead];
+        // Rewind |byteBuffer.position()| to pre-read() position.
+        byteBuffer.position(mBufferPositionBeforeRead);
+        // This restores |byteBuffer.position()| to its value on entrance to

[xunjieli, 2016/01/22 16:12:19]

nit: maybe remove pipes on 181 and 183.

[mef, 2016/01/22 17:36:07]

Done.

+        // this function.
+        byteBuffer.get(lastDataReceivedAsBytes);
+
         mResponseAsString += new String(lastDataReceivedAsBytes);
 
-        if (maybeThrowCancelOrPause(request)) {
+        if (maybeThrowCancelOrPause(stream, mReadStepBlock)) {
             return;
         }
-        startNextRead(request);
+        startNextRead(stream);
     }
 
     @Override
-    public void onSucceeded(UrlRequest request, UrlResponseInfo info) {
+    public void onWriteCompleted(
+            BidirectionalStream stream, UrlResponseInfo info, ByteBuffer buffer) {
         assertEquals(mExecutorThread, Thread.currentThread());
-        assertTrue(request.isDone());
+        assertFalse(stream.isDone());
+        assertNull(mError);
+        mResponseStep = ResponseStep.ON_WRITE_COMPLETED;
+        if (!mWriteBuffers.isEmpty()) {
+            assertEquals(buffer, mWriteBuffers.get(0));
+            mWriteBuffers.remove(0);
+        }
+        if (maybeThrowCancelOrPause(stream, mWriteStepBlock)) {
+            return;
+        }
+        startNextWrite(stream);
+    }
+
+    @Override
+    public void onResponseTrailersReceived(BidirectionalStream stream, UrlResponseInfo info,
+            UrlResponseInfo.HeaderBlock trailers) {
+        assertEquals(mExecutorThread, Thread.currentThread());
+        assertFalse(stream.isDone());
+        assertNull(mError);
+        mTrailers = trailers;

[xunjieli, 2016/01/22 16:12:19]

need to update mResponseStep.

[mef, 2016/01/22 17:36:07]

Done.

+        if (maybeThrowCancelOrPause(stream, mReadStepBlock)) {
+            return;
+        }
+    }
+
+    @Override
+    public void onSucceeded(BidirectionalStream stream, UrlResponseInfo info) {
+        assertEquals(mExecutorThread, Thread.currentThread());
+        assertTrue(stream.isDone());
         assertTrue(mResponseStep == ResponseStep.ON_RESPONSE_STARTED
-                || mResponseStep == ResponseStep.ON_READ_COMPLETED);
+                || mResponseStep == ResponseStep.ON_READ_COMPLETED
+                || mResponseStep == ResponseStep.ON_WRITE_COMPLETED);
         assertFalse(mOnErrorCalled);
         assertFalse(mOnCanceledCalled);
         assertNull(mError);
 
         mResponseStep = ResponseStep.ON_SUCCEEDED;
         mResponseInfo = info;
         openDone();
-        maybeThrowCancelOrPause(request);
+        maybeThrowCancelOrPause(stream, mReadStepBlock);
     }
 
     @Override
-    public void onFailed(UrlRequest request, UrlResponseInfo info, UrlRequestException error) {
+    public void onFailed(BidirectionalStream stream, UrlResponseInfo info, CronetException error) {
         assertEquals(mExecutorThread, Thread.currentThread());
-        assertTrue(request.isDone());
+        assertTrue(stream.isDone());
         // Shouldn't happen after success.
         assertTrue(mResponseStep != ResponseStep.ON_SUCCEEDED);
         // Should happen at most once for a single request.

[xunjieli, 2016/01/22 16:12:19]

nit: s/request/stream

[mef, 2016/01/22 17:36:07]

Done.

         assertFalse(mOnErrorCalled);
         assertFalse(mOnCanceledCalled);
         assertNull(mError);
+        mResponseStep = ResponseStep.ON_FAILED;
 
         mOnErrorCalled = true;
         mError = error;
         openDone();
-        maybeThrowCancelOrPause(request);
+        maybeThrowCancelOrPause(stream, mReadStepBlock);
     }
 
     @Override
-    public void onCanceled(UrlRequest request, UrlResponseInfo info) {
+    public void onCanceled(BidirectionalStream stream, UrlResponseInfo info) {
         assertEquals(mExecutorThread, Thread.currentThread());
-        assertTrue(request.isDone());
+        assertTrue(stream.isDone());
         // Should happen at most once for a single request.

[xunjieli, 2016/01/22 16:12:19]

nit: s/request/stream

[mef, 2016/01/22 17:36:07]

Done.

         assertFalse(mOnCanceledCalled);
         assertFalse(mOnErrorCalled);
         assertNull(mError);
+        mResponseStep = ResponseStep.ON_CANCELED;
 
         mOnCanceledCalled = true;
         openDone();
-        maybeThrowCancelOrPause(request);
+        maybeThrowCancelOrPause(stream, mReadStepBlock);
     }
 
-    public void startNextRead(UrlRequest request) {
-        startNextRead(request, ByteBuffer.allocateDirect(READ_BUFFER_SIZE));
+    public void startNextRead(BidirectionalStream stream) {
+        startNextRead(stream, ByteBuffer.allocateDirect(READ_BUFFER_SIZE));
     }
 
-    public void startNextRead(UrlRequest request, ByteBuffer buffer) {
+    public void startNextRead(BidirectionalStream stream, ByteBuffer buffer) {
         mBufferPositionBeforeRead = buffer.position();
-        request.readNew(buffer);
+        stream.read(buffer);
+    }
+
+    public void startNextWrite(BidirectionalStream stream) {
+        if (!mWriteBuffers.isEmpty()) {
+            boolean isLastBuffer = mWriteBuffers.size() == 1;
+            stream.write(mWriteBuffers.get(0), isLastBuffer);
+        }
     }
 
     public boolean isDone() {
         // It's not mentioned by the Android docs, but block(0) seems to block
         // indefinitely, so have to block for one millisecond to get state
         // without blocking.
         return mDone.block(1);
     }
 
     protected void openDone() {
         mDone.open();
     }
 
     /**
      * Returns {@code false} if the listener should continue to advance the
      * request.

[xunjieli, 2016/01/22 16:12:19]

nit: s/request/stream

[mef, 2016/01/22 17:36:07]

Done.

      */
-    private boolean maybeThrowCancelOrPause(final UrlRequest request) {
+    private boolean maybeThrowCancelOrPause(
+            final BidirectionalStream stream, ConditionVariable stepBlock) {
         if (mResponseStep != mFailureStep || mFailureType == FailureType.NONE) {
             if (!mAutoAdvance) {
-                mStepBlock.open();
+                stepBlock.open();
                 return true;
             }
             return false;
         }
 
         if (mFailureType == FailureType.THROW_SYNC) {
-            throw new IllegalStateException("Listener Exception.");
+            throw new IllegalStateException("Callback Exception.");
         }
         Runnable task = new Runnable() {
             public void run() {
-                request.cancel();
+                stream.cancel();
             }
         };
         if (mFailureType == FailureType.CANCEL_ASYNC
                 || mFailureType == FailureType.CANCEL_ASYNC_WITHOUT_PAUSE) {
             getExecutor().execute(task);
         } else {
             task.run();
         }
         return mFailureType != FailureType.CANCEL_ASYNC_WITHOUT_PAUSE;
     }
 }