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;
+
+/**
+ * BidirectionalStream implementation using Chromium network stack.
+ * All @CallByNative methods are called on native network thread
+ * and post tasks with callback calls onto Executor. Upon return from callback native
+ * stream is called on executor thread and posts native tasks to 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 stream state.
+     * There is one initial state - State.NOT_STARTED. There is one final state - State.SUCCESS,
+     * reached after State.READING_DONE and State.WRITING_DONE. There are 2 exception states -
+     * State.CANCELED and State.ERROR, which can be reached from any other state except
+     * State.SUCCESS.
+     */
+    private enum State {
+        /* Initial state, stream not started. */
+        NOT_STARTED,
+        /* Stream started, request headers are sent. */
+        STARTED,
+        /* Waiting for {@code read()} to be called. */
+        WAITING_ON_READ,
+        /* 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. */
+        SUCCESS,
+        /* Waiting for {@code write()} to be called. */
+        WAITING_ON_WRITE,
+        /* 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 ArrayList<Map.Entry<String, String>> mRequestHeaders;
+
+    /*
+     * Synchronize access to mNativeStream, mReadState and mWriteState.
+     */
+    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
+     */
+    @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 {
+                synchronized (mNativeStreamLock) {
+                    if (isDoneLocked() || mNativeStream == 0) {

[pauljensen, 2016/01/11 20:05:45]

how could mNativeStream == 0 here but isDoneLocked() return false?  I think the
stream would have to be in the NOT_STARTED state, which I think is impossible
considering we're issuing a read-completed callback.

[mef, 2016/01/11 23:22:50]

Good point. Done.

+                        return;
+                    }
+                    if (mEndOfStream) {
+                        mReadState = State.READING_DONE;
+                        if (maybeSucceeded()) return;

[pauljensen, 2016/01/11 20:05:45]

our style isn't to put return on the same line; we use curly braces, see line
127, 197, 260 for example

[mef, 2016/01/11 23:22:50]

Done.

+                    } else {
+                        mReadState = State.WAITING_ON_READ;
+                    }
+                }
+                // Null out mByteBuffer, to pass buffer ownership to callback.
+                ByteBuffer buffer = mByteBuffer;

[pauljensen, 2016/01/11 20:05:45]

I think there is a race here because after line 136 (when the sync block ends) a
misbehaving BiDiStream user could call read() which could result in a
onReadCompleted() getting called which could race to set mByteBuffer while this
line races to read mByteBuffer.  If the wrong side loses the race we could call
mCallback.onReadCompleted() with the new buffer rather than the old one.  I
wonder if moving this read up as I describe on the next line might fix this.

This kind of a race is a byproduct of reusing mOnReadCompletedCallback.  This
kind of reuse kinda goes against the Java pattern of new'ing all the time (like
how people tend to write execute(new Runable() { blah }).

[mef, 2016/01/11 23:22:50]

Done.

+                mByteBuffer = null;

[pauljensen, 2016/01/11 20:05:45]

this line should be moved up above all the "return" statements in this function,
so that if the BiDiStream is cancelled the buffer is freed.

[mef, 2016/01/11 23:22:50]

Done.

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

Can we make this change to CronetUrlRequest.java too?

+                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 {
+                synchronized (mNativeStreamLock) {
+                    if (isDoneLocked() || mNativeStream == 0) {
+                        return;
+                    }
+                    if (mWriteState == State.WRITING_END_OF_STREAM) {
+                        mWriteState = State.WRITING_DONE;

[pauljensen, 2016/01/11 20:05:45]

did we necessarily write all of the buffer's data here?  it'd be nice to have
some kind of an assertion check here to make sure we wrote all of it (i.e.
mByteBuffer.remaining() == 0)

[mef, 2016/01/11 23:22:50]

Current implementation always writes the complete buffer (see onWriteCompleted()
line 428).

+                        if (maybeSucceeded()) return;

[pauljensen, 2016/01/11 20:05:45]

our style isn't to put return on the same line; we use curly braces, see line
127, 197, 260 for example

[mef, 2016/01/11 23:22:50]

Done.

+                    } else {
+                        mWriteState = State.WAITING_ON_WRITE;
+                    }
+                }
+                // Null out mByteBuffer, to pass buffer ownership to callback.
+                ByteBuffer buffer = mByteBuffer;
+                mByteBuffer = null;

[pauljensen, 2016/01/11 20:05:45]

this line should be moved up above all the "return" statements in this function,
so that if the BiDiStream is cancelled the buffer is freed.

[mef, 2016/01/11 23:22:50]

Done.

+                mCallback.onWriteCompleted(CronetBidirectionalStream.this, mResponseInfo, buffer);
+            } catch (Exception e) {
+                onCallbackException(e);
+            }
+        }
+    }
+
+    CronetBidirectionalStream(CronetUrlRequestContext requestContext, long urlRequestContextAdapter,
+            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 = new ArrayList<Map.Entry<String, String>>(requestHeaders);
+    }
+
+    @GuardedBy("mNativeStreamLock")
+    private boolean maybeSucceeded() {
+        if (mReadState != State.READING_DONE || mWriteState != State.WRITING_DONE) {
+            return false;
+        }
+
+        mReadState = mWriteState = State.SUCCESS;
+        Runnable task = new Runnable() {
+            public void run() {
+                synchronized (mNativeStreamLock) {
+                    if (isDoneLocked()) {
+                        return;
+                    }
+                    // Destroy native stream first, so request context could be shut
+                    // down from the listener.
+                    destroyNativeStream(false);
+                }
+                try {
+                    mCallback.onSucceeded(CronetBidirectionalStream.this, mResponseInfo);
+                } catch (Exception e) {
+                    Log.e(CronetUrlRequestContext.LOG_TAG, "Exception in onSucceeded method", e);
+                }
+            }
+        };
+        postTaskToExecutor(task);
+        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();
+                String headers[] = stringsFromHeaderList(mRequestHeaders);
+                // Non-zero startResult means an argument error.
+                int startResult = nativeStart(mNativeStream, mInitialUrl, mInitialPriority,
+                        mInitialMethod, headers, !doesMethodAllowWriteData(mInitialMethod));
+                if (startResult == -1) {
+                    throw new IllegalArgumentException("Invalid http method " + mInitialMethod);
+                }
+                if (startResult > 0) {
+                    int headerPos = startResult - 1;
+                    throw new IllegalArgumentException(
+                            "Invalid header " + headers[headerPos] + "=" + headers[headerPos + 1]);
+                }
+                mReadState = mWriteState = State.STARTED;
+            } catch (RuntimeException e) {
+                // If there's an exception, cleanup and then throw the
+                // exception to the caller.
+                destroyNativeStream(false);
+                throw e;
+            }
+        }
+    }
+
+    @Override
+    public void read(ByteBuffer buffer) {
+        synchronized (mNativeStreamLock) {
+            if (!buffer.hasRemaining()) {
+                throw new IllegalArgumentException("ByteBuffer is already full.");
+            }
+            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;
+                // Since accessing byteBuffer's memory failed, it's presumably
+                // not a direct ByteBuffer.
+                throw new IllegalArgumentException("byteBuffer must be a direct ByteBuffer.");
+            }
+        }
+    }
+
+    @Override
+    public void write(ByteBuffer buffer, boolean endOfStream) {
+        synchronized (mNativeStreamLock) {
+            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;
+                // Since accessing byteBuffer's memory failed, it's presumably
+                // not a direct ByteBuffer.
+                throw new IllegalArgumentException("byteBuffer must be a direct ByteBuffer.");
+            }
+        }
+    }
+
+    @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;
+            destroyNativeStream(true);
+        }
+    }
+
+    @Override
+    public boolean isDone() {
+        synchronized (mNativeStreamLock) {
+            return isDoneLocked();
+        }
+    }
+
+    @GuardedBy("mNativeStreamLock")
+    private boolean isDoneLocked() {
+        return mReadState != State.NOT_STARTED && mNativeStream == 0;
+    }
+
+    @SuppressWarnings("unused")
+    @CalledByNative
+    private void onRequestHeadersSent() {
+        Runnable task = 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);
+                }
+            }
+        };
+        postTaskToExecutor(task);
+    }
+
+    /**
+     * 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[] headers) {
+        mResponseInfo = prepareResponseInfoOnNetworkThread(httpStatusCode, headers);
+        Runnable task = 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);
+                }
+            }
+        };
+        postTaskToExecutor(task);
+    }
+
+    @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 > byteBuffer.limit()) {
+            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));
+        Runnable task = new Runnable() {
+            public void run() {
+                synchronized (mNativeStreamLock) {
+                    if (isDoneLocked()) {
+                        return;
+                    }
+                }
+                try {
+                    mCallback.onResponseTrailersReceived(
+                            CronetBidirectionalStream.this, mResponseInfo, trailersBlock);
+                } catch (Exception e) {
+                    onCallbackException(e);
+                }
+            }
+        };
+        postTaskToExecutor(task);
+    }
+
+    @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() {
+        Runnable task = new Runnable() {
+            public void run() {
+                try {
+                    mCallback.onCanceled(CronetBidirectionalStream.this, mResponseInfo);
+                } catch (Exception e) {
+                    Log.e(CronetUrlRequestContext.LOG_TAG, "Exception in onCanceled method", e);
+                }
+            }
+        };
+        postTaskToExecutor(task);
+    }
+
+    @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 cancel the stream.
+            cancel();
+        }
+    }
+
+    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(
+            ArrayList<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[] headers) {
+        long nativeStream;
+        synchronized (mNativeStreamLock) {
+            if (mNativeStream == 0) {
+                return null;
+            }
+            // This method is running on network thread, so even if
+            // mNativeStream is set to 0 from another thread the actual
+            // deletion of the adapter is posted to network thread, so it is
+            // safe to preserve and use nativeStream outside the lock.
+            nativeStream = mNativeStream;
+        }
+
+        ArrayList<String> urlChain = new ArrayList<String>();
+        urlChain.add(mInitialUrl);
+
+        boolean wasCached = false;
+        String httpStatusText = "";
+        String negotiatedProtocol = nativeGetNegotiatedProtocol(nativeStream);
+        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:
+                return RequestPriority.MEDIUM;
+        }
+    }
+
+    private void destroyNativeStream(boolean sendOnCanceled) {
+        synchronized (mNativeStreamLock) {
+            Log.i(CronetUrlRequestContext.LOG_TAG, "destroyNativeStream " + this.toString());
+            if (mNativeStream == 0) {
+                return;
+            }
+            nativeDestroy(mNativeStream, sendOnCanceled);
+            mNativeStream = 0;
+            mRequestContext.onRequestDestroyed();
+            if (mOnDestroyedCallbackForTesting != null) {
+                mOnDestroyedCallbackForTesting.run();
+            }
+        }
+    }
+
+    /**
+     * 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);
+        // Do not call into listener if request is complete.
+        synchronized (mNativeStreamLock) {
+            if (isDoneLocked()) {
+                return;
+            }

[pauljensen, 2016/01/11 20:05:45]

should we insert "mReadState = mWriteState = State.ERROR;"?
if we do, should we then dedup this code with that in the next function?

[mef, 2016/01/11 23:22:50]

Done.

+            destroyNativeStream(false);
+        }
+        try {
+            mCallback.onFailed(this, mResponseInfo, streamError);
+        } catch (Exception failException) {
+            Log.e(CronetUrlRequestContext.LOG_TAG, "Exception notifying of failed request",
+                    failException);
+        }
+    }
+
+    /**
+     * Fails the stream with an exception. Can be called on any thread.
+     */
+    private void failWithException(final CronetException exception) {
+        Runnable task = new Runnable() {
+            public void run() {
+                synchronized (mNativeStreamLock) {
+                    if (isDoneLocked()) {
+                        return;
+                    }
+                    mReadState = mWriteState = State.ERROR;
+                    destroyNativeStream(false);
+                }
+                try {
+                    mCallback.onFailed(CronetBidirectionalStream.this, mResponseInfo, exception);
+                } catch (Exception e) {
+                    Log.e(CronetUrlRequestContext.LOG_TAG, "Exception in onError method", e);
+                }
+            }
+        };
+        postTaskToExecutor(task);
+    }
+
+    // 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);
+
+    @NativeClassQualifiedName("CronetBidirectionalStreamAdapter")
+    private native String nativeGetNegotiatedProtocol(long nativePtr);
+}