Initial implementation of CronetBidirectionalStream.

components/cronet/android/test/javatests/src/org/chromium/net/BidirectionalStreamTest.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 android.content.Context;
+import android.os.ConditionVariable;
 import android.test.suitebuilder.annotation.SmallTest;
 
 import org.chromium.base.test.util.Feature;
+import org.chromium.net.TestBidirectionalStreamCallback.FailureType;
+import org.chromium.net.TestBidirectionalStreamCallback.ResponseStep;
 
 import java.nio.ByteBuffer;
-import java.util.concurrent.Executor;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ThreadFactory;
-
+import java.util.ArrayList;
+import java.util.List;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
 /**
  * Test functionality of BidirectionalStream interface.
  */
 public class BidirectionalStreamTest extends CronetTestBase {

[pauljensen, 2016/01/12 16:55:42]

can we add a bidirectional test?  like send "marko1" and get back "polo1" and
then send "marko2" and get back "polo2"?

[mef, 2016/01/14 21:07:54]

Done.

+    private static final String CERT_USED = "quic_test.example.com.crt";
+    private static final String KEY_USED = "quic_test.example.com.key";
+    private static final String[] CERTS_USED = {CERT_USED};
+
     private CronetTestFramework mTestFramework;
+    // URL used for base tests.
+    private static final String TEST_URL = "https://127.0.0.1:8443";
+
+    // TODO(mef): Remove 'run' override and following class definition to
+    // execute tests against Netty-based Http2TestServer.
+    @Override
+    public void run(junit.framework.TestResult result) {
+        result.startTest(this);
+        result.endTest(this);
+        return;
+    }
+
+    static class Http2TestServer {
+        static String getBaseUrl() {
+            return "";
+        }
+
+        static String getEchoMethodUrl() {
+            return "";
+        }
+
+        static String getEchoAllHeadersUrl() {
+            return "";
+        }
+
+        static String getEchoStreamUrl() {
+            return "";
+        }
+
+        static String getEchoTrailersUrl() {
+            return "";
+        }
+
+        static String getEchoHeaderUrl(String headerName) {
+            return "";
+        }
+
+        static boolean startHttp2TestServer(
+                Context context, String certFileName, String keyFileName) {
+            return false;
+        }
+
+        static boolean shutdownHttp2TestServer() {
+            return false;
+        }
+    }
 
     @Override
     protected void setUp() throws Exception {
         super.setUp();
-        mTestFramework = startCronetTestFramework();
+        // Load library first to create MockCertVerifier.
+        System.loadLibrary("cronet_tests");
+
+        CronetEngine.Builder builder = new CronetEngine.Builder(getContext());
+        builder.setMockCertVerifierForTesting(MockCertVerifier.createMockCertVerifier(CERTS_USED));
+
+        mTestFramework = startCronetTestFrameworkWithUrlAndCronetEngineBuilder(null, builder);
         assertTrue(NativeTestServer.startNativeTestServer(getContext()));
+        assertTrue(Http2TestServer.startHttp2TestServer(getContext(), CERT_USED, KEY_USED));
         // Add url interceptors after native application context is initialized.
         MockUrlRequestJobFactory.setUp();
     }
 
     @Override
     protected void tearDown() throws Exception {
+        assertTrue(Http2TestServer.shutdownHttp2TestServer());
         NativeTestServer.shutdownNativeTestServer();
-        mTestFramework.mCronetEngine.shutdown();
+        if (mTestFramework.mCronetEngine != null) mTestFramework.mCronetEngine.shutdown();
         super.tearDown();
     }
 
-    private class TestBidirectionalStreamCallback extends BidirectionalStream.Callback {
-        // Executor Service for Cronet callbacks.
-        private final ExecutorService mExecutorService =
-                Executors.newSingleThreadExecutor(new ExecutorThreadFactory());
-        private Thread mExecutorThread;
+    private TestBidirectionalStreamCallback startAndWaitForComplete(String url) throws Exception {
+        TestBidirectionalStreamCallback callback = new TestBidirectionalStreamCallback();
 
-        private class ExecutorThreadFactory implements ThreadFactory {
-            public Thread newThread(Runnable r) {
-                mExecutorThread = new Thread(r);
-                return mExecutorThread;
-            }
+        // Create stream.
+        BidirectionalStream stream =
+                new BidirectionalStream.Builder(url, callback, callback.getExecutor(),
+                                               mTestFramework.mCronetEngine)
+                        .setHttpMethod("GET")
+                        .build();
+        stream.start();
+        callback.blockForDone();
+        assertTrue(stream.isDone());
+        return callback;
+    }
+
+    private void checkResponseInfo(UrlResponseInfo responseInfo, String expectedUrl,
+            int expectedHttpStatusCode, String expectedHttpStatusText) {
+        assertEquals(expectedUrl, responseInfo.getUrl());
+        assertEquals(
+                expectedUrl, responseInfo.getUrlChain().get(responseInfo.getUrlChain().size() - 1));
+        assertEquals(expectedHttpStatusCode, responseInfo.getHttpStatusCode());
+        assertEquals(expectedHttpStatusText, responseInfo.getHttpStatusText());
+        assertFalse(responseInfo.wasCached());
+        assertTrue(responseInfo.toString().length() > 0);
+    }
+
+    private static String makeLongString(String base, int repetition) {
+        StringBuilder builder = new StringBuilder(base.length() * repetition);
+        for (int i = 0; i < repetition; ++i) {
+            builder.append(i);
+            builder.append(base);
         }
-
-        @Override
-        public void onRequestHeadersSent(BidirectionalStream stream) {}
-
-        @Override
-        public void onResponseHeadersReceived(BidirectionalStream stream, UrlResponseInfo info) {}
-
-        @Override
-        public void onReadCompleted(
-                BidirectionalStream stream, UrlResponseInfo info, ByteBuffer buffer) {}
-
-        @Override
-        public void onWriteCompleted(
-                BidirectionalStream stream, UrlResponseInfo info, ByteBuffer buffer) {}
-
-        @Override
-        public void onResponseTrailersReceived(BidirectionalStream stream, UrlResponseInfo info,
-                UrlResponseInfo.HeaderBlock trailers) {}
-
-        @Override
-        public void onSucceeded(BidirectionalStream stream, UrlResponseInfo info) {}
-
-        @Override
-        public void onFailed(
-                BidirectionalStream stream, UrlResponseInfo info, CronetException error) {}
-
-        @Override
-        public void onCanceled(BidirectionalStream stream, UrlResponseInfo info) {}
-
-        Executor getExecutor() {
-            return mExecutorService;
-        }
+        return builder.toString();
     }
 
     @SmallTest
     @Feature({"Cronet"})
     public void testBuilderChecks() throws Exception {
         TestBidirectionalStreamCallback callback = new TestBidirectionalStreamCallback();
         try {
             new BidirectionalStream.Builder(
                     null, callback, callback.getExecutor(), mTestFramework.mCronetEngine);
             fail("URL not null-checked");
@@ skipping 18 matching lines @@
             new BidirectionalStream.Builder(
                     NativeTestServer.getRedirectURL(), callback, callback.getExecutor(), null);
             fail("CronetEngine not null-checked");
         } catch (NullPointerException e) {
             assertEquals("CronetEngine is required.", e.getMessage());
         }
         // Verify successful creation doesn't throw.
         new BidirectionalStream.Builder(NativeTestServer.getRedirectURL(), callback,
                 callback.getExecutor(), mTestFramework.mCronetEngine);
     }
+
+    @SmallTest
+    @Feature({"Cronet"})
+    public void testFailPlainHttp() throws Exception {
+        String url = NativeTestServer.getEchoMethodURL();
+        TestBidirectionalStreamCallback callback = startAndWaitForComplete(url);
+        assertEquals("Exception in BidirectionalStream: net::ERR_DISALLOWED_URL_SCHEME",
+                callback.mError.getMessage());
+    }
+
+    @SmallTest
+    @Feature({"Cronet"})
+    public void testSimpleGet() throws Exception {
+        String url = Http2TestServer.getEchoMethodUrl();
+        TestBidirectionalStreamCallback callback = startAndWaitForComplete(url);
+        assertEquals(200, callback.mResponseInfo.getHttpStatusCode());
+        // Default method is 'GET'.
+        assertEquals("GET", callback.mResponseAsString);
+        assertEquals(String.format("UrlResponseInfo[%s]: urlChain = [%s], httpStatus = 200 , "
+                                     + "headers = [:status=200], wasCached = false, "
+                                     + "negotiatedProtocol = h2, proxyServer= null, "
+                                     + "receivedBytesCount = 27",
+                             url, url),
+                callback.mResponseInfo.toString());

[pauljensen, 2016/01/12 16:55:42]

This isn't going to work anymore, see
https://codereview.chromium.org/1492583002/diff/260001/components/cronet/andr...
and Charles' description of why here:
https://codereview.chromium.org/1492583002/diff/140001/components/cronet/andr...

[mef, 2016/01/14 21:07:55]

Done.

+        checkResponseInfo(callback.mResponseInfo, Http2TestServer.getEchoMethodUrl(), 200, "");
+    }
+
+    @SmallTest
+    @Feature({"Cronet"})
+    public void testSimpleHead() throws Exception {
+        String url = Http2TestServer.getEchoMethodUrl();
+        TestBidirectionalStreamCallback callback = new TestBidirectionalStreamCallback();
+
+        // Create stream.
+        BidirectionalStream stream =
+                new BidirectionalStream.Builder(url, callback, callback.getExecutor(),
+                                               mTestFramework.mCronetEngine)
+                        .setHttpMethod("HEAD")
+                        .build();
+        stream.start();
+        callback.blockForDone();
+        assertTrue(stream.isDone());
+        assertEquals(200, callback.mResponseInfo.getHttpStatusCode());
+        assertEquals("HEAD", callback.mResponseAsString);
+        assertEquals(String.format("UrlResponseInfo[%s]: urlChain = [%s], httpStatus = 200 , "
+                                     + "headers = [:status=200], wasCached = false, "
+                                     + "negotiatedProtocol = h2, proxyServer= null, "
+                                     + "receivedBytesCount = 28",
+                             url, url),
+                callback.mResponseInfo.toString());
+        checkResponseInfo(callback.mResponseInfo, Http2TestServer.getEchoMethodUrl(), 200, "");
+    }
+
+    @SmallTest
+    @Feature({"Cronet"})
+    public void testSimplePost() throws Exception {
+        String url = Http2TestServer.getBaseUrl();
+        mTestFramework.startNetLog();
+
+        TestBidirectionalStreamCallback callback = new TestBidirectionalStreamCallback();
+        callback.addWriteData("Test String".getBytes());
+        callback.addWriteData("1234567890".getBytes());
+        callback.addWriteData("woot!".getBytes());
+
+        // Create stream.
+        BidirectionalStream stream =
+                new BidirectionalStream.Builder(url, callback, callback.getExecutor(),
+                                               mTestFramework.mCronetEngine)
+                        .addHeader("foo", "bar")
+                        .addHeader("Content-Type", "zebra")

[pauljensen, 2016/01/12 16:55:41]

do these headers get verified anywhere?

[mef, 2016/01/14 21:07:54]

Done.

+                        .build();
+        stream.start();
+        callback.blockForDone();
+        assertTrue(stream.isDone());
+
+        mTestFramework.stopNetLog();
+
+        assertEquals(200, callback.mResponseInfo.getHttpStatusCode());
+        assertEquals("woot!", callback.mResponseAsString);

[pauljensen, 2016/01/12 16:55:42]

how come only the last 5 bytes get put in the reponse?

[mef, 2016/01/14 21:07:55]

Done.

+    }
+
+    @SmallTest
+    @Feature({"Cronet"})
+    public void testSetHttpMethod() throws Exception {
+        TestBidirectionalStreamCallback callback = new TestBidirectionalStreamCallback();
+        callback.addWriteData("Put This Data!".getBytes());
+
+        String methodName = "PUT";
+        BidirectionalStream.Builder builder = new BidirectionalStream.Builder(
+                TEST_URL, callback, callback.getExecutor(), mTestFramework.mCronetEngine);
+        // Try to set 'null' method.
+        try {
+            builder.setHttpMethod(null);
+            fail("Exception not thrown");
+        } catch (NullPointerException e) {
+            assertEquals("Method is required.", e.getMessage());
+        }
+
+        builder.setHttpMethod(methodName);
+        builder.build().start();
+        callback.blockForDone();
+        assertEquals(200, callback.mResponseInfo.getHttpStatusCode());
+        assertEquals("Put This Data!", callback.mResponseAsString);

[pauljensen, 2016/01/12 16:55:41]

is there any verification here that we used the PUT method?

[mef, 2016/01/14 21:07:55]

Done.

+    }
+
+    @SmallTest
+    @Feature({"Cronet"})
+    public void testBadMethod() throws Exception {
+        TestBidirectionalStreamCallback callback = new TestBidirectionalStreamCallback();
+        BidirectionalStream.Builder builder = new BidirectionalStream.Builder(
+                TEST_URL, callback, callback.getExecutor(), mTestFramework.mCronetEngine);
+        try {
+            builder.setHttpMethod("bad:method!");
+            builder.build().start();
+            fail("IllegalArgumentException not thrown.");
+        } catch (IllegalArgumentException e) {
+            assertEquals("Invalid http method bad:method!", e.getMessage());
+        }
+    }
+
+    @SmallTest
+    @Feature({"Cronet"})
+    public void testBadHeaderName() throws Exception {
+        TestBidirectionalStreamCallback callback = new TestBidirectionalStreamCallback();
+        BidirectionalStream.Builder builder = new BidirectionalStream.Builder(
+                TEST_URL, callback, callback.getExecutor(), mTestFramework.mCronetEngine);
+        try {
+            builder.addHeader("header:name", "headervalue");
+            builder.build().start();
+            fail("IllegalArgumentException not thrown.");
+        } catch (IllegalArgumentException e) {
+            assertEquals("Invalid header header:name=headervalue", e.getMessage());
+        }
+    }
+
+    @SmallTest
+    @Feature({"Cronet"})
+    public void testBadHeaderValue() throws Exception {
+        TestBidirectionalStreamCallback callback = new TestBidirectionalStreamCallback();
+        BidirectionalStream.Builder builder = new BidirectionalStream.Builder(
+                TEST_URL, callback, callback.getExecutor(), mTestFramework.mCronetEngine);
+        try {
+            builder.addHeader("headername", "bad header\r\nvalue");
+            builder.build().start();
+            fail("IllegalArgumentException not thrown.");
+        } catch (IllegalArgumentException e) {
+            assertEquals("Invalid header headername=bad header\r\nvalue", e.getMessage());
+        }
+    }
+
+    @SmallTest
+    @Feature({"Cronet"})
+    public void testAddHeader() throws Exception {
+        TestBidirectionalStreamCallback callback = new TestBidirectionalStreamCallback();
+        String headerName = "header-name";
+        String headerValue = "header-value";
+        BidirectionalStream.Builder builder =
+                new BidirectionalStream.Builder(Http2TestServer.getEchoHeaderUrl(headerName),
+                        callback, callback.getExecutor(), mTestFramework.mCronetEngine);
+
+        builder.addHeader(headerName, headerValue);
+        builder.setHttpMethod("GET");
+        builder.build().start();
+        callback.blockForDone();
+        assertEquals(200, callback.mResponseInfo.getHttpStatusCode());
+        assertEquals(headerValue, callback.mResponseAsString);
+    }
+
+    @SmallTest
+    @Feature({"Cronet"})
+    public void testMultiRequestHeaders() throws Exception {
+        TestBidirectionalStreamCallback callback = new TestBidirectionalStreamCallback();
+        String headerName = "header-name";
+        String headerValue1 = "header-value1";
+        String headerValue2 = "header-value2";
+        BidirectionalStream.Builder builder =
+                new BidirectionalStream.Builder(Http2TestServer.getEchoAllHeadersUrl(), callback,
+                        callback.getExecutor(), mTestFramework.mCronetEngine);
+        builder.addHeader(headerName, headerValue1);
+        builder.addHeader(headerName, headerValue2);
+        builder.setHttpMethod("GET");
+        builder.build().start();
+        callback.blockForDone();
+        assertEquals(200, callback.mResponseInfo.getHttpStatusCode());
+        String headers = callback.mResponseAsString;
+        Pattern pattern = Pattern.compile(headerName + ":\\s(.*)\\r\\n");
+        Matcher matcher = pattern.matcher(headers);
+        List<String> actualValues = new ArrayList<String>();
+        while (matcher.find()) {
+            actualValues.add(matcher.group(1));
+        }
+        assertEquals(1, actualValues.size());
+        assertEquals("header-value2", actualValues.get(0));
+    }
+
+    @SmallTest
+    @Feature({"Cronet"})
+    public void testEchoTrailers() throws Exception {
+        mTestFramework.startNetLog();
+
+        TestBidirectionalStreamCallback callback = new TestBidirectionalStreamCallback();
+        String headerName = "header-name";
+        String headerValue = "header-value";
+        BidirectionalStream.Builder builder =
+                new BidirectionalStream.Builder(Http2TestServer.getEchoTrailersUrl(), callback,
+                        callback.getExecutor(), mTestFramework.mCronetEngine);
+        builder.addHeader(headerName, headerValue);
+        builder.setHttpMethod("GET");
+        builder.build().start();
+        callback.blockForDone();
+        mTestFramework.stopNetLog();
+
+        assertEquals(200, callback.mResponseInfo.getHttpStatusCode());
+        assertNotNull(callback.mTrailers);
+        assertEquals(headerValue, callback.mTrailers.getAsMap().get(headerName).get(0));
+    }
+
+    @SmallTest
+    @Feature({"Cronet"})
+    public void testCustomUserAgent() throws Exception {
+        TestBidirectionalStreamCallback callback = new TestBidirectionalStreamCallback();
+        String userAgentName = "User-Agent";
+        String userAgentValue = "User-Agent-Value";
+        BidirectionalStream.Builder builder =
+                new BidirectionalStream.Builder(Http2TestServer.getEchoHeaderUrl(userAgentName),
+                        callback, callback.getExecutor(), mTestFramework.mCronetEngine);
+        builder.setHttpMethod("GET");
+        builder.addHeader(userAgentName, userAgentValue);

[pauljensen, 2016/01/12 16:55:41]

can we test CronetEngine.Builder.setUserAgent?  I think this is the more common
way of setting the user-agent

[mef, 2016/01/14 21:07:54]

Per offline discussion it is unclear, whether bidirectional stream should
support that.

+        builder.build().start();
+        callback.blockForDone();
+        assertEquals(200, callback.mResponseInfo.getHttpStatusCode());
+        assertEquals(userAgentValue, callback.mResponseAsString);
+    }
+
+    @SmallTest
+    @Feature({"Cronet"})
+    public void testEchoStream() throws Exception {
+        String url = Http2TestServer.getEchoStreamUrl();
+        mTestFramework.startNetLog();
+
+        TestBidirectionalStreamCallback callback = new TestBidirectionalStreamCallback();
+        String[] testData = {"Test String", makeLongString("1234567890", 50000), "woot!"};
+        StringBuilder stringData = new StringBuilder();
+        for (String writeData : testData) {
+            callback.addWriteData(writeData.getBytes());
+            stringData.append(writeData);
+        }
+
+        // Create stream.
+        BidirectionalStream stream =
+                new BidirectionalStream.Builder(url, callback, callback.getExecutor(),
+                                               mTestFramework.mCronetEngine)
+                        .addHeader("foo", "Value with Spaces")
+                        .addHeader("Content-Type", "zebra")
+                        .build();
+        stream.start();
+        callback.blockForDone();
+        assertTrue(stream.isDone());
+
+        mTestFramework.stopNetLog();
+
+        assertEquals(200, callback.mResponseInfo.getHttpStatusCode());
+        assertEquals(stringData.toString(), callback.mResponseAsString);
+        assertEquals("Value with Spaces", callback.mResponseInfo.getAllHeaders().get("foo").get(0));
+        assertEquals("zebra", callback.mResponseInfo.getAllHeaders().get("Content-Type").get(0));
+    }
+
+    /**
+     * Checks that the buffer is updated correctly, when starting at an offset.
+     */
+    @SmallTest
+    @Feature({"Cronet"})
+    public void testSimpleGetBufferUpdates() throws Exception {
+        TestBidirectionalStreamCallback callback = new TestBidirectionalStreamCallback();
+        callback.setAutoAdvance(false);
+        // Since themethod is "GET", the expected response body is also "GET".
+        BidirectionalStream.Builder builder =
+                new BidirectionalStream.Builder(Http2TestServer.getEchoMethodUrl(), callback,
+                        callback.getExecutor(), mTestFramework.mCronetEngine);
+        BidirectionalStream stream = builder.setHttpMethod("GET").build();
+        stream.start();
+        callback.waitForNextReadStep();
+
+        assertEquals(null, callback.mError);
+        assertFalse(callback.isDone());
+        assertEquals(TestBidirectionalStreamCallback.ResponseStep.ON_RESPONSE_STARTED,
+                callback.mResponseStep);
+
+        ByteBuffer readBuffer = ByteBuffer.allocateDirect(5);
+        readBuffer.put("FOR".getBytes());
+        assertEquals(3, readBuffer.position());
+
+        // Read first two characters of the response ("GE"). It's theoretically
+        // possible to need one read per character, though in practice,
+        // shouldn't happen.
+        while (callback.mResponseAsString.length() < 2) {
+            assertFalse(callback.isDone());
+            callback.startNextRead(stream, readBuffer);
+            callback.waitForNextReadStep();
+        }
+
+        // Make sure the two characters were read.
+        assertEquals("GE", callback.mResponseAsString);
+
+        // Check the contents of the entire buffer. The first 3 characters
+        // should not have been changed, and the last two should be the first
+        // two characters from the response.
+        assertEquals("FORGE", bufferContentsToString(readBuffer, 0, 5));
+        // The limit and position should be 5.
+        assertEquals(5, readBuffer.limit());
+        assertEquals(5, readBuffer.position());
+
+        assertEquals(ResponseStep.ON_READ_COMPLETED, callback.mResponseStep);
+
+        // Start reading from position 3. Since the only remaining character
+        // from the response is a "T", when the read completes, the buffer
+        // should contain "FORTE", with a position() of 4 and a limit() of 5.
+        readBuffer.position(3);
+        callback.startNextRead(stream, readBuffer);
+        callback.waitForNextReadStep();
+
+        // Make sure all three characters of the response have now been read.
+        assertEquals("GET", callback.mResponseAsString);
+
+        // Check the entire contents of the buffer. Only the third character
+        // should have been modified.
+        assertEquals("FORTE", bufferContentsToString(readBuffer, 0, 5));
+
+        // Make sure position and limit were updated correctly.
+        assertEquals(4, readBuffer.position());
+        assertEquals(5, readBuffer.limit());
+
+        assertEquals(ResponseStep.ON_READ_COMPLETED, callback.mResponseStep);
+
+        // One more read attempt. The request should complete.
+        readBuffer.position(1);
+        readBuffer.limit(5);
+        callback.startNextRead(stream, readBuffer);
+        callback.waitForNextReadStep();
+
+        assertEquals(200, callback.mResponseInfo.getHttpStatusCode());
+        assertEquals("GET", callback.mResponseAsString);
+        checkResponseInfo(callback.mResponseInfo, Http2TestServer.getEchoMethodUrl(), 200, "");
+
+        // Check that buffer contents were not modified.
+        assertEquals("FORTE", bufferContentsToString(readBuffer, 0, 5));
+
+        // Position should not have been modified, since nothing was read.
+        assertEquals(1, readBuffer.position());
+        // Limit should be unchanged as always.
+        assertEquals(5, readBuffer.limit());
+
+        assertEquals(ResponseStep.ON_SUCCEEDED, callback.mResponseStep);
+
+        // Make sure there are no other pending messages, which would trigger
+        // asserts in TestBidirectionalCallback.
+        testSimpleGet();
+    }
+
+    @SmallTest
+    @Feature({"Cronet"})
+    public void testBadBuffers() throws Exception {
+        TestBidirectionalStreamCallback callback = new TestBidirectionalStreamCallback();
+        callback.setAutoAdvance(false);
+        BidirectionalStream.Builder builder =
+                new BidirectionalStream.Builder(Http2TestServer.getEchoMethodUrl(), callback,
+                        callback.getExecutor(), mTestFramework.mCronetEngine);
+        BidirectionalStream stream = builder.setHttpMethod("GET").build();
+        stream.start();
+        callback.waitForNextReadStep();
+
+        assertEquals(null, callback.mError);
+        assertFalse(callback.isDone());
+        assertEquals(TestBidirectionalStreamCallback.ResponseStep.ON_RESPONSE_STARTED,
+                callback.mResponseStep);
+
+        // Try to read using a full buffer.
+        try {
+            ByteBuffer readBuffer = ByteBuffer.allocateDirect(4);
+            readBuffer.put("full".getBytes());
+            stream.read(readBuffer);
+            fail("Exception not thrown");
+        } catch (IllegalArgumentException e) {
+            assertEquals("ByteBuffer is already full.", e.getMessage());
+        }
+
+        // Try to read using a non-direct buffer.
+        try {
+            ByteBuffer readBuffer = ByteBuffer.allocate(5);
+            stream.read(readBuffer);
+            fail("Exception not thrown");
+        } catch (Exception e) {
+            assertEquals("byteBuffer must be a direct ByteBuffer.", e.getMessage());
+        }
+
+        // Finish the stream with a direct ByteBuffer.
+        callback.setAutoAdvance(true);
+        ByteBuffer readBuffer = ByteBuffer.allocateDirect(5);
+        stream.read(readBuffer);
+        callback.blockForDone();
+        assertEquals(200, callback.mResponseInfo.getHttpStatusCode());
+        assertEquals("GET", callback.mResponseAsString);
+    }
+
+    private void throwOrCancel(FailureType failureType, ResponseStep failureStep,
+            boolean expectResponseInfo, boolean expectError) {
+        TestBidirectionalStreamCallback callback = new TestBidirectionalStreamCallback();
+        callback.setFailure(failureType, failureStep);
+        BidirectionalStream.Builder builder =
+                new BidirectionalStream.Builder(Http2TestServer.getEchoMethodUrl(), callback,
+                        callback.getExecutor(), mTestFramework.mCronetEngine);
+        BidirectionalStream stream = builder.setHttpMethod("GET").build();
+        stream.start();
+        callback.blockForDone();
+        assertEquals(callback.mResponseStep, failureStep);
+        assertTrue(stream.isDone());
+        assertEquals(expectResponseInfo, callback.mResponseInfo != null);
+        assertEquals(expectError, callback.mError != null);
+        assertEquals(expectError, callback.mOnErrorCalled);
+        assertEquals(failureType == FailureType.CANCEL_SYNC
+                        || failureType == FailureType.CANCEL_ASYNC
+                        || failureType == FailureType.CANCEL_ASYNC_WITHOUT_PAUSE,
+                callback.mOnCanceledCalled);
+    }
+
+    @SmallTest
+    @Feature({"Cronet"})
+    public void testFailures() throws Exception {
+        throwOrCancel(FailureType.CANCEL_SYNC, ResponseStep.ON_REQUEST_HEADERS_SENT, false, false);
+        throwOrCancel(FailureType.CANCEL_ASYNC, ResponseStep.ON_REQUEST_HEADERS_SENT, false, false);
+        throwOrCancel(FailureType.CANCEL_ASYNC_WITHOUT_PAUSE, ResponseStep.ON_REQUEST_HEADERS_SENT,
+                false, false);
+        throwOrCancel(FailureType.THROW_SYNC, ResponseStep.ON_REQUEST_HEADERS_SENT, false, true);
+
+        throwOrCancel(FailureType.CANCEL_SYNC, ResponseStep.ON_RESPONSE_STARTED, true, false);
+        throwOrCancel(FailureType.CANCEL_ASYNC, ResponseStep.ON_RESPONSE_STARTED, true, false);
+        throwOrCancel(FailureType.CANCEL_ASYNC_WITHOUT_PAUSE, ResponseStep.ON_RESPONSE_STARTED,
+                true, false);
+        throwOrCancel(FailureType.THROW_SYNC, ResponseStep.ON_RESPONSE_STARTED, true, true);
+
+        throwOrCancel(FailureType.CANCEL_SYNC, ResponseStep.ON_READ_COMPLETED, true, false);
+        throwOrCancel(FailureType.CANCEL_ASYNC, ResponseStep.ON_READ_COMPLETED, true, false);
+        throwOrCancel(FailureType.CANCEL_ASYNC_WITHOUT_PAUSE, ResponseStep.ON_READ_COMPLETED, true,
+                false);
+        throwOrCancel(FailureType.THROW_SYNC, ResponseStep.ON_READ_COMPLETED, true, true);
+    }
+
+    @SmallTest
+    @Feature({"Cronet"})
+    public void testThrowOnSucceeded() {
+        TestBidirectionalStreamCallback callback = new TestBidirectionalStreamCallback();
+        callback.setFailure(FailureType.THROW_SYNC, ResponseStep.ON_SUCCEEDED);
+        BidirectionalStream.Builder builder =
+                new BidirectionalStream.Builder(Http2TestServer.getEchoMethodUrl(), callback,
+                        callback.getExecutor(), mTestFramework.mCronetEngine);
+        BidirectionalStream stream = builder.setHttpMethod("GET").build();
+        stream.start();
+        callback.blockForDone();
+        assertEquals(callback.mResponseStep, ResponseStep.ON_SUCCEEDED);
+        assertTrue(stream.isDone());
+        assertNotNull(callback.mResponseInfo);
+        assertNull(callback.mError);
+        assertFalse(callback.mOnErrorCalled);
+    }
+
+    @SmallTest
+    @Feature({"Cronet"})
+    public void testExecutorShutdown() {
+        TestBidirectionalStreamCallback callback = new TestBidirectionalStreamCallback();
+
+        callback.setAutoAdvance(false);
+        BidirectionalStream.Builder builder =
+                new BidirectionalStream.Builder(Http2TestServer.getEchoMethodUrl(), callback,
+                        callback.getExecutor(), mTestFramework.mCronetEngine);
+        CronetBidirectionalStream stream =
+                (CronetBidirectionalStream) builder.setHttpMethod("GET").build();
+        stream.start();
+        callback.waitForNextReadStep();
+        assertFalse(callback.isDone());
+        assertFalse(stream.isDone());
+
+        final ConditionVariable requestDestroyed = new ConditionVariable(false);
+        stream.setOnDestroyedCallbackForTesting(new Runnable() {
+            @Override
+            public void run() {
+                requestDestroyed.open();
+            }
+        });
+
+        // Shutdown the executor, so posting the task will throw an exception.
+        callback.shutdownExecutor();
+        ByteBuffer readBuffer = ByteBuffer.allocateDirect(5);
+        stream.read(readBuffer);
+        // Callback will never be called again because executor is shutdown,
+        // but stream will be destroyed from network thread.
+        requestDestroyed.block();
+
+        assertFalse(callback.isDone());
+        assertTrue(stream.isDone());
+    }
+
+    /**
+     * Callback that shutdowns the engine when the stream has succeeded
+     * or failed.
+     */
+    class ShutdownTestBidirectionalStreamCallback extends TestBidirectionalStreamCallback {
+        @Override
+        public void onSucceeded(BidirectionalStream stream, UrlResponseInfo info) {
+            mTestFramework.mCronetEngine.shutdown();
+            // Clear mCronetEngine so it doesn't get shutdown second time in tearDown().
+            mTestFramework.mCronetEngine = null;
+            super.onSucceeded(stream, info);
+        }
+
+        @Override
+        public void onFailed(
+                BidirectionalStream stream, UrlResponseInfo info, CronetException error) {
+            mTestFramework.mCronetEngine.shutdown();
+            // Clear mCronetEngine so it doesn't get shutdown second time in tearDown().
+            mTestFramework.mCronetEngine = null;
+            super.onFailed(stream, info, error);
+        }
+    }
+
+    @SmallTest
+    @Feature({"Cronet"})
+    public void testShutdown() throws Exception {
+        TestBidirectionalStreamCallback callback = new ShutdownTestBidirectionalStreamCallback();
+        // Block callback when response starts to verify that shutdown fails
+        // if there are active stream.
+        callback.setAutoAdvance(false);
+        BidirectionalStream.Builder builder =
+                new BidirectionalStream.Builder(Http2TestServer.getEchoMethodUrl(), callback,
+                        callback.getExecutor(), mTestFramework.mCronetEngine);
+        CronetBidirectionalStream stream =
+                (CronetBidirectionalStream) builder.setHttpMethod("GET").build();
+        stream.start();
+        try {
+            mTestFramework.mCronetEngine.shutdown();
+            fail("Should throw an exception");
+        } catch (Exception e) {
+            assertEquals("Cannot shutdown with active requests.", e.getMessage());
+        }
+
+        callback.waitForNextReadStep();
+        assertEquals(ResponseStep.ON_RESPONSE_STARTED, callback.mResponseStep);
+        try {
+            mTestFramework.mCronetEngine.shutdown();
+            fail("Should throw an exception");
+        } catch (Exception e) {
+            assertEquals("Cannot shutdown with active requests.", e.getMessage());
+        }
+        callback.startNextRead(stream);
+
+        callback.waitForNextReadStep();
+        assertEquals(ResponseStep.ON_READ_COMPLETED, callback.mResponseStep);
+        try {
+            mTestFramework.mCronetEngine.shutdown();
+            fail("Should throw an exception");
+        } catch (Exception e) {
+            assertEquals("Cannot shutdown with active requests.", e.getMessage());
+        }
+
+        // May not have read all the data, in theory. Just enable auto-advance
+        // and finish the request.
+        callback.setAutoAdvance(true);
+        callback.startNextRead(stream);
+        callback.blockForDone();
+    }
+
+    @SmallTest
+    @Feature({"Cronet"})
+    public void testShutdownAfterError() throws Exception {
+        TestBidirectionalStreamCallback callback = new ShutdownTestBidirectionalStreamCallback();
+        BidirectionalStream.Builder builder =
+                new BidirectionalStream.Builder(Http2TestServer.getEchoMethodUrl(), callback,
+                        callback.getExecutor(), mTestFramework.mCronetEngine);
+        CronetBidirectionalStream stream =
+                (CronetBidirectionalStream) builder.setHttpMethod("GET").build();
+        stream.start();
+        callback.setFailure(FailureType.THROW_SYNC, ResponseStep.ON_READ_COMPLETED);
+        callback.blockForDone();
+        assertTrue(callback.mOnErrorCalled);
+    }
+
+    @SmallTest
+    @Feature({"Cronet"})
+    public void testShutdownAfterCancel() throws Exception {
+        TestBidirectionalStreamCallback callback = new ShutdownTestBidirectionalStreamCallback();
+        BidirectionalStream.Builder builder =
+                new BidirectionalStream.Builder(Http2TestServer.getEchoMethodUrl(), callback,
+                        callback.getExecutor(), mTestFramework.mCronetEngine);
+        CronetBidirectionalStream stream =
+                (CronetBidirectionalStream) builder.setHttpMethod("GET").build();
+
+        // Block callback when response starts to verify that shutdown fails
+        // if there are active requests.
+        callback.setAutoAdvance(false);
+        stream.start();
+        try {
+            mTestFramework.mCronetEngine.shutdown();
+            fail("Should throw an exception");
+        } catch (Exception e) {
+            assertEquals("Cannot shutdown with active requests.", e.getMessage());
+        }
+        callback.waitForNextReadStep();
+        assertEquals(ResponseStep.ON_RESPONSE_STARTED, callback.mResponseStep);
+        stream.cancel();
+    }
+
+    // Returns the contents of byteBuffer, from its position() to its limit(),
+    // as a String. Does not modify byteBuffer's position().
+    private String bufferContentsToString(ByteBuffer byteBuffer, int start, int end) {
+        // Use a duplicate to avoid modifying byteBuffer.
+        ByteBuffer duplicate = byteBuffer.duplicate();
+        duplicate.position(start);
+        duplicate.limit(end);
+        byte[] contents = new byte[duplicate.remaining()];
+        duplicate.get(contents);
+        return new String(contents);
+    }
 }