xref: /external/downloader/src/main/java/com/google/android/downloader/Downloader.java

// Copyright 2021 The Android Open Source Project
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//      http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package com.google.android.downloader;

import static com.google.common.base.Preconditions.checkArgument;
import static com.google.common.base.Preconditions.checkNotNull;
import static com.google.common.base.Preconditions.checkState;
import static com.google.common.base.Predicates.instanceOf;
import static com.google.common.base.Throwables.getCausalChain;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
import static java.util.concurrent.TimeUnit.SECONDS;

import com.google.auto.value.AutoValue;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Iterables;
import com.google.common.net.HttpHeaders;
import com.google.common.util.concurrent.ClosingFuture;
import com.google.common.util.concurrent.FluentFuture;
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.ListenableFutureTask;
import com.google.common.util.concurrent.MoreExecutors;
import com.google.errorprone.annotations.CheckReturnValue;
import com.google.errorprone.annotations.FormatMethod;
import com.google.errorprone.annotations.FormatString;
import com.google.errorprone.annotations.concurrent.GuardedBy;
import java.io.IOException;
import java.net.HttpURLConnection;
import java.net.URI;
import java.nio.channels.WritableByteChannel;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.IdentityHashMap;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Queue;
import java.util.TimeZone;
import java.util.concurrent.CancellationException;
import java.util.concurrent.Executor;
import java.util.concurrent.TimeoutException;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import javax.annotation.Nullable;

/**
 * In-process download system. Provides a light-weight mechanism to download resources over the
 * network. Downloader includes the following features:
 *
 * <ul>
 *   <li>Configurable choice of network stack, with several defaults available out of the box.
 *   <li>Fully asynchronous behavior, allowing downloads to progress and complete without blocking
 *       (assuming the underlying network stack can avoid blocking).
 *   <li>Support for HTTP range requests, allowing partial downloads to resume mid-way through, and
 *       avoiding redownloading of fully-downloaded requests.
 *   <li>Detection of network interruptions, and a configurable way to retry requests that have
 *       failed after losing connectivity.
 * </ul>
 *
 * <p>Note that because this library performs downloads in-process, it is subject to having
 * downloads stall or abort when the app is suspended or killed. Download requests only live in
 * memory, and thus are lost when the process ends. Library users should persist the set of
 * downloads to be executed in persistent storage, such as a SQLite database, and run the downloads
 * in the context of a persistent operation (either a foreground service or via Android's {@link
 * android.app.job.JobScheduler} mechanism).
 *
 * <p>This is intended as a functional (but not drop-in) replacement for Android's {@link
 * android.app.DownloadManager}, which has a number of issues:
 *
 * <ul>
 *   <li>It relies on the network stack built into Android, and thus cannot be patched to receive
 *       updates. For older versions of Android, that means the DownloadManager is vulnerable to
 *       issues in the network stack, such as b/18432707, which can result in MITM attacks.
 *   <li>When downloading to external storage on older versions of Android (via {@link
 *       android.app.DownloadManager.Request#setDestinationInExternalFilesDir} or {@link
 *       android.app.DownloadManager.Request#setDestinationInExternalPublicDir}), downloads may
 *       exceed the maximum size of the download directory (see b/22605830).
 *   <li>Downloads may pause and never resume again without external interference (b/18110151)
 *   <li>The DownloadManager may lose track of some files (b/18265497)
 * </ul>
 *
 * <p>This library mitigates these issues by performing downloads in-process over the app-provided
 * network stack, thus ensuring that an up-to-date network stack is used, and handing off storage
 * management directly to the app without any additional restrictions.
 */
public class Downloader {
  @VisibleForTesting static final int HTTP_RANGE_NOT_SATISFIABLE = 416;
  @VisibleForTesting static final int HTTP_PARTIAL_CONTENT = 206;

  private static final ImmutableSet<String> SCHEMES_REQUIRING_CONNECTIVITY =
      ImmutableSet.of("http", "https");

  private static final Pattern CONTENT_RANGE_HEADER_PATTERN =
      Pattern.compile("bytes (\\d+)-(\\d+)/(\\d+|\\*)");

  @GuardedBy("SIMPLE_DATE_FORMAT_LOCK")
  private static final SimpleDateFormat RFC_1123_FORMATTER;

  private static final Object SIMPLE_DATE_FORMAT_LOCK = new Object();

