Initial implementation of CronetBidirectionalStream.

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

Index: components/cronet/android/java/src/org/chromium/net/CronetBidirectionalStream.java
diff --git a/components/cronet/android/java/src/org/chromium/net/CronetBidirectionalStream.java b/components/cronet/android/java/src/org/chromium/net/CronetBidirectionalStream.java
new file mode 100644
index 0000000000000000000000000000000000000000..88a88c307c72d4e2f40d2819403db1d9babbecb8
--- /dev/null
+++ b/components/cronet/android/java/src/org/chromium/net/CronetBidirectionalStream.java
@@ -0,0 +1,646 @@
+// 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 mStreamState 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.
+     */
+    /* Initial state, stream not started. */
+    private static final int STATE_NOT_STARTED = 0;

[pauljensen, 2016/01/06 17:18:33]

let's change these states to an enum (they're allowed (and preferred for
type-safety) internally, just not as external APIs)

[mef, 2016/01/06 21:28:30]

Done.

+    /* Stream started, request headers are sent. */
+    private static final int STATE_STARTED = 1;
+    /* Waiting for {@code read()} to be called. */
+    private static final int STATE_WAITING_ON_READ = 2;
+    /* Reading from the remote, {@code onReadCompleted()} callback will be called when done. */
+    private static final int STATE_READING = 3;
+    /* There is no more data to read and stream is half-closed by the remote side. */
+    private static final int STATE_READING_DONE = 4;
+    /* Stream is canceled. */
+    private static final int STATE_CANCELED = 5;

[pauljensen, 2016/01/04 19:56:00]

no action required side-note: this state is set but never checked for, ditto for
STATE_ERROR.

[mef, 2016/01/04 22:27:07]

Good point, not sure where to check them. I think those are useful for debugging
(maybe I should add a toString() method?).
Should I add a checkStateForTesting() or something like that?

[pauljensen, 2016/01/06 17:18:33]

I don't think we need to take any action, or at least I don't see any action as
being very useful.

+    /* Error has occured, stream is closed. */
+    private static final int STATE_ERROR = 6;
+    /* Reading and writing is done, and the stream is closed successfully. */
+    private static final int STATE_SUCCESS = 7;
+    /* Waiting for {@code write()} to be called. */
+    private static final int STATE_WAITING_ON_WRITE = 10;
+    /* Writing to the remote, {@code onWriteCompleted()} callback will be called when done. */
+    private static final int STATE_WRITING = 11;
+    /* Writing the last frame, so {@code STATE_WRITING_DONE} will be set upon completion. */
+    private static final int STATE_WRITING_END_OF_STREAM = 12;
+    /* There is no more data to write and stream is half-closed by the local side. */
+    private static final int STATE_WRITING_DONE = 13;
+
+    /*

[pauljensen, 2016/01/06 17:18:33]

can we move these four lines down to line 75 so the lock sits beside the guarded

[mef, 2016/01/06 21:28:30]

Done.

+     * Synchronize access to mNativeStream, mStreamState and mWriteState.
+     */
+    private final Object mNativeStreamLock = new Object();
+    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;
+
+    /* Native BidirectionalStream object, owned by CronetBidirectionalStream. */
+    @GuardedBy("mNativeStreamLock") private long mNativeStream;
+
+    /**
+     * Stream state is tracking stream and reading flow.
+     * NOT_STARTED -> STARTED -> WAITING_ON_READ -> READING -> WAITING_ON_READ ->
+     * READING -> READING_DONE -> SUCCESS
+     */
+    @GuardedBy("mNativeStreamLock") private int mStreamState = STATE_NOT_STARTED;
+
+    /**
+     * Write state is tracking writing flow.
+     * NOT_STARTED -> WAITING_ON_WRITE -> WRITING -> WAITING_ON_WRITE ->
+     * WRITING_END_OF_STREAM -> WRITING_DONE -> SUCCESS

[pauljensen, 2016/01/04 19:56:00]

does mWriteState ever proceed to SUCCESS?

[mef, 2016/01/04 22:27:07]

Currently it stops at WRITING_DONE. I could fix the comment or fix the
mWriteState to be also set to SUCCESS / ERROR / CANCELED.
Helen has suggested to rename mStreamState into mReadState to match mWriteState.
 WDYT?

[pauljensen, 2016/01/06 17:18:33]

As we discussed offline, I think for consistency and for easier
readability/understandability, renaming mStreamState to mReadState and making
mWriteState progress through STARTED and going to SUCCESS/ERROR/CANCELED would
be an improvement.  Not a big improvement but an improvement.

[mef, 2016/01/06 21:28:30]

Done. I wonder whether we need all 3 - Stream State, Read State and Write State.

+     */
+    @GuardedBy("mNativeStreamLock") private int mWriteState = STATE_NOT_STARTED;
+
+    private UrlResponseInfo mResponseInfo;

