Initial implementation of CronetBidirectionalStream.

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;
+import org.chromium.base.annotations.NativeClassQualifiedName;
+
+import java.nio.ByteBuffer;
+import java.util.AbstractMap;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.Executor;
+import java.util.concurrent.RejectedExecutionException;
+
+import javax.annotation.concurrent.GuardedBy;
+
+/**
+ * {@link BidirectionalStream} implementation using Chromium network stack.
+ * All @CalledByNative methods are called on the native network thread
+ * and post tasks with callback calls onto Executor. Upon return from callback, the native
+ * stream is called on Executor thread and posts native tasks to the native network thread.
+ */
+@JNINamespace("cronet")
+class CronetBidirectionalStream extends BidirectionalStream {
+    /**
+     * States of BidirectionalStream are tracked in mReadState and mWriteState.
+     * The write state is separated out as it changes independently of the read state.
+     * There is one initial state - State.NOT_STARTED. There is one normal final state -

[xunjieli, 2016/01/21 19:13:08]

nit: single dash reads very much like a minus sign to me. Maybe consider using
double dash (--) or comma. 

[mef, 2016/01/22 14:33:44]

Interesting. Double dash seems unusual to me. :) I've changed dash to
semicolumn.

[xunjieli, 2016/01/22 14:52:23]

Acknowledged.

+     * State.SUCCESS, reached after State.READING_DONE and State.WRITING_DONE. There are two
+     * exceptional final states - State.CANCELED and State.ERROR, which can be reached from
+     * any other non-final state.
+     */
+    private enum State {
+        /* Initial state, stream not started. */
+        NOT_STARTED,
+        /* Stream started, request headers are being sent. */
+        STARTED,
+        /* Waiting for {@code read()} to be called. */
+        WAITING_ON_READ,

[xunjieli, 2016/01/21 19:13:08]

Should this say WAITING_FOR_READ instead of ON? Since we are waiting for
read() to be called.

[mef, 2016/01/22 14:33:44]

Excellent suggestion! Done.

+        /* Reading from the remote, {@code onReadCompleted()} callback will be called when done. */
+        READING,
+        /* There is no more data to read and stream is half-closed by the remote side. */
+        READING_DONE,
+        /* Stream is canceled. */
+        CANCELED,
+        /* Error has occured, stream is closed. */
+        ERROR,
+        /* Reading and writing is done, and the stream is closed successfully. */

[xunjieli, 2016/01/21 19:13:08]

nit: s/is/are. Reading and writing are done

[mef, 2016/01/22 14:33:44]

Done.

+        SUCCESS,
+        /* Waiting for {@code write()} to be called. */
+        WAITING_ON_WRITE,

[xunjieli, 2016/01/21 19:13:08]

Should this say WAITING_FOR_WRITE instead of ON? Since we are waiting for
write() to be called.

[mef, 2016/01/22 14:33:44]

Done.

+        /* Writing to the remote, {@code onWriteCompleted()} callback will be called when done. */
+        WRITING,
+        /* Writing the last frame, so {@code State.WRITING_DONE} will be set upon completion. */
+        WRITING_END_OF_STREAM,
+        /* There is no more data to write and stream is half-closed by the local side. */
+        WRITING_DONE,
+    }
+
+    private final CronetUrlRequestContext mRequestContext;
+    private final Executor mExecutor;
+    private final Callback mCallback;
+    private final String mInitialUrl;
+    private final int mInitialPriority;
+    private final String mInitialMethod;
+    private final String mRequestHeaders[];
+
+    /*
+     * Synchronize access to mNativeStream, mReadState and mWriteState.

[xunjieli, 2016/01/21 19:13:08]

nit: s/Synchronize/Synchronizes

[mef, 2016/01/22 14:33:44]

Done.

+     */
+    private final Object mNativeStreamLock = new Object();
+
+    /* Native BidirectionalStream object, owned by CronetBidirectionalStream. */
+    @GuardedBy("mNativeStreamLock") private long mNativeStream;
+
+    /**
+     * Read state is tracking reading flow.
+     *                         / <--- READING <--- \
+     *                         |                   |
+     *                         \                   /
+     * NOT_STARTED -> STARTED --> WAITING_ON_READ -> READING_DONE -> SUCCESS

[xunjieli, 2016/01/21 19:13:08]

Great flow diagram! This makes things much easier to understand. Have you
checked out the generated JavaDoc? Not sure if JavaDoc generator will mess this
up.

[mef, 2016/01/22 14:33:44]

Kudos to Paul for suggestion! I don't think it gets into javadoc as it is just a
private member.

[xunjieli, 2016/01/22 14:52:23]

Acknowledged. Yep you are right. Didn't realize that.

+     */
+    @GuardedBy("mNativeStreamLock") private State mReadState = State.NOT_STARTED;
+
+    /**
+     * Write state is tracking writing flow.
+     *                         / <--- WRITING <---  \
+     *                         |                    |
+     *                         \                    /
+     * NOT_STARTED -> STARTED --> WAITING_ON_WRITE -> WRITING_END_OF_STREAM -> WRITING_DONE ->
+     * SUCCESS
+     */
+    @GuardedBy("mNativeStreamLock") private State mWriteState = State.NOT_STARTED;
+
+    private UrlResponseInfo mResponseInfo;
+
+    /*
+     * OnReadCompleted callback is repeatedly invoked when each read is completed, so it
+     * is cached as a member variable.
+     */
+    private OnReadCompletedRunnable mOnReadCompletedTask;
+
+    /*
+     * OnWriteCompleted callback is repeatedly invoked when each write is completed, so it
+     * is cached as a member variable.
+     */
+    private OnWriteCompletedRunnable mOnWriteCompletedTask;
+
+    private Runnable mOnDestroyedCallbackForTesting;
+
+    private final class OnReadCompletedRunnable implements Runnable {
+        // Buffer passed back from current invocation of onReadCompleted.
+        ByteBuffer mByteBuffer;
+        // End of stream flag from current invocation of onReadCompleted.
+        boolean mEndOfStream;
+
+        @Override
+        public void run() {
+            try {
+                // Null out mByteBuffer, to pass buffer ownership to callback or release if done.
+                ByteBuffer buffer = mByteBuffer;
+                mByteBuffer = null;
+                synchronized (mNativeStreamLock) {
+                    if (isDoneLocked()) {
+                        return;
+                    }
+                    if (mEndOfStream) {
+                        mReadState = State.READING_DONE;
+                        if (maybeSucceededLocked()) {
+                            return;
+                        }
+                    } else {
+                        mReadState = State.WAITING_ON_READ;
+                    }
+                }
+                mCallback.onReadCompleted(CronetBidirectionalStream.this, mResponseInfo, buffer);
+            } catch (Exception e) {
+                onCallbackException(e);
+            }
+        }
+    }
+
+    private final class OnWriteCompletedRunnable implements Runnable {
+        // Buffer passed back from current invocation of onWriteCompleted.
+        ByteBuffer mByteBuffer;
+
+        @Override
+        public void run() {
+            try {
+                // Null out mByteBuffer, to pass buffer ownership to callback or release if done.
+                ByteBuffer buffer = mByteBuffer;
+                mByteBuffer = null;
+                synchronized (mNativeStreamLock) {
+                    if (isDoneLocked()) {
+                        return;
+                    }
+                    if (mWriteState == State.WRITING_END_OF_STREAM) {
+                        mWriteState = State.WRITING_DONE;
+                        if (maybeSucceededLocked()) {
+                            return;
+                        }
+                    } else {
+                        mWriteState = State.WAITING_ON_WRITE;
+                    }
+                }
+                mCallback.onWriteCompleted(CronetBidirectionalStream.this, mResponseInfo, buffer);
+            } catch (Exception e) {
+                onCallbackException(e);
+            }
+        }
+    }
+
+    CronetBidirectionalStream(CronetUrlRequestContext requestContext, String url,
+            @BidirectionalStream.Builder.StreamPriority int priority, Callback callback,
+            Executor executor, String httpMethod, List<Map.Entry<String, String>> requestHeaders) {
+        mRequestContext = requestContext;
+        mInitialUrl = url;
+        mInitialPriority = convertStreamPriority(priority);
+        mCallback = callback;
+        mExecutor = executor;
+        mInitialMethod = httpMethod;
+        mRequestHeaders = stringsFromHeaderList(requestHeaders);
+    }
+
+    @GuardedBy("mNativeStreamLock")
+    private boolean maybeSucceededLocked() {

[xunjieli, 2016/01/21 19:13:08]

- Suggest making this method return void. (Early return when not done). 
- I am trying to think of a better name for this method. The "Locked" suffix
makes it hard to read. Maybe just simply "maybeSucceed" ?
- Maybe also move this private method to where other private methods are.

[mef, 2016/01/22 14:33:44]

I'm not sure I understand, the result is important to callers.
It used to be called 'maybeSucceed', I've renamed it to maybeSucceedLocked
(among other xyzLocked methods) to signify that they must be guarded externally.
It was Paul's suggestion that I agree with.
Done.

[xunjieli, 2016/01/22 14:52:23]

I think I missed a part of the logic in the first pass. Could you add
documentation to this method? (specially the return value).
Right now the method is named as maybeSucceededLocked(). There are an extra "ed"
after Succeed. Could you remove that?

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

Done.

+        if (mReadState != State.READING_DONE || mWriteState != State.WRITING_DONE) {
+            return false;
+        }
+
+        mReadState = mWriteState = State.SUCCESS;
+        postTaskToExecutor(new Runnable() {
+            public void run() {
+                synchronized (mNativeStreamLock) {
+                    if (isDoneLocked()) {
+                        return;
+                    }
+                    // Destroy native stream first, so request context could be shut

[xunjieli, 2016/01/21 19:13:08]

nit: maybe spell out the "request context" as UrlRequestContext? a bit clearer
that way.

[mef, 2016/01/22 14:33:43]

Done.

+                    // down from the listener.
+                    destroyNativeStreamLocked(false);
+                }
+                try {
+                    mCallback.onSucceeded(CronetBidirectionalStream.this, mResponseInfo);
+                } catch (Exception e) {
+                    Log.e(CronetUrlRequestContext.LOG_TAG, "Exception in onSucceeded method", e);
+                }
+            }
+        });
+        return true;
+    }
+
+    private static boolean doesMethodAllowWriteData(String methodName) {
+        return !methodName.equals("GET") && !methodName.equals("HEAD");
+    }
+
+    @Override
+    public void start() {
+        synchronized (mNativeStreamLock) {
+            if (mReadState != State.NOT_STARTED) {
+                throw new IllegalStateException("Stream is already started.");
+            }
+            try {
+                mNativeStream = nativeCreateBidirectionalStream(
+                        mRequestContext.getUrlRequestContextAdapter());
+                mRequestContext.onRequestStarted();
+                // Non-zero startResult means an argument error.
+                int startResult = nativeStart(mNativeStream, mInitialUrl, mInitialPriority,
+                        mInitialMethod, mRequestHeaders, !doesMethodAllowWriteData(mInitialMethod));
+                if (startResult == -1) {
+                    throw new IllegalArgumentException("Invalid http method " + mInitialMethod);
+                }
+                if (startResult > 0) {
+                    int headerPos = startResult - 1;
+                    throw new IllegalArgumentException("Invalid header "
+                            + mRequestHeaders[headerPos] + "=" + mRequestHeaders[headerPos + 1]);
+                }
+                mReadState = mWriteState = State.STARTED;
+            } catch (RuntimeException e) {
+                // If there's an exception, cleanup and then throw the

[xunjieli, 2016/01/21 19:13:08]

nit: s/cleanup/clean up. (Cleanup is the noun)

[mef, 2016/01/22 14:33:44]

Done.

+                // exception to the caller.
+                destroyNativeStreamLocked(false);
+                throw e;
+            }
+        }
+    }
+
+    @Override
+    public void read(ByteBuffer buffer) {
+        synchronized (mNativeStreamLock) {
+            Preconditions.checkHasRemaining(buffer);
+            Preconditions.checkDirect(buffer);
+            if (mReadState != State.WAITING_ON_READ) {
+                throw new IllegalStateException("Unexpected read attempt.");
+            }
+            if (isDoneLocked()) {
+                return;
+            }
+            mReadState = State.READING;
+            if (!nativeReadData(mNativeStream, buffer, buffer.position(), buffer.limit())) {
+                // Still waiting on read. This is just to have consistent
+                // behavior with the other error cases.
+                mReadState = State.WAITING_ON_READ;
+                throw new IllegalArgumentException("Unable to call native read");
+            }
+        }
+    }
+
+    @Override
+    public void write(ByteBuffer buffer, boolean endOfStream) {
+        synchronized (mNativeStreamLock) {
+            Preconditions.checkDirect(buffer);
+            if (!buffer.hasRemaining() && !endOfStream) {
+                throw new IllegalArgumentException("Empty buffer before end of stream.");
+            }
+            if (mWriteState != State.WAITING_ON_WRITE) {
+                throw new IllegalStateException("Unexpected write attempt.");
+            }
+            if (isDoneLocked()) {
+                return;
+            }
+            mWriteState = endOfStream ? State.WRITING_END_OF_STREAM : State.WRITING;
+            if (!nativeWriteData(
+                        mNativeStream, buffer, buffer.position(), buffer.limit(), endOfStream)) {
+                // Still waiting on write. This is just to have consistent
+                // behavior with the other error cases.
+                mWriteState = State.WAITING_ON_WRITE;
+                throw new IllegalArgumentException("Unable to call native write");
+            }
+        }
+    }
+
+    @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.");
+    }
+
+    @Override
+    public void cancel() {
+        synchronized (mNativeStreamLock) {
+            if (isDoneLocked() || mReadState == State.NOT_STARTED) {
+                return;
+            }
+            mReadState = mWriteState = State.CANCELED;
+            destroyNativeStreamLocked(true);
+        }
+    }
+
+    @Override
+    public boolean isDone() {
+        synchronized (mNativeStreamLock) {
+            return isDoneLocked();
+        }
+    }
+
+    @GuardedBy("mNativeStreamLock")
+    private boolean isDoneLocked() {

[xunjieli, 2016/01/21 19:13:08]

I believe grabbing the same lock is essentially no-op and allowed in Java. Maybe
get rid of this method, and just use isDone().

[pauljensen, 2016/01/22 03:55:31]

Locks in Java are re-entrant but I imagine there may be a significant cost to
any synchronization operations.  My previous investigations into Cronet
performance saw significant fractions of our time in Java locking.

[mef, 2016/01/22 14:33:44]

See previous discussion with Paul. I would imagine that already-grabbed lock is
cheap, but not quite free.

+        return mReadState != State.NOT_STARTED && mNativeStream == 0;
+    }
+
+    @SuppressWarnings("unused")
+    @CalledByNative
+    private void onRequestHeadersSent() {
+        postTaskToExecutor(new Runnable() {
+            public void run() {
+                synchronized (mNativeStreamLock) {
+                    if (isDoneLocked()) {
+                        return;
+                    }
+                    if (doesMethodAllowWriteData(mInitialMethod)) {
+                        mWriteState = State.WAITING_ON_WRITE;
+                    } else {
+                        mWriteState = State.WRITING_DONE;
+                    }
+                }
+
+                try {
+                    mCallback.onRequestHeadersSent(CronetBidirectionalStream.this);
+                } catch (Exception e) {
+                    onCallbackException(e);
+                }
+            }
+        });
+    }
+
+    /**
+     * Called when the final set of headers, after all redirects,
+     * is received. Can only be called once for each stream.
+     */
+    @SuppressWarnings("unused")
+    @CalledByNative
+    private void onResponseHeadersReceived(int httpStatusCode, String negotiatedProtocol,
+            String[] headers, long receivedBytesCount) {
+        mResponseInfo =
+                prepareResponseInfoOnNetworkThread(httpStatusCode, negotiatedProtocol, headers);
+        mResponseInfo.setReceivedBytesCount(receivedBytesCount);
+        postTaskToExecutor(new Runnable() {
+            public void run() {
+                synchronized (mNativeStreamLock) {
+                    if (isDoneLocked()) {
+                        return;
+                    }
+                    mReadState = State.WAITING_ON_READ;
+                }
+
+                try {
+                    mCallback.onResponseHeadersReceived(
+                            CronetBidirectionalStream.this, mResponseInfo);
+                } catch (Exception e) {
+                    onCallbackException(e);
+                }
+            }
+        });
+    }
+
+    @SuppressWarnings("unused")
+    @CalledByNative
+    private void onReadCompleted(final ByteBuffer byteBuffer, int bytesRead, int initialPosition,
+            int initialLimit, long receivedBytesCount) {
+        mResponseInfo.setReceivedBytesCount(receivedBytesCount);
+        if (byteBuffer.position() != initialPosition || byteBuffer.limit() != initialLimit) {
+            failWithException(
+                    new CronetException("ByteBuffer modified externally during read", null));
+            return;
+        }
+        if (bytesRead < 0 || initialPosition + bytesRead > initialLimit) {
+            failWithException(new CronetException("Invalid number of bytes read", null));
+            return;
+        }
+        if (mOnReadCompletedTask == null) {
+            mOnReadCompletedTask = new OnReadCompletedRunnable();
+        }
+        byteBuffer.position(initialPosition + bytesRead);
+        mOnReadCompletedTask.mByteBuffer = byteBuffer;
+        mOnReadCompletedTask.mEndOfStream = (bytesRead == 0);
+        postTaskToExecutor(mOnReadCompletedTask);
+    }
+
+    @SuppressWarnings("unused")
+    @CalledByNative
+    private void onWriteCompleted(
+            final ByteBuffer byteBuffer, int initialPosition, int initialLimit) {
+        if (byteBuffer.position() != initialPosition || byteBuffer.limit() != initialLimit) {
+            failWithException(
+                    new CronetException("ByteBuffer modified externally during write", null));
+            return;
+        }
+        if (mOnWriteCompletedTask == null) {
+            mOnWriteCompletedTask = new OnWriteCompletedRunnable();
+        }
+        // Current implementation always writes the complete buffer.
+        byteBuffer.position(byteBuffer.limit());
+        mOnWriteCompletedTask.mByteBuffer = byteBuffer;
+        postTaskToExecutor(mOnWriteCompletedTask);
+    }
+
+    @SuppressWarnings("unused")
+    @CalledByNative
+    private void onResponseTrailersReceived(String[] trailers) {
+        final UrlResponseInfo.HeaderBlock trailersBlock =
+                new UrlResponseInfo.HeaderBlock(headersListFromStrings(trailers));
+        postTaskToExecutor(new Runnable() {
+            public void run() {
+                synchronized (mNativeStreamLock) {
+                    if (isDoneLocked()) {
+                        return;
+                    }
+                }
+                try {
+                    mCallback.onResponseTrailersReceived(
+                            CronetBidirectionalStream.this, mResponseInfo, trailersBlock);
+                } catch (Exception e) {
+                    onCallbackException(e);
+                }
+            }
+        });
+    }
+
+    @SuppressWarnings("unused")
+    @CalledByNative
+    private void onError(final int nativeError, final String errorString, long receivedBytesCount) {
+        if (mResponseInfo != null) {
+            mResponseInfo.setReceivedBytesCount(receivedBytesCount);
+        }
+        failWithException(new CronetException(
+                "Exception in BidirectionalStream: " + errorString, nativeError));
+    }
+
+    /**
+     * Called when request is canceled, no callbacks will be called afterwards.
+     */
+    @SuppressWarnings("unused")
+    @CalledByNative
+    private void onCanceled() {
+        postTaskToExecutor(new Runnable() {
+            public void run() {
+                try {
+                    mCallback.onCanceled(CronetBidirectionalStream.this, mResponseInfo);
+                } catch (Exception e) {
+                    Log.e(CronetUrlRequestContext.LOG_TAG, "Exception in onCanceled method", e);
+                }
+            }
+        });
+    }
+
+    @VisibleForTesting
+    public void setOnDestroyedCallbackForTesting(Runnable onDestroyedCallbackForTesting) {
+        mOnDestroyedCallbackForTesting = onDestroyedCallbackForTesting;
+    }
+
+    /**
+     * Posts task to application Executor. Used for callbacks
+     * and other tasks that should not be executed on network thread.
+     */
+    private void postTaskToExecutor(Runnable task) {
+        try {
+            mExecutor.execute(task);
+        } catch (RejectedExecutionException failException) {
+            Log.e(CronetUrlRequestContext.LOG_TAG, "Exception posting task to executor",
+                    failException);
+            // If posting a task throws an exception, then there is no choice
+            // but to destroy the stream without invoking the callback.
+            synchronized (mNativeStreamLock) {
+                mReadState = mWriteState = State.ERROR;
+                destroyNativeStreamLocked(false);
+            }
+        }
+    }
+
+    private static ArrayList<Map.Entry<String, String>> headersListFromStrings(String[] headers) {
+        ArrayList<Map.Entry<String, String>> headersList =
+                new ArrayList<Map.Entry<String, String>>(headers.length / 2);
+        for (int i = 0; i < headers.length; i += 2) {
+            headersList.add(new AbstractMap.SimpleImmutableEntry<String, String>(
+                    headers[i], headers[i + 1]));
+        }
+        return headersList;
+    }
+
+    private static String[] stringsFromHeaderList(List<Map.Entry<String, String>> headersList) {
+        String headersArray[] = new String[headersList.size() * 2];
+        int i = 0;
+        for (Map.Entry<String, String> requestHeader : headersList) {
+            headersArray[i++] = requestHeader.getKey();
+            headersArray[i++] = requestHeader.getValue();
+        }
+        return headersArray;
+    }
+
+    private UrlResponseInfo prepareResponseInfoOnNetworkThread(
+            int httpStatusCode, String negotiatedProtocol, String[] headers) {
+        synchronized (mNativeStreamLock) {
+            if (mNativeStream == 0) {
+                return null;
+            }
+        }
+
+        ArrayList<String> urlChain = new ArrayList<String>();
+        urlChain.add(mInitialUrl);
+
+        boolean wasCached = false;
+        String httpStatusText = "";
+        String proxyServer = null;
+
+        UrlResponseInfo responseInfo = new UrlResponseInfo(urlChain, httpStatusCode, httpStatusText,
+                headersListFromStrings(headers), wasCached, negotiatedProtocol, proxyServer);
+        return responseInfo;
+    }
+
+    private static int convertStreamPriority(
+            @BidirectionalStream.Builder.StreamPriority int priority) {
+        switch (priority) {
+            case Builder.STREAM_PRIORITY_IDLE:
+                return RequestPriority.IDLE;
+            case Builder.STREAM_PRIORITY_LOWEST:
+                return RequestPriority.LOWEST;
+            case Builder.STREAM_PRIORITY_LOW:
+                return RequestPriority.LOW;
+            case Builder.STREAM_PRIORITY_MEDIUM:
+                return RequestPriority.MEDIUM;
+            case Builder.STREAM_PRIORITY_HIGHEST:
+                return RequestPriority.HIGHEST;
+            default:
+                throw new IllegalArgumentException("Invalid stream priority.");
+        }
+    }
+
+    @GuardedBy("mNativeStreamLock")
+    private void destroyNativeStreamLocked(boolean sendOnCanceled) {

[xunjieli, 2016/01/21 19:13:08]

Having the "Locked" suffix is confusing. Suggest getting rid of it and grab the
lock. Grabbing lock twice is okay.

[pauljensen, 2016/01/22 03:55:31]

I thought the "Locked" suffix was a common pattern.  There are more then 1,000
instances of "Locked(" in Chromium and more than 10,000 instances in Android.  I
think the performance impact will be worth it here.

[xunjieli, 2016/01/22 13:45:07]

I see. Learned of it for the first time. SGTM then!

[mef, 2016/01/22 14:33:44]

Acknowledged.

+        Log.i(CronetUrlRequestContext.LOG_TAG, "destroyNativeStreamLocked " + this.toString());
+        if (mNativeStream == 0) {
+            return;
+        }
+        nativeDestroy(mNativeStream, sendOnCanceled);
+        mNativeStream = 0;
+        mRequestContext.onRequestDestroyed();
+        if (mOnDestroyedCallbackForTesting != null) {
+            mOnDestroyedCallbackForTesting.run();
+        }
+    }
+
+    /**
+     * Fails the stream with an exception. Only called on the Executor.
+     */
+    private void failWithExceptionOnExecutor(CronetException e) {
+        // Do not call into listener if request is complete.
+        synchronized (mNativeStreamLock) {
+            if (isDoneLocked()) {
+                return;
+            }
+            mReadState = mWriteState = State.ERROR;
+            destroyNativeStreamLocked(false);
+        }
+        try {
+            mCallback.onFailed(this, mResponseInfo, e);
+        } catch (Exception failException) {
+            Log.e(CronetUrlRequestContext.LOG_TAG, "Exception notifying of failed request",
+                    failException);
+        }
+    }
+
+    /**
+     * If callback method throws an exception, stream gets canceled
+     * and exception is reported via onFailed callback.
+     * Only called on the Executor.
+     */
+    private void onCallbackException(Exception e) {
+        CronetException streamError =
+                new CronetException("CalledByNative method has thrown an exception", e);
+        Log.e(CronetUrlRequestContext.LOG_TAG, "Exception in CalledByNative method", e);
+        failWithExceptionOnExecutor(streamError);
+    }
+
+    /**
+     * Fails the stream with an exception. Can be called on any thread.
+     */
+    private void failWithException(final CronetException exception) {
+        postTaskToExecutor(new Runnable() {
+            public void run() {
+                failWithExceptionOnExecutor(exception);
+            }
+        });
+    }
+
+    // Native methods are implemented in cronet_bidirectional_stream_adapter.cc.
+    private native long nativeCreateBidirectionalStream(long urlRequestContextAdapter);
+
+    @NativeClassQualifiedName("CronetBidirectionalStreamAdapter")
+    private native int nativeStart(long nativePtr, String url, int priority, String method,
+            String[] headers, boolean endOfStream);
+
+    @NativeClassQualifiedName("CronetBidirectionalStreamAdapter")
+    private native boolean nativeReadData(
+            long nativePtr, ByteBuffer byteBuffer, int position, int limit);
+
+    @NativeClassQualifiedName("CronetBidirectionalStreamAdapter")
+    private native boolean nativeWriteData(
+            long nativePtr, ByteBuffer byteBuffer, int position, int limit, boolean endOfStream);
+
+    @NativeClassQualifiedName("CronetBidirectionalStreamAdapter")
+    private native void nativeDestroy(long nativePtr, boolean sendOnCanceled);
+}