  static {
    synchronized (SIMPLE_DATE_FORMAT_LOCK) {
      RFC_1123_FORMATTER = new SimpleDateFormat("EEE, dd MMM yyyy HH:mm:ss 'GMT'", Locale.US);
      RFC_1123_FORMATTER.setTimeZone(TimeZone.getTimeZone("UTC"));
    }
  }

  /** Builder for configuring and constructing an instance of the Downloader. */
  public static class Builder {
    private final Map<String, UrlEngine> urlEngineMap = new HashMap<>();
    private Executor ioExecutor;
    private DownloaderLogger logger;
    private ConnectivityHandler connectivityHandler;
    private int maxConcurrentDownloads = 3;

    /**
     * Creates a downloader builder.
     *
     * <p>Note that all parameters are required except for {@link #withMaxConcurrentDownloads}.
     */
    public Builder() {}

    /**
     * Specifies the executor to use internally for I/O.
     *
     * <p>I/O operations can block so don't use a direct executor or one that runs on the main
     * thread.
     */
    public Builder withIOExecutor(Executor ioExecutor) {
      this.ioExecutor = ioExecutor;
      return this;
    }

    /**
     * Sets the {@link ConnectivityHandler} to use in order to determine if connectivity is
     * sufficient, and if not, how to handle it.
     */
    public Builder withConnectivityHandler(ConnectivityHandler connectivityHandler) {
      this.connectivityHandler = connectivityHandler;
      return this;
    }

    /**
     * Limits the number of downloads in flight at a time. If a download request arrives that would
     * exceed this limit, it will be queued until one already in flight completes. Beware that a
     * download that is waiting for connectivity requirements is still considered to be in flight,
     * so it is possible to saturate the downloader with requests waiting on connectivity
     * requirements if the number of concurrent downloads isn't set high enough.
     *
     * <p>Note that other factors might restrict download concurrency even further, for instance the
     * number of threads on the I/O executor when using a blocking engine.
     */
    public Builder withMaxConcurrentDownloads(int maxConcurrentDownloads) {
      checkArgument(maxConcurrentDownloads > 0);
      this.maxConcurrentDownloads = maxConcurrentDownloads;
      return this;
    }

    /**
     * Adds an {@link UrlEngine} to handle network requests for the given URL scheme. Note that the
     * engine must support the provided scheme, and only one engine may ever be registered for a
     * specific URL scheme. An {@link IllegalArgumentException} will be thrown if the engine doesn't
     * support the scheme or if an engine is already registered for the scheme.
     */
    public Builder addUrlEngine(String scheme, UrlEngine urlEngine) {
      checkArgument(
          urlEngine.supportedSchemes().contains(scheme),
          "Provided UrlEngine must support URL scheme: %s",
          scheme);
      checkArgument(
          !urlEngineMap.containsKey(scheme),
          "Requested scheme already has a UrlEngine registered: %s",
          scheme);
      urlEngineMap.put(scheme, urlEngine);
      return this;
    }

    /**
     * Adds an {@link UrlEngine} to handle network requests for the given URL schemes. Note that the
     * engine must support the provided schemes, and only one engine may ever be registered for a
     * specific URL scheme. An error will be thrown if the engine doesn't support the schemes or if
     * an engine is already registered for the schemes.
     */
    public Builder addUrlEngine(Iterable<String> schemes, UrlEngine urlEngine) {
      for (String scheme : schemes) {
        addUrlEngine(scheme, urlEngine);
      }
      return this;
    }

    /** Sets the {@link DownloaderLogger} to use for logging in this downloader instance. */
    public Builder withLogger(DownloaderLogger logger) {
      this.logger = logger;
      return this;
    }

    public Downloader build() {
      return new Downloader(this);
    }
  }

  /**
   * Value class for capturing a state snapshot of the downloader, for use in the state callbacks
   * that can be registered via {@link #registerStateChangeCallback}.
   */
  @AutoValue
  public abstract static class State {
    /**
     * Returns the current number of downloads currently in flight, i.e. the number of downloads
     * that are concurrently executing, or attempting to make progress on their underlying network
     * stack. Note that this number should be in the range of [0, maxConcurrentDownloads), as
     * configured by {@link Builder#withMaxConcurrentDownloads}.
     */
    public abstract int getNumDownloadsInFlight();