[pauljensen, 2016/01/06 17:18:33]

should this be marked volatile?  it's written on one thread and read by other
threads

[mef, 2016/01/06 21:28:30]

mResponseInfo doesn't change after creation in onResponseHeadersReceived, only
its setReceivedBytesCount() gets called, but it uses AtomicLong.

+
+    /*
+     * 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() {
+            if (isDone()) {
+                return;
+            }
+            try {
+                synchronized (mNativeStreamLock) {
+                    if (mNativeStream == 0) {
+                        return;
+                    }
+                    if (mEndOfStream) {
+                        mStreamState = STATE_READING_DONE;
+                        if (maybeSucceeded()) return;
+                    } else {
+                        mStreamState = STATE_WAITING_ON_READ;
+                    }
+                }
+                // Null out mByteBuffer, out of paranoia. Has to be done before

[pauljensen, 2016/01/04 19:56:00]

nit: "out of paranoia" seems more like to facilitate garbage collection of large
buffers.  By setting it to null it's explicitly transferring ownership.

[mef, 2016/01/04 22:27:07]

Good point. The comment hasn't changed since times when mByteBuffer was pointing
to internal buffer allocated by UrlRequest.
Will change here and below.

[mef, 2016/01/06 21:28:30]

Done.

+                // mCallback call, to avoid any race when there are multiple
+                // executor threads.
+                ByteBuffer buffer = mByteBuffer;
+                mByteBuffer = null;
+                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() {
+            if (isDone()) {
+                return;
+            }
+            try {
+                synchronized (mNativeStreamLock) {
+                    if (mNativeStream == 0) {
+                        return;
+                    }
+                    if (mWriteState == STATE_WRITING_END_OF_STREAM) {
+                        mWriteState = STATE_WRITING_DONE;
+                        if (maybeSucceeded()) return;
+                    } else {
+                        mWriteState = STATE_WAITING_ON_WRITE;
+                    }
+                }
+                // Null out mByteBuffer, out of paranoia. Has to be done before
+                // mCallback call, to avoid any race when there are multiple
+                // executor threads.
+                ByteBuffer buffer = mByteBuffer;
+                mByteBuffer = null;
+                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("nativeStreamLock")
+    private boolean maybeSucceeded() {
+        if (mStreamState != STATE_READING_DONE || mWriteState != STATE_WRITING_DONE) {
+            return false;
+        }
+
+        mStreamState = STATE_SUCCESS;
+        Runnable task = new Runnable() {
+            public void run() {
+                synchronized (mNativeStreamLock) {
+                    if (isDone()) {
+                        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 (mStreamState != STATE_NOT_STARTED) {
+                throw new IllegalStateException("Stream is already started.");
+            }
+            try {
+                mNativeStream = nativeCreateBidirectionalStream(
+                        mRequestContext.getUrlRequestContextAdapter());
+                mRequestContext.onRequestStarted();
+                String headers[] = stringsFromHeaderList(mRequestHeaders);

[pauljensen, 2016/01/04 19:56:00]

no action required side-note:  interesting that you consolidated adding all
headers at once.  Helen made this change for response headers but not request
headers, see comment
https://code.google.com/p/chromium/issues/detail?id=536119#c4

[mef, 2016/01/04 22:27:07]

She didn't change request headers because we didn't think of a good way to
report an error in particular header.
I came up with this solution, which is a bit ugly, so I'm open for criticism.

+                // Non-zero startResult means an argument error.
+                int startResult = nativeStart(mNativeStream, mInitialUrl, mInitialPriority,
+                        mInitialMethod, headers, !doesMethodAllowWriteData(mInitialMethod));

[pauljensen, 2016/01/06 17:18:33]

Can we combine nativeStart() and nativeCreateBidirectionalStream()?

[mef, 2016/01/06 21:28:30]

We could, but then how do we report back faulty method or headers?

[pauljensen, 2016/01/07 02:56:34]

Yuck.  JNI is awful sometimes.  All we need is a pointer to an int...but that's
too much to ask.  There are a few ways but they're all awful:
http://stackoverflow.com/questions/29043872/android-jni-return-multiple-varia...
I guess the way we have it is fine then.

[mef, 2016/01/07 03:22:15]

Acknowledged.

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

Another idea is to throw the exception from native code:
    newExcCls = (*env)->FindClass(env, "java/lang/IllegalArgumentException");
    if (newExcCls == 0) /* Unable to find the new exception class, give up. */
      return;
    (*env)->ThrowNew(env, newExcCls, "Invalid HTTP method");
Probably not worth implementing this, but it would speed up the general case by
having only one JNI call.

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

Would it make sense to add a @CalledByNative metod onStartFailure(String
invalidMethod, String invalidHeader) that would post the error to an executor,
and combine create/start to return 0 in failure case?

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

