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.

[pauljensen, 2016/01/19 16:03:40]

BidirectionalStream ->{@link BidirectionalStream}

[mef, 2016/01/20 15:37:40]

Done.

+ * All @CallByNative methods are called on native network thread

[pauljensen, 2016/01/19 16:03:40]

native->the native

[pauljensen, 2016/01/19 16:03:40]

CallByNative->CalledByNative

[mef, 2016/01/20 15:37:40]

Done.

[mef, 2016/01/20 15:37:40]

Done.

+ * and post tasks with callback calls onto Executor. Upon return from callback native

[pauljensen, 2016/01/19 16:03:40]

callback->callback,

[mef, 2016/01/20 15:37:40]

Hrm. I think done, but I'm not sure.

+ * stream is called on executor thread and posts native tasks to native network thread.

[pauljensen, 2016/01/19 16:03:40]

native->the native

[pauljensen, 2016/01/19 16:03:40]

executor->Executor

[mef, 2016/01/20 15:37:40]

Done.

[mef, 2016/01/20 15:37:40]

Done.

+ */
+@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.

[pauljensen, 2016/01/19 16:03:40]

nit: might want to update reference to "stream state"

[mef, 2016/01/20 15:37:40]

Done.

+     * There is one initial state - State.NOT_STARTED. There is one final state - State.SUCCESS,

[pauljensen, 2016/01/19 16:03:40]

Might want to mention that CANCELED and ERROR states are final, i.e. they cannot
be left.

[mef, 2016/01/20 15:37:40]

Done.

+     * 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. */

[pauljensen, 2016/01/19 16:03:40]

sent->being sent

[mef, 2016/01/20 15:37:40]

Done.

+        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 {
+                // 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 (maybeSucceeded()) {
+                            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 (maybeSucceeded()) {
+                            return;
+                        }
+                    } else {
+                        mWriteState = State.WAITING_ON_WRITE;
+                    }
+                }
+                mCallback.onWriteCompleted(CronetBidirectionalStream.this, mResponseInfo, buffer);
+            } catch (Exception e) {
+                onCallbackException(e);
+            }
+        }
+    }
+
+    CronetBidirectionalStream(CronetUrlRequestContext requestContext, long urlRequestContextAdapter,

[pauljensen, 2016/01/19 16:03:40]

remove unused urlRequestContextAdapater argument

[mef, 2016/01/20 15:37:40]

Done.

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

[pauljensen, 2016/01/19 16:03:40]

nit: add "Locked" suffix to maybeSucceeded

[mef, 2016/01/20 15:37:40]

Done.

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

[pauljensen, 2016/01/19 16:03:40]

replace this with Preconditions.checkHasRemaining(buffer);

[mef, 2016/01/20 15:37:40]

Done.

+                throw new IllegalArgumentException("ByteBuffer is already full.");
+            }

[pauljensen, 2016/01/19 16:03:40]

add Preconditions.checkDirect(buffer);

[mef, 2016/01/20 15:37:40]

Done.

+            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.");
+            }

[pauljensen, 2016/01/19 16:03:40]

add Preconditions.checkDirect(buffer);

[mef, 2016/01/20 15:37:40]

Done.

+            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 negotiatedProtocol, String[] headers) {
+        mResponseInfo =
+                prepareResponseInfoOnNetworkThread(httpStatusCode, negotiatedProtocol, 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()) {

[pauljensen, 2016/01/19 16:03:40]

do one of these for consistency:
1. initialPosition->byteBuffer.position()
2. byteBuffer.limit()->initialLimit

[mef, 2016/01/20 15:37:40]

Done.

+            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);

[pauljensen, 2016/01/19 16:03:40]

no action required nit: it'd be a performance boost if client's didn't have to
call read() just to get back an end-of-stream (EOS) signal.  I imagine the last
data packet sent has some EOS marker.  If onReadCompleted was called with a
boolean conveying this EOS marker, we could have mOnReadCompletedTask queue the
onSucceeded() callback (if writing is also done, which is likely).  I imagine
this saving some thread hops, but it may be ugly to implement so I assume we
shouldn't even think about it for this change.

[mef, 2016/01/20 15:37:40]

Acknowledged. This is also current contract with underlying net::
implementation.

+        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);

[pauljensen, 2016/01/19 16:03:40]

nit: any particular reason we do this:
  Runnable task = blah...;
  postTaskToExecutor(task);
versus:
  postTaskToExecutor(blah...);
It seems like more code and hence takes me longer to read.

[mef, 2016/01/20 15:37:40]

Historical reasons. :) Done.

+    }
+
+    @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 destroy the stream without invoking the callback.
+            mReadState = mWriteState = State.ERROR;

[pauljensen, 2016/01/19 16:03:40]

seems like this is unlocked access to mReadState and mWriteState?

[mef, 2016/01/20 15:37:40]

Done.

+            destroyNativeStream(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(
+            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 negotiatedProtocol, 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 = "";

[pauljensen, 2016/01/19 16:03:40]

there's no status line for BiDiStreams?  I would have expected there was if
there is a response code, but maybe HTTP2 got rid of this formality

[mef, 2016/01/20 15:37:40]

Correct. Http2 doesn't have special header for http status text, only status
code.

+        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;

[pauljensen, 2016/01/19 16:03:40]

shouldn't this throw IllegalArgumentException?

[mef, 2016/01/20 15:37:40]

Done.

+        }
+    }
+
+    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();
+            }
+        }
+    }
+
+    /**
+     * 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;
+            destroyNativeStream(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) {
+        Runnable task = new Runnable() {
+            public void run() {
+                failWithExceptionOnExecutor(exception);
+            }
+        };
+        postTaskToExecutor(task);
+    }
+
+    // Native methods are implemented in cronet_bidirectional_stream_adapter.cc.
+    private native long nativeCreateBidirectionalStream(long urlRequestContextAdapter);

[pauljensen, 2016/01/19 16:03:40]

static?

[mef, 2016/01/20 15:37:40]

Why? It gets |this| and stores as java owner for callbacks.

+
+    @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);
+}