    /**
     * Returns the number of downloads that have been requested via {@link #execute} but not yet
     * started. They may not have been started due to internal asynchronous code, due to waiting on
     * connectivity requirements, or due to the limit enforced by {@code maxConcurrentDownloads}.
     */
    public abstract int getNumQueuedDownloads();

    /**
     * Returns the current number of downloads that are waiting for sufficient connectivity
     * conditions to be met. These downloads will start running once the device network conditions
     * change (e.g. connects to WiFi), if there is enough space as determined by {@code
     * maxConcurrentDownloads}.
     *
     * <p>Note that this number should be in the range of [0, numQueuedDownloads], as a download
     * pending connectivity is necessarily queued. However, a download may also be queued due to
     * {@code maxConcurrentDownloads}.
     */
    public abstract int getNumDownloadsPendingConnectivity();

    /** Creates an instance of the state object. Internal-only. */
    @VisibleForTesting
    public static State create(
        int numDownloadsInFlight, int numQueuedDownloads, int numDownloadsPendingConnectivity) {
      return new AutoValue_Downloader_State(
          numDownloadsInFlight, numQueuedDownloads, numDownloadsPendingConnectivity);
    }
  }

  /**
   * Functional callback interface for observing changes to Downloader {@link State}. Used with
   * {@link #registerStateChangeCallback} and {@link #unregisterStateChangeCallback}.
   *
   * <p>Note that the callbacks could just use {@code java.util.function.Consumer} in contexts that
   * support the Java 8 SDK, and {@code com.google.common.base.Receiver} in contexts that don't but
   * have access to Guava APIs that aren't made public.
   */
  public interface StateChangeCallback {
    void onStateChange(State state);
  }

  private final ImmutableMap<String, UrlEngine> urlEngineMap;
  private final Executor ioExecutor;
  private final DownloaderLogger logger;
  private final ConnectivityHandler connectivityHandler;
  private final int maxConcurrentDownloads;

  @GuardedBy("lock")
  private final IdentityHashMap<StateChangeCallback, Executor> stateCallbackMap =
      new IdentityHashMap<>();

  // TODO: Consider using a PriorityQueue here instead, so that queued downloads can
  // retain the order in the queue as they get added/removed due to waiting on connectivity.
  @GuardedBy("lock")
  private final Queue<QueuedDownload> queuedDownloads = new ArrayDeque<>();

  @GuardedBy("lock")
  private final List<FluentFuture<DownloadResult>> unresolvedFutures = new ArrayList<>();

  private final Object lock = new Object();

  @GuardedBy("lock")
  private int numDownloadsInFlight = 0;

  @GuardedBy("lock")
  private int numDownloadsPendingConnectivity = 0;

  private Downloader(Builder builder) {
    ImmutableMap<String, UrlEngine> urlEngineMap = ImmutableMap.copyOf(builder.urlEngineMap);
    checkArgument(!urlEngineMap.isEmpty(), "Must have at least one UrlEngine");
    checkArgument(builder.ioExecutor != null, "Must set a callback executor");
    checkArgument(builder.logger != null, "Must set a logger");
    checkArgument(builder.connectivityHandler != null, "Must set a connectivity handler");

    this.urlEngineMap = urlEngineMap;
    this.ioExecutor = builder.ioExecutor;
    this.logger = builder.logger;
    this.connectivityHandler = builder.connectivityHandler;
    this.maxConcurrentDownloads = builder.maxConcurrentDownloads;
  }

  /**
   * Creates a new {@link DownloadRequest.Builder} for the given {@link URI} and {@link
   * DownloadDestination}. The builder may be used to further customize the request. To execute the
   * request, pass the built request to {@link #execute}.
   */
  @CheckReturnValue
  public DownloadRequest.Builder newRequestBuilder(URI uri, DownloadDestination destination) {
    return DownloadRequest.newBuilder()
        .setDestination(destination)
        .setUri(uri)
        .setDownloadConstraints(DownloadConstraints.NETWORK_CONNECTED);
  }