I don't think there is any need to involve an Executor or more JNI.  WDYT about
my throw-from-native idea?  Frankly I'm fine leaving it the way it is for now.

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

sg, let's keep it this way and refactor both UrlRequest and BidirectionalStream
in consistent matter as separate CL.

+                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]);
+                }
+                mStreamState = 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 (mStreamState != STATE_WAITING_ON_READ) {
+                throw new IllegalStateException("Unexpected read attempt.");
+            }
+            if (isDone()) {
+                return;
+            }
+            mStreamState = 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.
+                mStreamState = 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 (isDone()) {
+                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 (isDone() || mStreamState == STATE_NOT_STARTED) {
+                return;
+            }
+            mStreamState = STATE_CANCELED;
+            destroyNativeStream(true);
+        }
+    }
+
+    @Override
+    public boolean isDone() {

[pauljensen, 2016/01/06 17:18:33]

This function is only called from within "synchronized (mNativeStreamLock) {" or
immediately prior to code that enters such a block.  I think the code could be
more performant if the synchronized statement here was removed, @GuardedBy was
added, and callsites immediately before entering the sychronized blocks were
moved into the synchronized blocks.  This will also make the code more
consistent with maybeSucceeded().

[mef, 2016/01/06 21:28:30]

It is also public API, so it has to be synchronized.

[pauljensen, 2016/01/07 02:56:34]

how about:
isDone() {
  synchronized (mNativeStreamLock) {
    isDoneLocked();
  }
}
@GuardedBy
isDoneLocked() {
  return mStreamState != STATE_NOT_STARTED && mNativeStream == 0;
}
and then change all the internal call sites to call isDoneLocked.
maybe this is too much micro-optimization.

[mef, 2016/01/07 03:22:15]

I think it is micro-optimization when lock is already acquired, but maybe worthy
when isDone() was called outside of synchronized section.
Anyway, Done. 

+        synchronized (mNativeStreamLock) {
+            return mStreamState != STATE_NOT_STARTED && mNativeStream == 0;
+        }
+    }
+
+    @SuppressWarnings("unused")
+    @CalledByNative
+    private void onRequestHeadersSent() {
+        Runnable task = new Runnable() {
+            public void run() {
+                synchronized (mNativeStreamLock) {
+                    if (isDone()) {
+                        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 (isDone()) {
+                        return;
+                    }
+                    mStreamState = 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,
+            long receivedBytesCount) {
+        mResponseInfo.setReceivedBytesCount(receivedBytesCount);
+        if (byteBuffer.position() != initialPosition) {
+            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) {
+        if (byteBuffer.position() != initialPosition) {

[pauljensen, 2016/01/04 19:56:00]

no action required side-note:  Kinda weird to me that we verify position()
hasn't changed, even though we're about to overwrite position().  I guess a
change to position() is a common indicator of aberrant behavior.  Seems like
verifying limit() hasn't changed might not be as good an indicator of aberrant
behavior but does a better job of verifying things are going to keep working.

[mef, 2016/01/04 22:27:07]

Good point, I will preserve and check limits as well.

[mef, 2016/01/06 21:28:30]

Done.

+            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 (isDone()) {
+                        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>>();

[pauljensen, 2016/01/06 17:18:33]

()->(headers.length/2)

[mef, 2016/01/06 21:28:30]

Done.

+        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);
+            mRequestContext.onRequestDestroyed();
+            mNativeStream = 0;

[pauljensen, 2016/01/06 17:18:33]

can we swap this line with the line above?  I'd rather decrease the lifetime of
dangling pointers

[mef, 2016/01/06 21:28:30]

Done.

+            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 (isDone()) {
+                return;
+            }
+            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 (isDone()) {
+                        return;
+                    }
+                    mStreamState = 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.cc.
+    private native long nativeCreateBidirectionalStream(long urlRequestContextAdapter);
+
+    @NativeClassQualifiedName("CronetBidirectionalStream")
+    private native int nativeStart(long nativePtr, String url, int priority, String method,
+            String[] headers, boolean endOfStream);
+
+    @NativeClassQualifiedName("CronetBidirectionalStream")
+    private native boolean nativeReadData(
+            long nativePtr, ByteBuffer byteBuffer, int position, int capacity);
+
+    @NativeClassQualifiedName("CronetBidirectionalStream")
+    private native boolean nativeWriteData(
+            long nativePtr, ByteBuffer byteBuffer, int position, int capacity, boolean endOfStream);
+
+    @NativeClassQualifiedName("CronetBidirectionalStream")
+    private native void nativeDestroy(long nativePtr, boolean sendOnCanceled);
+
+    @NativeClassQualifiedName("CronetBidirectionalStream")
+    private native String nativeGetNegotiatedProtocol(long nativePtr);
+}