Add HttpUrlConnection backed implementation of CronetEngine.

components/cronet/android/api/src/org/chromium/net/JavaUrlRequest.java

+// Copyright 2015 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+package org.chromium.net;
+
+import android.annotation.TargetApi;
+import android.net.TrafficStats;
+import android.os.Build;
+import android.util.Log;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.net.HttpURLConnection;
+import java.net.URI;
+import java.net.URL;
+import java.nio.ByteBuffer;
+import java.nio.channels.Channels;
+import java.nio.channels.ReadableByteChannel;
+import java.nio.channels.WritableByteChannel;
+import java.util.AbstractMap.SimpleEntry;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.TreeMap;
+import java.util.concurrent.Executor;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.atomic.AtomicReference;
+
+/**
+ * Pure java UrlRequest, backed by {@link HttpURLConnection}.
+ */
+@TargetApi(15)
+final class JavaUrlRequest implements UrlRequest {
+    private static final String X_ANDROID_SELECTED_TRANSPORT = "X-Android-Selected-Transport";
+    private static final String TAG = "JavaUrlConnection";
+    private static final int DEFAULT_UPLOAD_BUFFER_SIZE = 8192;
+    private static final int DEFAULT_CHUNK_LENGTH = DEFAULT_UPLOAD_BUFFER_SIZE;
+    public static final String USER_AGENT = "User-Agent";

[pauljensen, 2015/12/14 20:02:05]

why public?

[Charles, 2015/12/15 22:32:36]

Done.

+    private final AsyncUrlRequestCallback mCallbackAsync;
+    private final Executor mExecutor;
+    private final String mUserAgent;
+    private final Map<String, String> mRequestHeaders =
+            new TreeMap<>(String.CASE_INSENSITIVE_ORDER);

[pauljensen, 2015/12/14 20:02:05]

nit: I prefer hash tables as they're more computationally efficient...  probably
doesn't matter much for headers I suppose.

[Charles, 2015/12/15 22:32:36]

Doesn't matter much for headers, and I'd have to implement my own custom Map
class to provide the case insensitive behavior I need.

+    private final List<String> mUrlChain = new ArrayList<>();
+    /**
+     * This is the source of thread safety in this class - no other synchronization is performed.
+     * By compare-and-swapping from one state to another, we guarantee that operations aren't
+     * running concurrently. Only the winner of a CAS proceeds.
+     *
+     * <p>A caller can lose a CAS for three reasons - user error (two calls to read() without
+     * waiting for the read to succeed), runtime error (network code or user code throws an
+     * exception), or cancellation.
+     */
+    private final AtomicReference<State> mState = new AtomicReference<>(State.NOT_STARTED);
+    private final int mTrafficStatsTag;
+
+    /* These don't change with redirects */
+    private String mInitialMethod;
+    private UploadDataProvider mUploadDataProvider;
+    private Executor mUploadExecutor;
+    /**
+     *  Holds a subset of StatusValues - {@link State#STARTED} can represent

[pauljensen, 2015/12/14 20:02:05]

indented an extra space

[Charles, 2015/12/15 22:32:36]

Done.

+     * {@link Status#SENDING_REQUEST} or {@link Status#WAITING_FOR_RESPONSE}. While the distinction
+     * isn't needed to implement the logic in this class, it is needed to implement
+     * {@link #getStatus(StatusListener)}.
+     *
+     * <p>Concurrency notes - this value is not atomically updated with mState, so there is some
+     * risk

[pauljensen, 2015/12/14 20:02:05]

funny line wrapping

[Charles, 2015/12/15 22:32:36]

Done.

+     * that we'd get an inconsistent snapshot of both - however, it also happens that this value is
+     * only used with the STARTED state, so it's inconsequential.
+     */
+    @Status.StatusValues private volatile int mAdditionalStatusDetails = Status.INVALID;
+
+    /* These change with redirects. */
+    private String mCurrentUrl;
+    private ReadableByteChannel mResponseChannel;
+    private UrlResponseInfo mUrlResponseInfo;
+    private String mPendingRedirectUrl;
+    /**
+     * The happens-before edges created by the executor submission and AtomicReference setting are
+     * sufficient to guarantee the correct behavior of this field; however, this is an
+     * AtomicReference so that we can cleanly dispose of a new connection if we're cancelled during
+     * a redirect, which requires get-and-set semantics.
+     * */
+    private final AtomicReference<HttpURLConnection> mCurrentUrlConnection =
+            new AtomicReference<>();
+
+    /**
+     *             /- REDIRECTED <-\  /---- READING <----\
+     *             |               |  |                  |
+     *             \               /  \                  /
+     * NOT_STARTED ---> STARTED ---------> AWAITING_READ ---> COMPLETE
+     */
+    private enum State {
+        NOT_STARTED,
+        STARTED,
+        REDIRECTED,
+        AWAITING_READ,
+        READING,
+        ERROR,
+        COMPLETE,
+        CANCELLED,
+    }
+
+    /**
+     *

[pauljensen, 2015/12/14 20:02:06]

can we get rid of this blank line?

[Charles, 2015/12/15 22:32:36]

Done.

+     * @param executor The executor used for reading and writing from sockets
+     * @param userExecutor The executor used to dispatch to {@code callback}
+     */
+    JavaUrlRequest(Callback callback, final Executor executor, Executor userExecutor, String url,
+            String userAgent) {
+        if (url == null) {
+            throw new NullPointerException("URL is required");
+        }
+        if (callback == null) {
+            throw new NullPointerException("Listener is required");
+        }
+        if (executor == null) {
+            throw new NullPointerException("Executor is required");
+        }
+        if (userExecutor == null) {
+            throw new NullPointerException("Executor is required");
+        }
+        this.mCallbackAsync = new AsyncUrlRequestCallback(callback, userExecutor);
+        this.mTrafficStatsTag = TrafficStats.getThreadStatsTag();
+        this.mExecutor = new Executor() {
+            @Override
+            public void execute(final Runnable command) {
+                executor.execute(new Runnable() {
+                    @Override
+                    public void run() {
+                        int oldTag = TrafficStats.getThreadStatsTag();
+                        TrafficStats.setThreadStatsTag(mTrafficStatsTag);
+                        try {
+                            command.run();
+                        } finally {
+                            TrafficStats.setThreadStatsTag(oldTag);
+                        }
+                    }
+                });
+            }
+        };
+        this.mCurrentUrl = url;
+        this.mUserAgent = userAgent;
+    }
+
+    @Override
+    public void setHttpMethod(String method) {
+        checkNotStarted();
+        if (method == null) {
+            throw new NullPointerException("Method is required.");
+        }
+        if ("OPTIONS".equalsIgnoreCase(method) || "GET".equalsIgnoreCase(method)
+                || "HEAD".equalsIgnoreCase(method) || "POST".equalsIgnoreCase(method)
+                || "PUT".equalsIgnoreCase(method) || "DELETE".equalsIgnoreCase(method)
+                || "TRACE".equalsIgnoreCase(method) || "PATCH".equalsIgnoreCase(method)) {

[pauljensen, 2015/12/14 20:02:05]

CONNECT?

[Charles, 2015/12/15 22:32:36]

It's not mentioned as a supported method in the Android docs:
https://developer.android.com/reference/java/net/HttpURLConnection.html

And in getRequestMethod, it says "All possible methods of this HTTP
implementation is listed in the class definition"

+            mInitialMethod = method;
+        } else {
+            throw new IllegalArgumentException("Invalid http method " + method);
+        }
+    }
+
+    private void checkNotStarted() {
+        State state = mState.get();
+        if (state != State.NOT_STARTED) {
+            throw new IllegalStateException("Request is already started. State is: " + state);
+        }
+    }
+
+    @Override
+    public void addHeader(String header, String value) {
+        checkNotStarted();
+        if (!isValidHeaderName(header) || value.contains("\r\n")) {
+            throw new IllegalArgumentException("Invalid header " + header + "=" + value);
+        }
+        if (mRequestHeaders.containsKey(header)) {
+            mRequestHeaders.remove(header);
+        }
+        mRequestHeaders.put(header, value);
+    }
+
+    private boolean isValidHeaderName(String header) {
+        for (int i = 0; i < header.length(); i++) {
+            char c = header.charAt(i);
+            switch (c) {
+                case '(':
+                case ')':
+                case '<':
+                case '>':
+                case '@':
+                case ',':
+                case ';':
+                case ':':
+                case '\\':
+                case '\'':
+                case '/':
+                case '[':
+                case ']':
+                case '?':
+                case '=':
+                case '{':
+                case '}':
+                    return false;
+                default: {
+                    if (Character.isISOControl(c) || Character.isWhitespace(c)) {
+                        return false;
+                    }
+                }
+            }
+        }
+        return true;
+    }
+
+    @Override
+    public void setUploadDataProvider(UploadDataProvider uploadDataProvider, Executor executor) {
+        if (uploadDataProvider == null) {
+            throw new NullPointerException("Invalid UploadDataProvider.");
+        }
+        if (!mRequestHeaders.containsKey("Content-Type")) {
+            throw new IllegalArgumentException(
+                    "Requests with upload data must have a Content-Type.");
+        }
+        checkNotStarted();
+        if (mInitialMethod == null) {
+            mInitialMethod = "POST";

[pauljensen, 2015/12/14 20:02:06]

nit: I like calling setHttpMethod() here in case it does something more than
just setting mInitialMethod, but this is probably just a personal preference. 
Either way is fine.

[Charles, 2015/12/15 22:32:36]

I don't create the HttpUrlConnection instance to call this on until start() is
called, and I don't want to move that earlier, when it's happening on an
arbitrary thread.

[pauljensen, 2015/12/21 19:58:44]

I was talking about JavaUrlRequest.setHttpMethod() which doesn't create the
HttpUrlConnection.

+        }
+        this.mUploadDataProvider = uploadDataProvider;
+        this.mUploadExecutor = executor;
+    }
+
+    private final class OutputStreamDataSink implements UploadDataSink {
+        private final Executor mUserExecutor;
+        private final Executor mExecutor;
+        private final HttpURLConnection mUrlConnection;
+        private WritableByteChannel mOutputChannel;
+        private final UploadDataProvider mUploadProvider;
+        private ByteBuffer mBuffer;
+        /** This holds the total bytes to send (the content-length) */

[pauljensen, 2015/12/14 20:02:05]

-1 has a special meaning, correct?

[Charles, 2015/12/15 22:32:36]

Done.

+        private long mTotalBytes;
+        /** This holds the bytes written so far */
+        private long mWrittenBytes = 0;
+
+        private OutputStreamDataSink(final Executor userExecutor, Executor executor,
+                HttpURLConnection urlConnection, UploadDataProvider provider) {
+            this.mUserExecutor = new Executor() {
+                @Override
+                public void execute(Runnable runnable) {
+                    try {
+                        userExecutor.execute(runnable);
+                    } catch (RejectedExecutionException e) {
+                        enterUploadErrorState(e);
+                    }
+                }
+            };
+            this.mExecutor = executor;
+            this.mUrlConnection = urlConnection;
+            this.mUploadProvider = provider;
+        }
+
+        @Override
+        public void onReadSucceeded(final boolean finalChunk) {
+            mExecutor.execute(errorSetting(State.STARTED, new CheckedRunnable() {
+                @Override
+                public void run() throws Exception {
+                    mBuffer.flip();
+                    while (mBuffer.hasRemaining()) {
+                        mWrittenBytes += mOutputChannel.write(mBuffer);
+                    }
+                    if (mWrittenBytes < mTotalBytes || (mTotalBytes == -1 && !finalChunk)) {
+                        mBuffer.clear();
+                        mUserExecutor.execute(uploadErrorSetting(new CheckedRunnable() {
+                            @Override
+                            public void run() throws Exception {
+                                mUploadProvider.read(OutputStreamDataSink.this, mBuffer);
+                            }
+                        }));
+                    } else if (mTotalBytes == -1 && finalChunk) {
+                        finish();
+                    } else if (mTotalBytes == mWrittenBytes) {
+                        finish();
+                    } else {
+                        throw new IllegalStateException("Wrote more bytes than were available");
+                    }
+                }
+            }));
+        }
+
+        @Override
+        public void onReadError(Exception exception) {
+            enterUploadErrorState(exception);
+        }
+
+        @Override
+        public void onRewindSucceeded() {
+            startRead();
+        }
+
+        private void startRead() {
+            mUserExecutor.execute(uploadErrorSetting(new CheckedRunnable() {
+                @Override
+                public void run() throws Exception {
+                    if (mOutputChannel == null) {
+                        mAdditionalStatusDetails = Status.CONNECTING;
+                        mUrlConnection.connect();
+                        mAdditionalStatusDetails = Status.SENDING_REQUEST;
+                        mOutputChannel = Channels.newChannel(mUrlConnection.getOutputStream());
+                    }
+                    mUploadProvider.read(OutputStreamDataSink.this, mBuffer);
+                }
+            }));
+        }
+
+        @Override
+        public void onRewindError(Exception exception) {
+            enterUploadErrorState(exception);
+        }
+
+        private void finish() throws IOException {
+            if (mOutputChannel != null) {
+                mOutputChannel.close();
+            }
+            mAdditionalStatusDetails = Status.WAITING_FOR_RESPONSE;
+            fireGetHeaders();
+        }
+
+        private void start(final boolean firstTime) {
+            mUserExecutor.execute(uploadErrorSetting(new CheckedRunnable() {
+                @Override
+                public void run() throws Exception {
+                    mTotalBytes = mUploadProvider.getLength();
+                    if (mTotalBytes == 0) {
+                        finish();
+                    } else {
+                        // If we know how much data we have to upload, and it's small, we can save
+                        // memory by allocating a reasonably sized buffer to read into.
+                        if (mTotalBytes > 0 && mTotalBytes < DEFAULT_UPLOAD_BUFFER_SIZE) {
+                            mBuffer = ByteBuffer.allocateDirect((int) mTotalBytes + 1);

[pauljensen, 2015/12/14 20:02:05]

Why +1 ?

[Charles, 2015/12/15 22:32:36]

Commented.

+                        } else {
+                            mBuffer = ByteBuffer.allocateDirect(DEFAULT_UPLOAD_BUFFER_SIZE);
+                        }
+
+                        if (mTotalBytes > 0 && mTotalBytes <= Integer.MAX_VALUE) {
+                            mUrlConnection.setFixedLengthStreamingMode((int) mTotalBytes);
+                        } else if (mTotalBytes > Integer.MAX_VALUE && Build.VERSION.SDK_INT >= 19) {

[pauljensen, 2015/12/14 20:02:05]

I'd prefer a constant is used rather than an integer literal.  I imagine
dev.android.com lists the integer literal next to each API for brevity.

[Charles, 2015/12/15 22:32:36]

Done.

+                            mUrlConnection.setFixedLengthStreamingMode(mTotalBytes);
+                        } else {
+                            // If we know the length, but we're running pre-kitkat and it's larger
+                            // than an int can hold, we have to use chunked - otherwise we'll end up
+                            // buffering the whole response in memory.
+                            mUrlConnection.setChunkedStreamingMode(DEFAULT_CHUNK_LENGTH);
+                        }
+                        if (firstTime) {
+                            startRead();
+                        } else {
+                            mUploadProvider.rewind(OutputStreamDataSink.this);
+                        }
+                    }
+                }
+            }));
+        }
+    }
+
+    @Override
+    public void start() {
+        mAdditionalStatusDetails = Status.CONNECTING;
+        transitionStates(State.NOT_STARTED, State.STARTED);
+        mUrlChain.add(mCurrentUrl);
+        fireOpenConnection();
+    }
+
+    private void enterErrorState(State previousState, final UrlRequestException error) {
+        if (mState.compareAndSet(previousState, State.ERROR)) {
+            fireDisconnect();
+            mCallbackAsync.onFailed(mUrlResponseInfo, error);
+        }
+    }
+
+    /** Ends the reqeust with an error, caused by an exception thrown from user code. */
+    private void enterUserErrorState(State previousState, final Throwable error) {
+        enterErrorState(previousState, new UrlRequestException("User Error", error));
+    }
+
+    /** Ends the requst with an error, caused by an exception thrown from user code. */
+    private void enterUploadErrorState(final Throwable error) {
+        enterErrorState(State.STARTED,
+                new UrlRequestException("Exception received from UploadDataProvider", error));
+    }
+
+    private void enterCronetErrorState(State previousState, final Throwable error) {
+        // TODO(clm) mapping from Java exception (UnknownHostException, for example) to net error
+        // code goes here.
+        enterErrorState(previousState, new UrlRequestException("System error", error));
+    }
+
+    /**
+     * Atomically swaps from the expected state to a new state. If the swap fails, and it's not
+     * due to an earlier error or cancellation, throws an exception.
+     */
+    private void transitionStates(State expected, State newState) {
+        if (!mState.compareAndSet(expected, newState)) {
+            State state = mState.get();
+            if (!(state == State.CANCELLED || state == State.ERROR)) {

[pauljensen, 2015/12/14 20:02:05]

Do we really want to be continuing along if the request is cancelled or failed?
For example followRedirect() will set mCurrentUrl to mPendingRedirectUrl which
is probably null.  Then it will call fireOpenConnection which will probably NPE
on some worker thread.

Seems like transitionStates() should return the CAS result and callers should
bail out (i.e. return early) when false.

[Charles, 2015/12/15 22:32:36]

Total error on my part. You're correct that we should be returning early. I
implemented it with a callback, so that there's no way to forget.

+                throw new IllegalStateException(
+                        "Invalid state transition - expected " + expected + " but was " + state);
+            }
+        }
+    }
+
+    @Override
+    public void followRedirect() {
+        transitionStates(State.REDIRECTED, State.STARTED);
+        mCurrentUrl = mPendingRedirectUrl;
+        mPendingRedirectUrl = null;
+        fireOpenConnection();
+    }
+
+    private void fireGetHeaders() {
+        mExecutor.execute(errorSetting(State.STARTED, new CheckedRunnable() {
+            @Override
+            public void run() throws Exception {
+                HttpURLConnection connection = mCurrentUrlConnection.get();
+                if (connection == null) {
+                    return; // We've been cancelled
+                }
+                final List<Map.Entry<String, String>> headerList = new ArrayList<>();
+                String selectedTransport = "http/1.1";

[pauljensen, 2015/12/14 20:02:05]

I'm not sure it's a good idea to make this default to a believable value; this
is likely to cover up problems (like X-Android header names changing).

[Charles, 2015/12/15 22:32:36]

All of that behavior is undocumented and best-effort anyway. What would you like
the behavior to be?

[pauljensen, 2015/12/21 19:58:44]

The API is documented like so:
  "Returns an empty string if no protocol was negotiated, the protocol is not
known, or when using plain HTTP or HTTPS."
and considering that the system's HttpURLConnection impl doesn't support QUIC or
SPDY (disabled by Android), I think we may always want to just return an empty
string always, not just in the default case.

+                String headerKey;
+                for (int i = 0; (headerKey = connection.getHeaderFieldKey(i)) != null; i++) {
+                    if (X_ANDROID_SELECTED_TRANSPORT.equalsIgnoreCase(headerKey)) {
+                        selectedTransport = connection.getHeaderField(i);
+                    }
+                    if (!headerKey.startsWith("X-Android")) {
+                        headerList.add(new SimpleEntry<>(headerKey, connection.getHeaderField(i)));
+                    }
+                }
+
+                int responseCode = connection.getResponseCode();
+                // Important to copy the list here, because although we never concurrently modify
+                // the list ourselves, user code might iterate over it while we're redirecting, and
+                // that would throw ConcurrentModificationException.
+                mUrlResponseInfo = new UrlResponseInfo(new ArrayList<>(mUrlChain), responseCode,
+                        connection.getResponseMessage(), Collections.unmodifiableList(headerList),
+                        false, selectedTransport, "");
+                // TODO(clm) actual redirect handling? post -> get and whatnot?
+                if (responseCode >= 300 && responseCode < 400) {
+                    fireRedirectReceived(mUrlResponseInfo.getAllHeaders());
+                } else if (responseCode >= 400) {
+                    mResponseChannel = InputStreamChannel.wrap(connection.getErrorStream());
+                    mCallbackAsync.onResponseStarted(mUrlResponseInfo);
+                } else {
+                    mResponseChannel = InputStreamChannel.wrap(connection.getInputStream());
+                    mCallbackAsync.onResponseStarted(mUrlResponseInfo);
+                }
+            }
+        }));
+    }
+
+    private void fireRedirectReceived(Map<String, List<String>> headerFields) {
+        if (mState.compareAndSet(State.STARTED, State.REDIRECTED)) {
+            mPendingRedirectUrl =
+                    URI.create(mCurrentUrl).resolve(headerFields.get("location").get(0)).toString();
+            mUrlChain.add(mPendingRedirectUrl);
+            mCallbackAsync.onRedirectReceived(mUrlResponseInfo, mPendingRedirectUrl);
+        }
+    }
+
+    private static final class CaseInsensitiveString {
+        private int mHashCode;
+        private final String mString;
+
+        private CaseInsensitiveString(String string) {
+            if (string == null) {
+                throw new NullPointerException();
+            }
+            mString = string;
+        }
+
+        @Override
+        public boolean equals(Object o) {
+            return o instanceof CaseInsensitiveString
+                    && mString.equalsIgnoreCase(((CaseInsensitiveString) o).mString);
+        }
+
+        @Override
+        public int hashCode() {
+            if (mHashCode == 0) {
+                mHashCode = mString.toLowerCase(Locale.US).hashCode();
+            }
+            return mHashCode;
+        }
+
+        String getString() {
+            return mString;
+        }
+    }
+
+    private void fireOpenConnection() {
+        mExecutor.execute(errorSetting(State.STARTED, new CheckedRunnable() {
+            @Override
+            public void run() throws Exception {
+                // If we're cancelled, then our old connection will be disconnected for us and
+                // we shouldn't open a new one.
+                if (mState.get() == State.CANCELLED) {
+                    return;
+                }
+
+                final URL url = new URL(mCurrentUrl);
+                HttpURLConnection newConnection = (HttpURLConnection) url.openConnection();
+                HttpURLConnection oldConnection = mCurrentUrlConnection.getAndSet(newConnection);
+                if (oldConnection != null) {
+                    oldConnection.disconnect();
+                }
+                newConnection.setInstanceFollowRedirects(false);
+                if (!mRequestHeaders.containsKey(USER_AGENT)) {
+                    mRequestHeaders.put(USER_AGENT, mUserAgent);
+                }
+                for (Map.Entry<String, String> entry : mRequestHeaders.entrySet()) {
+                    newConnection.setRequestProperty(entry.getKey(), entry.getValue());
+                }
+                if (mInitialMethod == null) {
+                    mInitialMethod = "GET";
+                }
+                newConnection.setRequestMethod(mInitialMethod);
+                if (mUploadDataProvider != null) {
+                    OutputStreamDataSink dataSink = new OutputStreamDataSink(
+                            mUploadExecutor, mExecutor, newConnection, mUploadDataProvider);
+                    dataSink.start(mUrlChain.size() == 1);
+                } else {
+                    mAdditionalStatusDetails = Status.CONNECTING;
+                    newConnection.connect();
+                    mAdditionalStatusDetails = Status.WAITING_FOR_RESPONSE;
+                    fireGetHeaders();
+                }
+            }
+        }));
+    }
+
+    @Override
+    public void read(final ByteBuffer buffer) {
+        Preconditions.checkDirect(buffer);
+        if (!(buffer.capacity() - buffer.position() > 0)) {
+            throw new IllegalArgumentException("ByteBuffer is already full.");
+        }
+        transitionStates(State.AWAITING_READ, State.READING);
+        mExecutor.execute(errorSetting(State.READING, new CheckedRunnable() {
+            @Override
+            public void run() throws IOException {
+                int oldPosition = buffer.position();
+                buffer.limit(buffer.capacity());
+                int read = mResponseChannel.read(buffer);
+                buffer.limit(buffer.position());
+                buffer.position(oldPosition);
+                processReadResult(read, buffer);
+            }
+        }));
+    }
+
+    private Runnable errorSetting(final State expectedState, final CheckedRunnable delegate) {
+        return new Runnable() {
+            @Override
+            public void run() {
+                try {
+                    delegate.run();
+                } catch (Throwable t) {
+                    enterCronetErrorState(expectedState, t);
+                }
+            }
+        };
+    }
+
+    private Runnable userErrorSetting(final State expectedState, final CheckedRunnable delegate) {
+        return new Runnable() {
+            @Override
+            public void run() {
+                try {
+                    delegate.run();
+                } catch (Throwable t) {
+                    enterUserErrorState(expectedState, t);
+                }
+            }
+        };
+    }
+
+    private Runnable uploadErrorSetting(final CheckedRunnable delegate) {
+        return new Runnable() {
+            @Override
+            public void run() {
+                try {
+                    delegate.run();
+                } catch (Throwable t) {
+                    enterUploadErrorState(t);
+                }
+            }
+        };
+    }
+
+    private interface CheckedRunnable { void run() throws Exception; }
+
+    @Override
+    public void readNew(final ByteBuffer buffer) {
+        Preconditions.checkDirect(buffer);
+        Preconditions.checkHasRemaining(buffer);
+        transitionStates(State.AWAITING_READ, State.READING);
+        mExecutor.execute(errorSetting(State.READING, new CheckedRunnable() {
+            @Override
+            public void run() throws Exception {
+                int read = mResponseChannel.read(buffer);
+                processReadResult(read, buffer);
+            }
+        }));
+    }
+
+    private void processReadResult(int read, final ByteBuffer buffer) throws IOException {
+        if (read != -1) {
+            mCallbackAsync.onReadCompleted(mUrlResponseInfo, buffer);
+        } else {
+            mResponseChannel.close();
+            if (mState.compareAndSet(State.READING, State.COMPLETE)) {
+                fireDisconnect();

[pauljensen, 2015/12/14 20:02:05]

nit: seems a little silly that fireDisconnect() posts a task to mExecutor, from
processReadResult() which is already posted to mExecutor.

[Charles, 2015/12/15 22:32:36]

If it didn't, we'd delay the onSucceeded callback until disconnect() returned. I
suppose I could call it after, but it's easier to reason about if we always post
to the executor for disconnection, and the overhead isn't high.

+                mCallbackAsync.onSucceeded(mUrlResponseInfo);
+            }
+        }
+    }
+
+    private void fireDisconnect() {
+        final HttpURLConnection connection = mCurrentUrlConnection.getAndSet(null);
+        mExecutor.execute(new Runnable() {
+            @Override
+            public void run() {
+                if (connection != null) {

[pauljensen, 2015/12/14 20:02:05]

seems like this null check could be before the execute() call

[Charles, 2015/12/15 22:32:36]

Done.

+                    connection.disconnect();
+                }
+            }
+        });
+    }
+
+    @Override
+    public void cancel() {
+        State oldState = mState.getAndSet(State.CANCELLED);
+        switch (oldState) {
+            // We've just scheduled some user code to run. When they perform their next operation,
+            // they'll observe it and fail. However, if user code is cancelling in response to one
+            // of these callbacks, we'll never actually cancel!
+            // TODO(clm) figure out if it's possible to avoid concurrency in user callbacks.
+            case REDIRECTED:
+            case AWAITING_READ:
+
+            // User code is waiting on us - cancel away!
+            case STARTED:
+            case READING:
+                fireDisconnect();
+                mCallbackAsync.onCanceled(mUrlResponseInfo);
+                break;
+            // The rest are all termination cases - we're too late to cancel.
+            case ERROR:
+            case COMPLETE:
+            case CANCELLED:
+                break;
+        }
+    }
+
+    @Override
+    public boolean isDone() {
+        State state = mState.get();
+        return state == State.COMPLETE | state == State.ERROR | state == State.CANCELLED;
+    }
+
+    @Override
+    public void disableCache() {
+        // We have no cache
+    }
+
+    @Override
+    public void getStatus(StatusListener listener) {
+        State state = mState.get();
+        int extraStatus = this.mAdditionalStatusDetails;
+
+        @Status.StatusValues final int status;
+        switch (state) {
+            case ERROR:
+            case COMPLETE:
+            case CANCELLED:
+            case NOT_STARTED:
+                status = Status.INVALID;
+                break;
+            case STARTED:
+                status = extraStatus;
+                break;
+            case REDIRECTED:
+            case AWAITING_READ:
+                status = Status.IDLE;
+                break;
+            case READING:
+                status = Status.READING_RESPONSE;
+                break;
+            default:
+                throw new IllegalStateException("Switch is exhaustive: " + state);
+        }
+
+        mCallbackAsync.sendStatus(listener, status);
+    }
+
+    /** This wrapper ensures that callbacks are always called on the correct executor */
+    private final class AsyncUrlRequestCallback {
+        private final UrlRequest.Callback mCallback;
+        private final Executor mUserExecutor;
+
+        private AsyncUrlRequestCallback(Callback callback, final Executor userExecutor) {
+            this.mCallback = callback;
+            this.mUserExecutor = userExecutor;
+        }
+
+        void sendStatus(final StatusListener listener, final int status) {
+            mUserExecutor.execute(new Runnable() {
+                @Override
+                public void run() {
+                    listener.onStatus(status);
+                }
+            });
+        }
+
+        void execute(State currentState, CheckedRunnable runnable) {

[pauljensen, 2015/12/14 20:02:05]

private?

[Charles, 2015/12/15 22:32:36]

No point in adding private to methods on private inner classes. I've removed
private from all the variables and methods too.

+            try {
+                mUserExecutor.execute(userErrorSetting(currentState, runnable));
+            } catch (RejectedExecutionException e) {
+                enterUserErrorState(currentState, e);
+            }
+        }
+
+        void onRedirectReceived(final UrlResponseInfo info, final String newLocationUrl) {
+            execute(State.REDIRECTED, new CheckedRunnable() {
+                @Override
+                public void run() throws Exception {
+                    mCallback.onRedirectReceived(JavaUrlRequest.this, info, newLocationUrl);
+                }
+            });
+        }
+
+        void onResponseStarted(UrlResponseInfo info) {
+            execute(State.AWAITING_READ, new CheckedRunnable() {
+                @Override
+                public void run() {
+                    if (mState.compareAndSet(State.STARTED, State.AWAITING_READ)) {
+                        mCallback.onResponseStarted(JavaUrlRequest.this, mUrlResponseInfo);
+                    }
+                }
+            });
+        }
+
+        void onReadCompleted(final UrlResponseInfo info, final ByteBuffer byteBuffer) {
+            execute(State.AWAITING_READ, new CheckedRunnable() {
+                @Override
+                public void run() {
+                    if (mState.compareAndSet(State.READING, State.AWAITING_READ)) {
+                        mCallback.onReadCompleted(JavaUrlRequest.this, info, byteBuffer);
+                    }
+                }
+            });
+        }
+
+        void onCanceled(final UrlResponseInfo info) {
+            closeQuietly(mResponseChannel);
+            mUserExecutor.execute(new Runnable() {
+                @Override
+                public void run() {
+                    try {
+                        mCallback.onCanceled(JavaUrlRequest.this, info);
+                    } catch (Exception exception) {
+                        Log.e(TAG, "Exception in onCanceled method", exception);
+                    }
+                }
+            });
+        }
+
+        void onSucceeded(final UrlResponseInfo info) {
+            mUserExecutor.execute(new Runnable() {
+                @Override
+                public void run() {
+                    try {
+                        mCallback.onSucceeded(JavaUrlRequest.this, info);
+                    } catch (Exception exception) {
+                        Log.e(TAG, "Exception in onSucceeded method", exception);
+                    }
+                }
+            });
+        }
+
+        void onFailed(final UrlResponseInfo urlResponseInfo, final UrlRequestException e) {
+            closeQuietly(mResponseChannel);
+            mUserExecutor.execute(new Runnable() {
+                @Override
+                public void run() {
+                    try {
+                        mCallback.onFailed(JavaUrlRequest.this, urlResponseInfo, e);
+                    } catch (Exception exception) {
+                        Log.e(TAG, "Exception in onFailed method", exception);
+                    }
+                }
+            });
+        }
+    }
+
+    private static void closeQuietly(Closeable closeable) {
+        if (closeable == null) {
+            return;
+        }
+        try {
+            closeable.close();
+        } catch (IOException e) {
+            e.printStackTrace();
+        }
+    }
+}