  /**
   * Executes the provided request. The request will be handled by the underlying {@link UrlEngine}
   * and the result is streamed to the {@link DownloadDestination} created for the request. The
   * download result is provided asynchronously as a {@link FluentFuture} that resolves when the
   * download is complete.
   *
   * <p>The download can be cancelled by calling {@link FluentFuture#cancel} on the future instance
   * returned by this method. Cancellation is best-effort and does not guarantee that the download
   * will stop immediately, as it is impossible to stop a thread that is in the middle of reading
   * bytes off the network.
   *
   * <p>Note that this method is not idempotent! The downloader does not attempt to merge/de-dupe
   * incoming requests, even if the same exact request is executed twice. The calling code needs to
   * be careful to manage its downloads in such a way that duplicated downloads don't occur.
   *
   * <p>TODO: Document what types of exceptions can be set on the returned future, and how clients
   * are expected to handle them.
   */
  @CheckReturnValue
  public FluentFuture<DownloadResult> execute(DownloadRequest request) {
    FluentFuture<DownloadResult> resultFuture;

    synchronized (lock) {
      ClosingFuture<DownloadResult> closingFuture = enqueueRequest(request);
      closingFuture.statusFuture().addListener(() -> onDownloadComplete(request), ioExecutor);

      resultFuture = closingFuture.finishToFuture();
      unresolvedFutures.add(resultFuture);
      resultFuture.addListener(
          () -> {
            synchronized (lock) {
              unresolvedFutures.remove(resultFuture);
            }
          },
          MoreExecutors.directExecutor());
    }

    logger.logFine("New request enqueued, running queue: %s", request.uri());
    maybeRunQueuedDownloads();
    return resultFuture;
  }

  /**
   * Cancels <strong>all</strong> ongoing downloads. This will result in all unresolved {@link
   * FluentFuture} instances returned by {@link #execute} to be cancelled immediately and have their
   * error callbacks invoked.
   *
   * <p>Because implementations of {@link FluentFuture} allow callbacks to be garbage collected
   * after the future is resolved, calling {@code cancelAll} is an effective way to avoid having the
   * Downloader leak memory after it is logically no longer needed, e.g. if it is only used from an
   * Android {@link android.app.Activity}, and that Activity is destroyed.
   *
   * <p>However, in general the Downloader should be run from a process-level context, e.g. in an
   * Android {@link android.app.Service}, so that the downloader doesn't implicitly hold on to
   * UI-scoped objects.
   */
  public void cancelAll() {
    List<FluentFuture<DownloadResult>> unresolvedFuturesCopy;

    synchronized (lock) {
      // Copy the set of unresolved futures to a local variable to avoid hitting
      // ConcurrentModificationExceptions, which could happen since canceling the future may
      // trigger the future callback in execute() that removes the future from the set of
      // unresolved futures.
      unresolvedFuturesCopy = new ArrayList<>(unresolvedFutures);
      unresolvedFutures.clear();
    }

    for (FluentFuture<DownloadResult> unresolvedFuture : unresolvedFuturesCopy) {
      unresolvedFuture.cancel(true);
    }
  }

  /**
   * Registers an {@link StateChangeCallback} with this downloader instance, to be run when various
   * state changes occur. The callback will be executed on the provided {@link Executor}. Use {@link
   * #unregisterStateChangeCallback} to unregister a previously registered callback. The callback is
   * invoked when the following state transitions occur:
   *
   * <ul>
   *   <li>A download was requested via a call to {@link #execute}
   *   <li>A download completed
   *   <li>A download started waiting for its connectivity constraints to be satisfied
   *   <li>A download stopped waiting for its connectivity constraints to be satisfied
   * </ul>
   *
   * <p>A newly registered callback instance will only called for state changes that are triggered
   * after the registration; there is no mechanism to replay previous state changes on the callback.
   *
   * <p>Registering the same callback multiple times has no effect, except that it will overwrite
   * the {@link Executor} that was used in a previous registration call.
   *
   * <p>Invocation of the callbacks will be internally serialized to avoid concurrent invocations of
   * the callback with possibly conflicting state. A new invocation of the callback will not start
   * running until the previous one has completed. If multiple callbacks are being registered that
   * must be synchronized with each other, then the caller must take care to coordinate this
   * externally, such as locking in the callbacks or using an executor that guarantees
   * serialization, such as {@link MoreExecutors#newSequentialExecutor}.
   *
   * <p>Warning: Do not use {@link MoreExecutors#directExecutor} or similar executor mechanisms, as
   * doing so can easily lead to a deadlock, since the internal downloader lock is held while
   * scheduling the callbacks on the provided executor.
   */
  public void registerStateChangeCallback(StateChangeCallback callback, Executor executor) {
    synchronized (lock) {
      stateCallbackMap.put(callback, MoreExecutors.newSequentialExecutor(executor));
    }
  }

  /**
   * Unregisters a previously registered {@link StateChangeCallback} with this downloader instance.
   *
   * <p>Attempting to unregister a callback that was never registered is a no-op.
   */
  public void unregisterStateChangeCallback(StateChangeCallback callback) {
    synchronized (lock) {
      stateCallbackMap.remove(callback);
    }
  }

  private void onDownloadComplete(DownloadRequest request) {
    synchronized (lock) {
      // The number of downloads should be well-balanced and this case should never
      // trigger, so this is just a defensive check.
      checkState(
          numDownloadsInFlight >= 0, "Encountered < 0 downloads in flight, this shouldn't happen");
    }

    logger.logFine("Download complete, running queued downloads: %s", request.uri());
    maybeRunQueuedDownloads();
  }

  private void maybeRunQueuedDownloads() {
    // Loop until we run out of download slots or out of queued downloads. It should be impossible
    // for this to loop forever. Also, note that the synchronized block is inside the loop, since
    // the outer loop conditions don't touch shared mutable state.
    while (true) {
      synchronized (lock) {
        if (numDownloadsInFlight >= maxConcurrentDownloads) {
          logger.logInfo("Exceeded max concurrent downloads, not running another queued request");
          return;
        }

        QueuedDownload queuedDownload = queuedDownloads.poll();
        if (queuedDownload == null) {
          return;
        }

        DownloadRequest request = queuedDownload.request();
        ListenableFuture<Void> connectivityFuture = checkConnectivity(request);
        if (connectivityFuture.isDone()) {
          logger.logFine("Connectivity satisfied; running request. uri=%s", request.uri());
          numDownloadsInFlight++;
          ListenableFuture<?> statusFuture = queuedDownload.resultFuture().statusFuture();
          statusFuture.addListener(
              () -> {
                synchronized (lock) {
                  numDownloadsInFlight--;

                  // One less download in flight, let the state change listeners know.
                  runStateCallbacks();
                }

                // A download slot was just freed up, run queued downloads again.
                logger.logFine(
                    "Queued download completed, running queued downloads: %s", request.uri());
                maybeRunQueuedDownloads();
              },
              ioExecutor);

          // A new download is about to be in flight, let state callbacks know about the
          // state change.
          runStateCallbacks();
          queuedDownload.task().run();
        } else {
          logger.logInfo("Waiting on connectivity for request: uri=%s", request.uri());
          handlePendingConnectivity(connectivityFuture, queuedDownload);
        }
      }
    }
  }

  @GuardedBy("lock")
  private void handlePendingConnectivity(
      ListenableFuture<Void> connectivityFuture, QueuedDownload queuedDownload) {
    // Keep track of the number of requests waiting.
    numDownloadsPendingConnectivity++;
    connectivityFuture.addListener(
        () -> {
          synchronized (lock) {
            numDownloadsPendingConnectivity--;

            // Let the listeners know we are no longer waiting.
            runStateCallbacks();
          }
        },
        MoreExecutors.directExecutor());

    // Let the listeners know we're waiting.
    runStateCallbacks();

    Futures.addCallback(
        connectivityFuture,
        new FutureCallback<Void>() {
          @Override
          public void onSuccess(Void result1) {
            logger.logInfo("Connectivity changed, running queued requests");
            requeue(queuedDownload);
          }

          @Override
          public void onFailure(Throwable t) {
            if (t instanceof TimeoutException) {
              logger.logInfo("Timed out waiting for connectivity change");
              requeue(queuedDownload);
            } else if (t instanceof CancellationException) {
              logger.logFine("Connectivity future cancelled, running queued downloads");
              maybeRunQueuedDownloads();
            } else {
              logger.logError(t, "Error observing connectivity changes");
              cancelAll();
            }
          }
        },
        ioExecutor);

    // Run state callbacks and cancel the connectivity future when the result task resolves.
    // It doesn't matter if it succeeded or failed, either way it means we no longer need to wait
    // for connectivity.
    queuedDownload
        .task()
        .addListener(
            () -> {
              synchronized (lock) {
                logger.logInfo("Queued task completed, cancelling connectivity check");
                runStateCallbacks();
                connectivityFuture.cancel(false);
              }
            },
            MoreExecutors.directExecutor());
  }

  private void requeue(QueuedDownload queuedDownload) {
    synchronized (lock) {
      addToQueue(queuedDownload);
    }

    logger.logInfo(
        "Requeuing download after connectivity change: %s", queuedDownload.request().uri());
    maybeRunQueuedDownloads();
  }

  private ClosingFuture<DownloadResult> enqueueRequest(DownloadRequest request) {
    synchronized (lock) {
      ListenableFutureTask<Void> task = ListenableFutureTask.create(() -> null);
      // When the task runs (i.e. is taken off the queue and is explicitly run), all pre-flight
      // checks should have been made, so at that point the request is send to the underlying
      // network stack for execution.
      ClosingFuture<DownloadResult> resultFuture =
          ClosingFuture.from(task)
              .transformAsync((closer, result) -> runRequest(request), ioExecutor);
      addToQueue(QueuedDownload.create(request, task, resultFuture));
      return resultFuture;
    }
  }

  @GuardedBy("lock")
  private void addToQueue(QueuedDownload queuedDownload) {
    queuedDownloads.add(queuedDownload);

    // Make sure that when the task completes, the queued download is definitely removed
    // from the queue. This is necessary to be robust in the face of cancellation, as
    // canceled tasks may not get removed from the queue otherwise.
    queuedDownload
        .task()
        .addListener(
            () -> {
              synchronized (lock) {
                if (queuedDownloads.remove(queuedDownload)) {
                  // If the queued download was actually removed, update the state callbacks
                  // to reflect the state change.
                  runStateCallbacks();
                }
              }
            },
            MoreExecutors.directExecutor());

    // Now that a new request is on the queue, run the state callbacks.
    runStateCallbacks();
  }

  private ClosingFuture<DownloadResult> runRequest(DownloadRequest request) throws IOException {
    URI uri = request.uri();
    UrlEngine urlEngine = checkNotNull(urlEngineMap.get(uri.getScheme()));
    UrlRequest.Builder urlRequestBuilder = urlEngine.createRequest(uri.toString());
    for (Map.Entry<String, String> entry : request.headers().entries()) {
      urlRequestBuilder.addHeader(entry.getKey(), entry.getValue());
    }

    long numExistingBytes = request.destination().numExistingBytes();
    if (numExistingBytes > 0) {
      logger.logInfo(
          "Existing bytes found. numExistingBytes=%d, uri=%s", numExistingBytes, request.uri());
      urlRequestBuilder.addHeader(HttpHeaders.RANGE, "bytes=" + numExistingBytes + "-");

      DownloadMetadata destinationMetadata = request.destination().readMetadata();
      String contentTag = destinationMetadata.getContentTag();
      long lastModifiedTimeSeconds = destinationMetadata.getLastModifiedTimeSeconds();
      if (!contentTag.isEmpty()) {
        urlRequestBuilder.addHeader(HttpHeaders.IF_RANGE, contentTag);
      } else if (lastModifiedTimeSeconds > 0) {
        urlRequestBuilder.addHeader(
            HttpHeaders.IF_RANGE, formatRfc1123Date(lastModifiedTimeSeconds));
      } else {
        // TODO: This should probably just clear the destination and remove the Range
        // header so there's no chance of data corruption. Leaving this as-is for now to
        // keep supporting range requests for offline maps.
        logger.logWarning(
            "Sending range request without If-Range header, due to missing destination "
                + "metadata. Data corruption is possible.");
      }
    }

    ListenableFuture<UrlRequest> urlRequestFuture;
    OAuthTokenProvider oAuthTokenProvider = request.oAuthTokenProvider();
    if (oAuthTokenProvider != null) {
      urlRequestFuture =
          Futures.transform(
              oAuthTokenProvider.provideOAuthToken(uri),
              authToken -> {
                if (authToken != null) {
                  urlRequestBuilder.addHeader(HttpHeaders.AUTHORIZATION, "Bearer " + authToken);
                }
                return urlRequestBuilder.build();
              },
              ioExecutor);
    } else {
      urlRequestFuture = Futures.immediateFuture(urlRequestBuilder.build());
    }

    return ClosingFuture.from(urlRequestFuture)
        .transform((closer, urlRequest) -> checkNotNull(urlRequest).send(), ioExecutor)
        .transformAsync(
            (closer, responseFuture) -> completeRequest(request, checkNotNull(responseFuture)),
            ioExecutor);
  }

  /**
   * @param request the original request that triggered this call
   * @param responseFuture the {@link UrlResponse}, provided asynchronously via a {@link
   *     ListenableFuture}
   */
  private ClosingFuture<DownloadResult> completeRequest(
      DownloadRequest request, ListenableFuture<UrlResponse> responseFuture) {
    return ClosingFuture.from(responseFuture)
        .transformAsync(
            (closer, urlResponse) -> {
              checkNotNull(urlResponse);
              // We want to close the response regardless of whether we succeed or fail.
              closer.eventuallyClose(urlResponse, ioExecutor);
              logger.logFine(
                  "Got URL response, starting to read response body. uri=%s", request.uri());
              DownloadDestination destination = request.destination();
              if (request.headers().containsKey(HttpHeaders.RANGE)
                  && checkNotNull(urlResponse).getResponseCode() != HTTP_PARTIAL_CONTENT) {
                logger.logFine("Clearing %s as our range request wasn't honored", destination);
                destination.clear();
              }
              WritableByteChannel byteChannel =
                  destination.openByteChannel(
                      parseResponseStartOffset(urlResponse), parseResponseMetadata(urlResponse));
              closer.eventuallyClose(byteChannel, ioExecutor);
              return ClosingFuture.from(checkNotNull(urlResponse).readResponseBody(byteChannel));
            },
            ioExecutor)
        .catching(
            RequestException.class,
            (closer, requestException) -> {
              if (checkNotNull(requestException).getErrorDetails().getHttpStatusCode()
                  == HTTP_RANGE_NOT_SATISFIABLE) {
                // This is a bit of a special edge case. Encountering this error means the server
                // rejected our HTTP range request because it was outside the range of available
                // bytes. This may well mean that the request was malformed (e.g. data on disk was
                // corrupted and the existing file size ended up larger than what exists on the
                // server). But the more common cause for this error is that the entire file was
                // in fact already downloaded, so the requested range would cover 0 bytes, which
                // the server interprets as not satisfiable. To mitigate that case we simply return
                // a success state here by indicating 0 bytes were downloaded.

                // This isn't exactly ideal, as it means that a potential class of errors will
                // go unnoticed. A better solution might find a way to distinguish between the
                // common, file-already-downloaded case and the file corrupted case. This solution
                // is put in place mainly to retain parity with the older downloader implementation.
                return 0L;
              } else {
                throw new DownloadException(requestException);
              }
            },
            ioExecutor)
        .transform(
            (closer, bytesWritten) -> {
              logger.logFine(
                  "Response body written. bytesWritten=%d, uri=%s",
                  checkNotNull(bytesWritten), request.uri());
              return DownloadResult.create(checkNotNull(bytesWritten));
            },
            ioExecutor)
        .catchingAsync(
            Exception.class,
            (closer, exception) -> {
              ClosingFuture<DownloadResult> result;
              synchronized (lock) {
                logger.logWarning(
                    exception, "Error reading download result. uri=%s", request.uri());
                RequestException requestException = getRequestException(exception);
                if (requestException != null
                    && requestException.getErrorDetails().isRetryableAsIs()) {
                  // Retry the request by just re-enqueueing it. Note that we also need to
                  // call maybeRunQueuedRequest to keep the queue moving. Also, in this particular
                  // case we need to decrement the in-flight downloads count, as we are taking a
                  // previously in-flight download and putting it back in the queue without
                  // resolving the underlying result future.
                  numDownloadsInFlight--;
                  result = enqueueRequest(request);
                } else {
                  throw new DownloadException(exception);
                }
              }

              logger.logInfo("Running queued downloads after handling request exception");
              maybeRunQueuedDownloads();
              return result;
            },
            ioExecutor);
  }

  @GuardedBy("lock")
  private ListenableFuture<Void> checkConnectivity(DownloadRequest request) {
    if (!SCHEMES_REQUIRING_CONNECTIVITY.contains(request.uri().getScheme())) {
      return Futures.immediateVoidFuture();
    }

    return connectivityHandler.checkConnectivity(request.downloadConstraints());
  }

  @GuardedBy("lock")
  private void runStateCallbacks() {
    State state =
        State.create(numDownloadsInFlight, queuedDownloads.size(), numDownloadsPendingConnectivity);
    for (Map.Entry<StateChangeCallback, Executor> callbackEntry : stateCallbackMap.entrySet()) {
      callbackEntry.getValue().execute(() -> callbackEntry.getKey().onStateChange(state));
    }
  }

  private static String formatRfc1123Date(long unixTimeSeconds) {
    synchronized (SIMPLE_DATE_FORMAT_LOCK) {
      return RFC_1123_FORMATTER.format(new Date(SECONDS.toMillis(unixTimeSeconds)));
    }
  }

  private static long parseResponseStartOffset(UrlResponse response) throws DownloadException {
    if (response.getResponseCode() != HttpURLConnection.HTTP_PARTIAL) {
      return 0;
    }

    List<String> contentRangeHeaders = response.getResponseHeaders().get(HttpHeaders.CONTENT_RANGE);

    checkDownloadState(
        contentRangeHeaders != null && !contentRangeHeaders.isEmpty(),
        "Host returned 206/PARTIAL response code but didn't provide a "
            + "'Content-Range' response header");
    String contentRangeHeader = checkNotNull(contentRangeHeaders).get(0);
    Matcher matcher = CONTENT_RANGE_HEADER_PATTERN.matcher(contentRangeHeader);
    checkDownloadState(
        matcher.matches() && matcher.groupCount() > 0,
        "Content-Range response header didn't match expected pattern. " + "Was '%s', expected '%s'",
        contentRangeHeader,
        CONTENT_RANGE_HEADER_PATTERN.pattern());
    return Long.parseLong(checkNotNull(matcher.group(1)));
  }

  private static DownloadMetadata parseResponseMetadata(UrlResponse response)
      throws DownloadException {
    String contentTag = parseResponseContentTag(response);
    long lastModifiedTimeSeconds = parseResponseModifiedTime(response);
    return DownloadMetadata.create(contentTag, lastModifiedTimeSeconds);
  }

  private static long parseResponseModifiedTime(UrlResponse response) throws DownloadException {
    List<String> lastModifiedHeaders = response.getResponseHeaders().get(HttpHeaders.LAST_MODIFIED);
    if (lastModifiedHeaders == null || lastModifiedHeaders.isEmpty()) {
      return 0L;
    }

    String lastModifiedHeader = lastModifiedHeaders.get(0);
    Date date;

    try {
      synchronized (SIMPLE_DATE_FORMAT_LOCK) {
        date = RFC_1123_FORMATTER.parse(lastModifiedHeader);
      }
      if (date == null) {
        throw new DownloadException("Invalid Last-Modified header: " + lastModifiedHeader);
      }
    } catch (ParseException e) {
      throw new DownloadException("Invalid Last-Modified header: " + lastModifiedHeader, e);
    }

    return MILLISECONDS.toSeconds(date.getTime());
  }

  private static String parseResponseContentTag(UrlResponse response) {
    List<String> contentTagHeaders = response.getResponseHeaders().get(HttpHeaders.ETAG);
    if (contentTagHeaders == null || contentTagHeaders.isEmpty()) {
      return "";
    }

    return contentTagHeaders.get(0);
  }

  @AutoValue
  abstract static class QueuedDownload {
    abstract DownloadRequest request();

    abstract ListenableFutureTask<?> task();

    abstract ClosingFuture<DownloadResult> resultFuture();

    static QueuedDownload create(
        DownloadRequest request,
        ListenableFutureTask<?> task,
        ClosingFuture<DownloadResult> resultFuture) {
      return new AutoValue_Downloader_QueuedDownload(request, task, resultFuture);
    }
  }

  @VisibleForTesting
  @Nullable
  static RequestException getRequestException(@Nullable Throwable throwable) {
    if (throwable == null) {
      return null;
    } else {
      return (RequestException)
          Iterables.find(
              getCausalChain(throwable),
              instanceOf(RequestException.class),
              /* defaultValue= */ null);
    }
  }

  @FormatMethod
  private static void checkDownloadState(
      boolean state, @FormatString String message, Object... args) throws DownloadException {
    if (!state) {
      throw new DownloadException(String.format(message, args));
    }
  }
}