Support multiple mirror URLs for external repos

This change improves upon 4c67807964e37cfd55bbcda4c6374fcc480bcecc.

- A urls attribute has been added to the native workspace rules, with
  the exception of maven_jar and git_repository. The Skylark repository
  API also supports multiple URLs now.

- The earlier mirrors in the list are preferred. Failover will happen
  automatically in parallel.

- The first 32kB of data is checked before choosing a mirror in order
  to evade captive portals.

- If one's Internet goes down or a download times out, then the
  download will resume automatically where it left off, provided the
  server supports RFC7233 for that particular file. Please note that
  GitHub does not support this for archive snapshots. Files should
  always be mirrored to a CDN, e.g. GCS, because they support this.

- A semaphore is now used on downloads so only 8 can happen at once.

Fixes #1814
Fixes #2131
Fixes #2008
Fixes #1968
Fixes #1717
Fixes #943
Wont fix #1194
Fixes tensorflow/tensorflow#5933
Fixes tensorflow/tensorflow#5924
Fixes tensorflow/tensorflow#5924
Fixes tensorflow/tensorflow#5432
See #1607
See #821
See tensorflow/tensorflow#5080
See tensorflow/tensorflow#5029
See tensorflow/tensorflow#4583
See tensorflow/tensorflow#4058

RELNOTES: A urls attribute has been added to repository rules to support multiple mirror URLs for reliably downloading files.

--
MOS_MIGRATED_REVID=140495736

15 files changed, 1807 insertions, 167 deletions

diff --git a/src/test/java/com/google/devtools/build/lib/bazel/repository/downloader/BUILD b/src/test/java/com/google/devtools/build/lib/bazel/repository/downloader/BUILD
index 6d9253a262..ceaaf305bf 100644
--- a/src/test/java/com/google/devtools/build/lib/bazel/repository/downloader/BUILD
+++ b/src/test/java/com/google/devtools/build/lib/bazel/repository/downloader/BUILD
@@ -13,9 +13,11 @@ java_test(
     ],
     deps = [
         "//src/main/java/com/google/devtools/build/lib:events",
+        "//src/main/java/com/google/devtools/build/lib:util",
         "//src/main/java/com/google/devtools/build/lib/bazel/repository/downloader",
         "//src/test/java/com/google/devtools/build/lib:foundations_testutil",
         "//src/test/java/com/google/devtools/build/lib:test_runner",
+        "//src/test/java/com/google/devtools/build/lib:testutil",
         "//third_party:guava",
         "//third_party:junit4",
         "//third_party:mockito",
diff --git a/src/test/java/com/google/devtools/build/lib/bazel/repository/downloader/DownloaderTestSuite.java b/src/test/java/com/google/devtools/build/lib/bazel/repository/downloader/DownloaderTestSuite.java
index 1a48a1ceb5..1c2477e64b 100644
--- a/src/test/java/com/google/devtools/build/lib/bazel/repository/downloader/DownloaderTestSuite.java
+++ b/src/test/java/com/google/devtools/build/lib/bazel/repository/downloader/DownloaderTestSuite.java
@@ -21,7 +21,14 @@ import org.junit.runners.Suite.SuiteClasses;
 /** Test suite for downloader package. */
 @RunWith(Suite.class)
 @SuiteClasses({
+  HashInputStreamTest.class,
+  HttpConnectorMultiplexerIntegrationTest.class,
+  HttpConnectorMultiplexerTest.class,
   HttpConnectorTest.class,
+  HttpStreamTest.class,
+  HttpUtilsTest.class,
+  ProgressInputStreamTest.class,
   ProxyHelperTest.class,
+  RetryingInputStreamTest.class,
 })
 class DownloaderTestSuite {}
diff --git a/src/test/java/com/google/devtools/build/lib/bazel/repository/downloader/DownloaderTestUtils.java b/src/test/java/com/google/devtools/build/lib/bazel/repository/downloader/DownloaderTestUtils.java
new file mode 100644
index 0000000000..1f55667efa
--- /dev/null
+++ b/src/test/java/com/google/devtools/build/lib/bazel/repository/downloader/DownloaderTestUtils.java
@@ -0,0 +1,45 @@
+// Copyright 2016 The Bazel Authors. All rights reserved.
+//
+// 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.devtools.build.lib.bazel.repository.downloader;
+
+import static java.nio.charset.StandardCharsets.ISO_8859_1;
+
+import com.google.common.base.Joiner;
+import com.google.common.io.ByteStreams;
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.net.MalformedURLException;
+import java.net.Socket;
+import java.net.URL;
+import javax.annotation.WillNotClose;
+
+final class DownloaderTestUtils {
+
+  static URL makeUrl(String url) {
+    try {
+      return new URL(url);
+    } catch (MalformedURLException e) {
+      throw new RuntimeException(e);
+    }
+  }
+
+  static void sendLines(@WillNotClose Socket socket, String... data) throws IOException {
+    ByteStreams.copy(
+        new ByteArrayInputStream(Joiner.on("\r\n").join(data).getBytes(ISO_8859_1)),
+        socket.getOutputStream());
+  }
+
+  private DownloaderTestUtils() {}
+}
diff --git a/src/test/java/com/google/devtools/build/lib/bazel/repository/downloader/HashInputStreamTest.java b/src/test/java/com/google/devtools/build/lib/bazel/repository/downloader/HashInputStreamTest.java
new file mode 100644
index 0000000000..c1e14df907
--- /dev/null
+++ b/src/test/java/com/google/devtools/build/lib/bazel/repository/downloader/HashInputStreamTest.java
@@ -0,0 +1,67 @@
+// Copyright 2016 The Bazel Authors. All rights reserved.
+//
+// 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.devtools.build.lib.bazel.repository.downloader;
+
+import static com.google.common.truth.Truth.assertThat;
+import static java.nio.charset.StandardCharsets.UTF_8;
+
+import com.google.common.hash.HashCode;
+import com.google.common.hash.Hashing;
+import com.google.common.io.CharStreams;
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.ExpectedException;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+/** Unit tests for {@link HashInputStream}. */
+@RunWith(JUnit4.class)
+@SuppressWarnings("resource")
+public class HashInputStreamTest {
+
+  @Rule
+  public final ExpectedException thrown = ExpectedException.none();
+
+  @Test
+  public void validChecksum_readsOk() throws Exception {
+    assertThat(
+            CharStreams.toString(
+                new InputStreamReader(
+                    new HashInputStream(
+                        new ByteArrayInputStream("hello".getBytes(UTF_8)),
+                        Hashing.sha1(),
+                        HashCode.fromString("aaf4c61ddcc5e8a2dabede0f3b482cd9aea9434d")),
+                    UTF_8)))
+        .isEqualTo("hello");
+  }
+
+  @Test
+  public void badChecksum_throwsIOException() throws Exception {
+    thrown.expect(IOException.class);
+    thrown.expectMessage("Checksum");
+    assertThat(
+            CharStreams.toString(
+                new InputStreamReader(
+                    new HashInputStream(
+                        new ByteArrayInputStream("hello".getBytes(UTF_8)),
+                        Hashing.sha1(),
+                        HashCode.fromString("0000000000000000000000000000000000000000")),
+                    UTF_8)))
+        .isNull();  // Only here to make @CheckReturnValue happy.
+  }
+}
diff --git a/src/test/java/com/google/devtools/build/lib/bazel/repository/downloader/HttpConnectorMultiplexerIntegrationTest.java b/src/test/java/com/google/devtools/build/lib/bazel/repository/downloader/HttpConnectorMultiplexerIntegrationTest.java
new file mode 100644
index 0000000000..b5b77a28cf
--- /dev/null
+++ b/src/test/java/com/google/devtools/build/lib/bazel/repository/downloader/HttpConnectorMultiplexerIntegrationTest.java
@@ -0,0 +1,227 @@
+// Copyright 2016 The Bazel Authors. All rights reserved.
+//
+// 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.devtools.build.lib.bazel.repository.downloader;
+
+import static com.google.common.io.ByteStreams.toByteArray;
+import static com.google.common.truth.Truth.assertThat;
+import static com.google.devtools.build.lib.bazel.repository.downloader.DownloaderTestUtils.sendLines;
+import static com.google.devtools.build.lib.bazel.repository.downloader.HttpParser.readHttpRequest;
+import static java.nio.charset.StandardCharsets.US_ASCII;
+import static java.util.Arrays.asList;
+import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.anyLong;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import com.google.common.collect.ImmutableList;
+import com.google.devtools.build.lib.events.EventHandler;
+import com.google.devtools.build.lib.testutil.ManualClock;
+import com.google.devtools.build.lib.util.Sleeper;
+import java.io.IOException;
+import java.net.InetAddress;
+import java.net.Proxy;
+import java.net.ServerSocket;
+import java.net.Socket;
+import java.net.URL;
+import java.util.Locale;
+import java.util.concurrent.Callable;
+import java.util.concurrent.CyclicBarrier;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Phaser;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.ExpectedException;
+import org.junit.rules.Timeout;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.stubbing.Answer;
+
+/** Black box integration tests for {@link HttpConnectorMultiplexer}. */
+@RunWith(JUnit4.class)
+public class HttpConnectorMultiplexerIntegrationTest {
+
+  @Rule
+  public final ExpectedException thrown = ExpectedException.none();
+
+  @Rule
+  public final Timeout globalTimeout = new Timeout(10000);
+
+  private final ExecutorService executor = Executors.newFixedThreadPool(3);
+  private final ProxyHelper proxyHelper = mock(ProxyHelper.class);
+  private final EventHandler eventHandler = mock(EventHandler.class);
+  private final ManualClock clock = new ManualClock();
+  private final Sleeper sleeper = mock(Sleeper.class);
+  private final Locale locale = Locale.US;
+  private final HttpConnector connector =
+      new HttpConnector(locale, eventHandler, proxyHelper, sleeper);
+  private final ProgressInputStream.Factory progressInputStreamFactory =
+      new ProgressInputStream.Factory(locale, clock, eventHandler);
+  private final HttpStream.Factory httpStreamFactory =
+      new HttpStream.Factory(progressInputStreamFactory);
+  private final HttpConnectorMultiplexer multiplexer =
+      new HttpConnectorMultiplexer(eventHandler, connector, httpStreamFactory, clock, sleeper);
+
+  @Before
+  public void before() throws Exception {
+    when(proxyHelper.createProxyIfNeeded(any(URL.class))).thenReturn(Proxy.NO_PROXY);
+  }
+
+  @After
+  public void after() throws Exception {
+    executor.shutdown();
+  }
+
+  @Test
+  public void normalRequest() throws Exception {
+    final Phaser phaser = new Phaser(3);
+    try (ServerSocket server1 = new ServerSocket(0, 1, InetAddress.getByName("127.0.0.1"));
+        ServerSocket server2 = new ServerSocket(0, 1, InetAddress.getByName("127.0.0.1"))) {
+      for (final ServerSocket server : asList(server1, server2)) {
+        executor.submit(
+            new Callable<Object>() {
+              @Override
+              public Object call() throws Exception {
+                for (String status : asList("503 MELTDOWN", "500 ERROR", "200 OK")) {
+                  phaser.arriveAndAwaitAdvance();
+                  try (Socket socket = server.accept()) {
+                    readHttpRequest(socket.getInputStream());
+                    sendLines(socket,
+                        "HTTP/1.1 " + status,
+                        "Date: Fri, 31 Dec 1999 23:59:59 GMT",
+                        "Connection: close",
+                        "",
+                        "hello");
+                  }
+                }
+                return null;
+              }
+            });
+      }
+      phaser.arriveAndAwaitAdvance();
+      phaser.arriveAndDeregister();
+      try (HttpStream stream =
+              multiplexer.connect(
+                  ImmutableList.of(
+                      new URL(String.format("http://127.0.0.1:%d", server1.getLocalPort())),
+                      new URL(String.format("http://127.0.0.1:%d", server2.getLocalPort()))),
+                  "2cf24dba5fb0a30e26e83b2ac5b9e29e1b161e5c1fa7425e73043362938b9824")) {
+        assertThat(toByteArray(stream)).isEqualTo("hello".getBytes(US_ASCII));
+      }
+    }
+  }
+
+  @Test
+  public void captivePortal_isAvoided() throws Exception {
+    final CyclicBarrier barrier = new CyclicBarrier(2);
+    doAnswer(
+        new Answer<Void>() {
+          @Override
+          public Void answer(InvocationOnMock invocation) throws Throwable {
+            barrier.await();
+            return null;
+          }
+        }).when(sleeper).sleepMillis(anyLong());
+    try (final ServerSocket server1 = new ServerSocket(0, 1, InetAddress.getByName("127.0.0.1"));
+        final ServerSocket server2 = new ServerSocket(0, 1, InetAddress.getByName("127.0.0.1"))) {
+      executor.submit(
+          new Callable<Object>() {
+            @Override
+            public Object call() throws Exception {
+              try (Socket socket = server1.accept()) {
+                readHttpRequest(socket.getInputStream());
+                sendLines(socket,
+                    "HTTP/1.1 200 OK",
+                    "Date: Fri, 31 Dec 1999 23:59:59 GMT",
+                    "Warning: https://youtu.be/rJ6O5sTPn1k",
+                    "Connection: close",
+                    "",
+                    "Und wird die Welt auch in Flammen stehen",
+                    "Wir werden wieder auferstehen");
+              }
+              barrier.await();
+              return null;
+            }
+          });
+      executor.submit(
+          new Callable<Object>() {
+            @Override
+            public Object call() throws Exception {
+              try (Socket socket = server2.accept()) {
+                readHttpRequest(socket.getInputStream());
+                sendLines(socket,
+                    "HTTP/1.1 200 OK",
+                    "Date: Fri, 31 Dec 1999 23:59:59 GMT",
+                    "Connection: close",
+                    "",
+                    "hello");
+              }
+              return null;
+            }
+          });
+      try (HttpStream stream =
+              multiplexer.connect(
+                  ImmutableList.of(
+                      new URL(String.format("http://127.0.0.1:%d", server1.getLocalPort())),
+                      new URL(String.format("http://127.0.0.1:%d", server2.getLocalPort()))),
+                  "2cf24dba5fb0a30e26e83b2ac5b9e29e1b161e5c1fa7425e73043362938b9824")) {
+        assertThat(toByteArray(stream)).isEqualTo("hello".getBytes(US_ASCII));
+      }
+    }
+  }
+
+  @Test
+  public void allMirrorsDown_throwsIOException() throws Exception {
+    final CyclicBarrier barrier = new CyclicBarrier(4);
+    try (ServerSocket server1 = new ServerSocket(0, 1, InetAddress.getByName("127.0.0.1"));
+        ServerSocket server2 = new ServerSocket(0, 1, InetAddress.getByName("127.0.0.1"));
+        ServerSocket server3 = new ServerSocket(0, 1, InetAddress.getByName("127.0.0.1"))) {
+      for (final ServerSocket server : asList(server1, server2, server3)) {
+        executor.submit(
+            new Callable<Object>() {
+              @Override
+              public Object call() throws Exception {
+                barrier.await();
+                while (true) {
+                  try (Socket socket = server.accept()) {
+                    readHttpRequest(socket.getInputStream());
+                    sendLines(socket,
+                        "HTTP/1.1 503 MELTDOWN",
+                        "Date: Fri, 31 Dec 1999 23:59:59 GMT",
+                        "Warning: https://youtu.be/6M6samPEMpM",
+                        "Connection: close",
+                        "",
+                        "");
+                  }
+                }
+              }
+            });
+      }
+      barrier.await();
+      thrown.expect(IOException.class);
+      thrown.expectMessage("All mirrors are down: [GET returned 503 MELTDOWN]");
+      multiplexer.connect(
+          ImmutableList.of(
+              new URL(String.format("http://127.0.0.1:%d", server1.getLocalPort())),
+              new URL(String.format("http://127.0.0.1:%d", server2.getLocalPort())),
+              new URL(String.format("http://127.0.0.1:%d", server3.getLocalPort()))),
+          "2cf24dba5fb0a30e26e83b2ac5b9e29e1b161e5c1fa7425e73043362938b9825");
+    }
+  }
+}
diff --git a/src/test/java/com/google/devtools/build/lib/bazel/repository/downloader/HttpConnectorMultiplexerTest.java b/src/test/java/com/google/devtools/build/lib/bazel/repository/downloader/HttpConnectorMultiplexerTest.java
new file mode 100644
index 0000000000..dedf316868
--- /dev/null
+++ b/src/test/java/com/google/devtools/build/lib/bazel/repository/downloader/HttpConnectorMultiplexerTest.java
@@ -0,0 +1,266 @@
+// Copyright 2016 The Bazel Authors. All rights reserved.
+//
+// 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.devtools.build.lib.bazel.repository.downloader;
+
+import static com.google.common.io.ByteStreams.toByteArray;
+import static com.google.common.truth.Truth.assertThat;
+import static com.google.devtools.build.lib.bazel.repository.downloader.DownloaderTestUtils.makeUrl;
+import static java.nio.charset.StandardCharsets.UTF_8;
+import static java.util.Arrays.asList;
+import static org.junit.Assert.fail;
+import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.anyLong;
+import static org.mockito.Matchers.anyString;
+import static org.mockito.Matchers.eq;
+import static org.mockito.Matchers.same;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoMoreInteractions;
+import static org.mockito.Mockito.verifyZeroInteractions;
+import static org.mockito.Mockito.when;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import com.google.devtools.build.lib.bazel.repository.downloader.RetryingInputStream.Reconnector;
+import com.google.devtools.build.lib.events.EventHandler;
+import com.google.devtools.build.lib.testutil.ManualClock;
+import com.google.devtools.build.lib.util.Sleeper;
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.io.InterruptedIOException;
+import java.net.URL;
+import java.net.URLConnection;
+import java.util.concurrent.CyclicBarrier;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.ExpectedException;
+import org.junit.rules.Timeout;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.stubbing.Answer;
+
+/** Unit tests for {@link HttpConnectorMultiplexer}. */
+@RunWith(JUnit4.class)
+@SuppressWarnings("unchecked")
+public class HttpConnectorMultiplexerTest {
+
+  private static final URL URL1 = makeUrl("http://first.example");
+  private static final URL URL2 = makeUrl("http://second.example");
+  private static final URL URL3 = makeUrl("http://third.example");
+  private static final byte[] data1 = "first".getBytes(UTF_8);
+  private static final byte[] data2 = "second".getBytes(UTF_8);
+  private static final byte[] data3 = "third".getBytes(UTF_8);
+
+  @Rule
+  public final ExpectedException thrown = ExpectedException.none();
+
+  @Rule
+  public final Timeout globalTimeout = new Timeout(10000);
+
+  private final HttpStream stream1 = fakeStream(URL1, data1);
+  private final HttpStream stream2 = fakeStream(URL2, data2);
+  private final HttpStream stream3 = fakeStream(URL3, data3);
+  private final ManualClock clock = new ManualClock();
+  private final Sleeper sleeper = mock(Sleeper.class);
+  private final HttpConnector connector = mock(HttpConnector.class);
+  private final URLConnection connection1 = mock(URLConnection.class);
+  private final URLConnection connection2 = mock(URLConnection.class);
+  private final URLConnection connection3 = mock(URLConnection.class);
+  private final EventHandler eventHandler = mock(EventHandler.class);
+  private final HttpStream.Factory streamFactory = mock(HttpStream.Factory.class);
+  private final HttpConnectorMultiplexer multiplexer =
+      new HttpConnectorMultiplexer(eventHandler, connector, streamFactory, clock, sleeper);
+
+  @Before
+  public void before() throws Exception {
+    when(connector.connect(eq(URL1), any(ImmutableMap.class))).thenReturn(connection1);
+    when(connector.connect(eq(URL2), any(ImmutableMap.class))).thenReturn(connection2);
+    when(connector.connect(eq(URL3), any(ImmutableMap.class))).thenReturn(connection3);
+    when(streamFactory
+            .create(same(connection1), any(URL.class), anyString(), any(Reconnector.class)))
+        .thenReturn(stream1);
+    when(streamFactory
+            .create(same(connection2), any(URL.class), anyString(), any(Reconnector.class)))
+        .thenReturn(stream2);
+    when(streamFactory
+            .create(same(connection3), any(URL.class), anyString(), any(Reconnector.class)))
+        .thenReturn(stream3);
+  }
+
+  @Test
+  public void emptyList_throwsIae() throws Exception {
+    thrown.expect(IllegalArgumentException.class);
+    multiplexer.connect(ImmutableList.<URL>of(), "");
+  }
+
+  @Test
+  public void ftpUrl_throwsIae() throws Exception {
+    thrown.expect(IllegalArgumentException.class);
+    multiplexer.connect(asList(new URL("ftp://lol.example")), "");
+  }
+
+  @Test
+  public void threadIsInterrupted_throwsIeProntoAndDoesNothingElse() throws Exception {
+    final AtomicBoolean wasInterrupted = new AtomicBoolean(true);
+    Thread task = new Thread(
+        new Runnable() {
+          @Override
+          public void run() {
+            Thread.currentThread().interrupt();
+            try {
+              multiplexer.connect(asList(new URL("http://lol.example")), "");
+            } catch (InterruptedIOException ignored) {
+              return;
+            } catch (Exception ignored) {
+              // ignored
+            }
+            wasInterrupted.set(false);
+          }
+        });
+    task.start();
+    task.join();
+    assertThat(wasInterrupted.get()).isTrue();
+    verifyZeroInteractions(connector);
+  }
+
+  @Test
+  public void singleUrl_justCallsConnector() throws Exception {
+    assertThat(toByteArray(multiplexer.connect(asList(URL1), "abc"))).isEqualTo(data1);
+    verify(connector).connect(eq(URL1), any(ImmutableMap.class));
+    verify(streamFactory)
+        .create(any(URLConnection.class), any(URL.class), eq("abc"), any(Reconnector.class));
+    verifyNoMoreInteractions(sleeper, connector, streamFactory);
+  }
+
+  @Test
+  public void multipleUrlsFail_throwsIOException() throws Exception {
+    when(connector.connect(any(URL.class), any(ImmutableMap.class))).thenThrow(new IOException());
+    try {
+      multiplexer.connect(asList(URL1, URL2, URL3), "");
+      fail("Expected IOException");
+    } catch (IOException e) {
+      assertThat(e.getMessage()).contains("All mirrors are down");
+    }
+    verify(connector, times(3)).connect(any(URL.class), any(ImmutableMap.class));
+    verify(sleeper, times(2)).sleepMillis(anyLong());
+    verifyNoMoreInteractions(sleeper, connector, streamFactory);
+  }
+
+  @Test
+  public void firstUrlFails_returnsSecond() throws Exception {
+    doAnswer(
+        new Answer<Void>() {
+          @Override
+          public Void answer(InvocationOnMock invocation) throws Throwable {
+            clock.advanceMillis(1000);
+            return null;
+          }
+        }).when(sleeper).sleepMillis(anyLong());
+    when(connector.connect(eq(URL1), any(ImmutableMap.class))).thenThrow(new IOException());
+    assertThat(toByteArray(multiplexer.connect(asList(URL1, URL2), "abc"))).isEqualTo(data2);
+    assertThat(clock.currentTimeMillis()).isEqualTo(1000L);
+    verify(connector).connect(eq(URL1), any(ImmutableMap.class));
+    verify(connector).connect(eq(URL2), any(ImmutableMap.class));
+    verify(streamFactory)
+        .create(any(URLConnection.class), any(URL.class), eq("abc"), any(Reconnector.class));
+    verify(sleeper).sleepMillis(anyLong());
+    verifyNoMoreInteractions(sleeper, connector, streamFactory);
+  }
+
+  @Test
+  public void twoSuccessfulUrlsAndFirstWins_returnsFirstAndInterruptsSecond() throws Exception {
+    final CyclicBarrier barrier = new CyclicBarrier(2);
+    final AtomicBoolean wasInterrupted = new AtomicBoolean(true);
+    when(connector.connect(eq(URL1), any(ImmutableMap.class))).thenAnswer(
+        new Answer<URLConnection>() {
+          @Override
+          public URLConnection answer(InvocationOnMock invocation) throws Throwable {
+            barrier.await();
+            return connection1;
+          }
+        });
+    doAnswer(
+        new Answer<Void>() {
+          @Override
+          public Void answer(InvocationOnMock invocation) throws Throwable {
+            barrier.await();
+            TimeUnit.MILLISECONDS.sleep(10000);
+            wasInterrupted.set(false);
+            return null;
+          }
+        }).when(sleeper).sleepMillis(anyLong());
+    assertThat(toByteArray(multiplexer.connect(asList(URL1, URL2), "abc"))).isEqualTo(data1);
+    assertThat(wasInterrupted.get()).isTrue();
+  }
+
+  @Test
+  public void parentThreadGetsInterrupted_interruptsChildrenThenThrowsIe() throws Exception {
+    final CyclicBarrier barrier = new CyclicBarrier(3);
+    final AtomicBoolean wasInterrupted1 = new AtomicBoolean(true);
+    final AtomicBoolean wasInterrupted2 = new AtomicBoolean(true);
+    final AtomicBoolean wasInterrupted3 = new AtomicBoolean(true);
+    when(connector.connect(eq(URL1), any(ImmutableMap.class))).thenAnswer(
+        new Answer<URLConnection>() {
+          @Override
+          public URLConnection answer(InvocationOnMock invocation) throws Throwable {
+            barrier.await();
+            TimeUnit.MILLISECONDS.sleep(10000);
+            wasInterrupted1.set(false);
+            throw new RuntimeException();
+          }
+        });
+    when(connector.connect(eq(URL2), any(ImmutableMap.class))).thenAnswer(
+        new Answer<URLConnection>() {
+          @Override
+          public URLConnection answer(InvocationOnMock invocation) throws Throwable {
+            barrier.await();
+            TimeUnit.MILLISECONDS.sleep(10000);
+            wasInterrupted2.set(false);
+            throw new RuntimeException();
+          }
+        });
+    Thread task = new Thread(
+        new Runnable() {
+          @Override
+          public void run() {
+            try {
+              multiplexer.connect(asList(URL1, URL2), "");
+            } catch (InterruptedIOException ignored) {
+              return;
+            } catch (Exception ignored) {
+              // ignored
+            }
+            wasInterrupted3.set(false);
+          }
+        });
+    task.start();
+    barrier.await();
+    task.interrupt();
+    task.join();
+    assertThat(wasInterrupted1.get()).isTrue();
+    assertThat(wasInterrupted2.get()).isTrue();
+    assertThat(wasInterrupted3.get()).isTrue();
+  }
+
+  private static HttpStream fakeStream(URL url, byte[] data) {
+    return new HttpStream(new ByteArrayInputStream(data), url);
+  }
+}
diff --git a/src/test/java/com/google/devtools/build/lib/bazel/repository/downloader/HttpConnectorTest.java b/src/test/java/com/google/devtools/build/lib/bazel/repository/downloader/HttpConnectorTest.java
index fa5a1b6bb6..ccc50196d7 100644
--- a/src/test/java/com/google/devtools/build/lib/bazel/repository/downloader/HttpConnectorTest.java
+++ b/src/test/java/com/google/devtools/build/lib/bazel/repository/downloader/HttpConnectorTest.java
@@ -15,41 +15,54 @@
 package com.google.devtools.build.lib.bazel.repository.downloader;
 
 import static com.google.common.truth.Truth.assertThat;
+import static com.google.devtools.build.lib.bazel.repository.downloader.DownloaderTestUtils.sendLines;
+import static com.google.devtools.build.lib.bazel.repository.downloader.HttpParser.readHttpRequest;
 import static java.nio.charset.StandardCharsets.ISO_8859_1;
+import static java.nio.charset.StandardCharsets.US_ASCII;
 import static java.nio.charset.StandardCharsets.UTF_8;
+import static org.junit.Assert.fail;
+import static org.mockito.Matchers.any;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
-import com.google.common.io.ByteSource;
+import com.google.common.collect.ImmutableMap;
 import com.google.common.io.ByteStreams;
 import com.google.common.io.CharStreams;
 import com.google.devtools.build.lib.events.EventHandler;
+import com.google.devtools.build.lib.testutil.ManualClock;
+import com.google.devtools.build.lib.testutil.ManualSleeper;
 import java.io.File;
 import java.io.FileOutputStream;
 import java.io.IOException;
+import java.io.InputStream;
 import java.io.InputStreamReader;
 import java.io.Reader;
-import java.net.HttpURLConnection;
 import java.net.InetAddress;
 import java.net.Proxy;
 import java.net.ServerSocket;
 import java.net.Socket;
 import java.net.URL;
+import java.net.URLConnection;
+import java.util.Locale;
+import java.util.Map;
 import java.util.concurrent.Callable;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CyclicBarrier;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
-import java.util.concurrent.Future;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
 import org.junit.After;
+import org.junit.Before;
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.rules.ExpectedException;
 import org.junit.rules.TemporaryFolder;
+import org.junit.rules.Timeout;
 import org.junit.runner.RunWith;
 import org.junit.runners.JUnit4;
 
-/**
- * Unit tests for {@link HttpConnector}.
- */
+/** Unit tests for {@link HttpConnector}. */
 @RunWith(JUnit4.class)
 public class HttpConnectorTest {
 
@@ -57,191 +70,368 @@ public class HttpConnectorTest {
   public final ExpectedException thrown = ExpectedException.none();
 
   @Rule
-  public TemporaryFolder testFolder = new TemporaryFolder();
+  public final TemporaryFolder testFolder = new TemporaryFolder();
 
-  private final ExecutorService executor = Executors.newSingleThreadExecutor();
-  private final HttpURLConnection connection = mock(HttpURLConnection.class);
+  @Rule
+  public final Timeout globalTimeout = new Timeout(10000);
+
+  private final ExecutorService executor = Executors.newFixedThreadPool(2);
+  private final ManualClock clock = new ManualClock();
+  private final ManualSleeper sleeper = new ManualSleeper(clock);
   private final EventHandler eventHandler = mock(EventHandler.class);
+  private final ProxyHelper proxyHelper = mock(ProxyHelper.class);
+  private final HttpConnector connector =
+      new HttpConnector(Locale.US, eventHandler, proxyHelper, sleeper);
+
+  @Before
+  public void before() throws Exception {
+    when(proxyHelper.createProxyIfNeeded(any(URL.class))).thenReturn(Proxy.NO_PROXY);
+  }
 
   @After
   public void after() throws Exception {
-    executor.shutdownNow();
+    executor.shutdown();
   }
 
   @Test
-  public void testLocalFileDownload() throws Exception {
+  public void localFileDownload() throws Exception {
     byte[] fileContents = "this is a test".getBytes(UTF_8);
     assertThat(
             ByteStreams.toByteArray(
-                HttpConnector.connect(
-                    createTempFile(fileContents).toURI().toURL(),
-                    Proxy.NO_PROXY,
-                    eventHandler)))
+                connector.connect(
+                        createTempFile(fileContents).toURI().toURL(),
+                        ImmutableMap.<String, String>of())
+                    .getInputStream()))
         .isEqualTo(fileContents);
   }
 
   @Test
-  public void missingLocationInRedirect_throwsIOException() throws Exception {
-    thrown.expect(IOException.class);
-    when(connection.getURL()).thenReturn(new URL("http://lol.example"));
-    HttpConnector.getLocation(connection);
-  }
-
-  @Test
-  public void absoluteLocationInRedirect_returnsNewUrl() throws Exception {
-    when(connection.getURL()).thenReturn(new URL("http://lol.example"));
-    when(connection.getHeaderField("Location")).thenReturn("http://new.example/hi");
-    assertThat(HttpConnector.getLocation(connection)).isEqualTo(new URL("http://new.example/hi"));
-  }
-
-  @Test
-  public void redirectOnlyHasPath_mergesHostFromOriginalUrl() throws Exception {
-    when(connection.getURL()).thenReturn(new URL("http://lol.example"));
-    when(connection.getHeaderField("Location")).thenReturn("/hi");
-    assertThat(HttpConnector.getLocation(connection)).isEqualTo(new URL("http://lol.example/hi"));
-  }
-
-  @Test
-  public void locationOnlyHasPathWithoutSlash_failsToMerge() throws Exception {
+  public void badHost_throwsIOException() throws Exception {
     thrown.expect(IOException.class);
-    thrown.expectMessage("Could not merge");
-    when(connection.getURL()).thenReturn(new URL("http://lol.example"));
-    when(connection.getHeaderField("Location")).thenReturn("omg");
-    HttpConnector.getLocation(connection);
-  }
-
-  @Test
-  public void locationHasFragment_prefersNewFragment() throws Exception {
-    when(connection.getURL()).thenReturn(new URL("http://lol.example#a"));
-    when(connection.getHeaderField("Location")).thenReturn("http://new.example/hi#b");
-    assertThat(HttpConnector.getLocation(connection)).isEqualTo(new URL("http://new.example/hi#b"));
+    thrown.expectMessage("Unknown host: bad.example");
+    connector.connect(new URL("http://bad.example"), ImmutableMap.<String, String>of());
   }
 
   @Test
-  public void locationHasNoFragmentButOriginalDoes_mergesOldFragment() throws Exception {
-    when(connection.getURL()).thenReturn(new URL("http://lol.example#a"));
-    when(connection.getHeaderField("Location")).thenReturn("http://new.example/hi");
-    assertThat(HttpConnector.getLocation(connection)).isEqualTo(new URL("http://new.example/hi#a"));
+  public void normalRequest() throws Exception {
+    final Map<String, String> headers = new ConcurrentHashMap<>();
+    try (ServerSocket server = new ServerSocket(0, 1, InetAddress.getByName("127.0.0.1"))) {
+      executor.submit(
+          new Callable<Object>() {
+            @Override
+            public Object call() throws Exception {
+              try (Socket socket = server.accept()) {
+                readHttpRequest(socket.getInputStream(), headers);
+                sendLines(socket,
+                    "HTTP/1.1 200 OK",
+                    "Date: Fri, 31 Dec 1999 23:59:59 GMT",
+                    "Connection: close",
+                    "Content-Type: text/plain",
+                    "Content-Length: 5",
+                    "",
+                    "hello");
+              }
+              return null;
+            }
+          });
+      try (Reader payload =
+              new InputStreamReader(
+                  connector.connect(
+                          new URL(String.format("http://127.0.0.1:%d/boo", server.getLocalPort())),
+                          ImmutableMap.of("Content-Encoding", "gzip"))
+                      .getInputStream(),
+                  ISO_8859_1)) {
+        assertThat(CharStreams.toString(payload)).isEqualTo("hello");
+      }
+    }
+    assertThat(headers).containsEntry("x-method", "GET");
+    assertThat(headers).containsEntry("x-request-uri", "/boo");
+    assertThat(headers).containsEntry("content-encoding", "gzip");
   }
 
   @Test
-  public void oldUrlHasPasswordRedirectingToSameDomain_mergesPassword() throws Exception {
-    when(connection.getURL()).thenReturn(new URL("http://a:b@lol.example"));
-    when(connection.getHeaderField("Location")).thenReturn("http://lol.example/hi");
-    assertThat(HttpConnector.getLocation(connection))
-        .isEqualTo(new URL("http://a:b@lol.example/hi"));
-    when(connection.getURL()).thenReturn(new URL("http://a:b@lol.example"));
-    when(connection.getHeaderField("Location")).thenReturn("/hi");
-    assertThat(HttpConnector.getLocation(connection))
-        .isEqualTo(new URL("http://a:b@lol.example/hi"));
+  public void serverError_retriesConnect() throws Exception {
+    try (ServerSocket server = new ServerSocket(0, 1, InetAddress.getByName("127.0.0.1"))) {
+      executor.submit(
+          new Callable<Object>() {
+            @Override
+            public Object call() throws Exception {
+              try (Socket socket = server.accept()) {
+                readHttpRequest(socket.getInputStream());
+                sendLines(socket,
+                    "HTTP/1.1 500 Incredible Catastrophe",
+                    "Date: Fri, 31 Dec 1999 23:59:59 GMT",
+                    "Connection: close",
+                    "Content-Type: text/plain",
+                    "Content-Length: 8",
+                    "",
+                    "nononono");
+              }
+              try (Socket socket = server.accept()) {
+                readHttpRequest(socket.getInputStream());
+                sendLines(socket,
+                    "HTTP/1.1 200 OK",
+                    "Date: Fri, 31 Dec 1999 23:59:59 GMT",
+                    "Connection: close",
+                    "Content-Type: text/plain",
+                    "Content-Length: 5",
+                    "",
+                    "hello");
+              }
+              return null;
+            }
+          });
+      try (Reader payload =
+              new InputStreamReader(
+                  connector.connect(
+                          new URL(String.format("http://127.0.0.1:%d", server.getLocalPort())),
+                          ImmutableMap.<String, String>of())
+                      .getInputStream(),
+                  ISO_8859_1)) {
+        assertThat(CharStreams.toString(payload)).isEqualTo("hello");
+        assertThat(clock.currentTimeMillis()).isEqualTo(100L);
+      }
+    }
   }
 
   @Test
-  public void oldUrlHasPasswordRedirectingToNewServer_doesntMergePassword() throws Exception {
-    when(connection.getURL()).thenReturn(new URL("http://a:b@lol.example"));
-    when(connection.getHeaderField("Location")).thenReturn("http://new.example/hi");
-    assertThat(HttpConnector.getLocation(connection)).isEqualTo(new URL("http://new.example/hi"));
-    when(connection.getURL()).thenReturn(new URL("http://a:b@lol.example"));
-    when(connection.getHeaderField("Location")).thenReturn("http://lol.example:81/hi");
-    assertThat(HttpConnector.getLocation(connection))
-        .isEqualTo(new URL("http://lol.example:81/hi"));
+  public void permanentError_doesNotRetryAndThrowsIOException() throws Exception {
+    try (ServerSocket server = new ServerSocket(0, 1, InetAddress.getByName("127.0.0.1"))) {
+      executor.submit(
+          new Callable<Object>() {
+            @Override
+            public Object call() throws Exception {
+              try (Socket socket = server.accept()) {
+                readHttpRequest(socket.getInputStream());
+                sendLines(socket,
+                    "HTTP/1.1 404 Not Here",
+                    "Date: Fri, 31 Dec 1999 23:59:59 GMT",
+                    "Connection: close",
+                    "Content-Type: text/plain",
+                    "Content-Length: 0",
+                    "",
+                    "");
+              }
+              return null;
+            }
+          });
+      thrown.expect(IOException.class);
+      thrown.expectMessage("404 Not Here");
+      connector.connect(
+          new URL(String.format("http://127.0.0.1:%d", server.getLocalPort())),
+          ImmutableMap.<String, String>of());
+    }
   }
 
   @Test
-  public void redirectToFtp_throwsIOException() throws Exception {
-    thrown.expect(IOException.class);
-    thrown.expectMessage("Bad Location");
-    when(connection.getURL()).thenReturn(new URL("http://lol.example"));
-    when(connection.getHeaderField("Location")).thenReturn("ftp://lol.example");
-    HttpConnector.getLocation(connection);
+  public void permanentError_consumesPayloadBeforeReturningn() throws Exception {
+    final CyclicBarrier barrier = new CyclicBarrier(2);
+    final AtomicBoolean consumed = new AtomicBoolean();
+    try (ServerSocket server = new ServerSocket(0, 1, InetAddress.getByName("127.0.0.1"))) {
+      executor.submit(
+          new Callable<Object>() {
+            @Override
+            public Object call() throws Exception {
+              try (Socket socket = server.accept()) {
+                readHttpRequest(socket.getInputStream());
+                sendLines(socket,
+                    "HTTP/1.1 501 Oh No",
+                    "Date: Fri, 31 Dec 1999 23:59:59 GMT",
+                    "Connection: close",
+                    "Content-Type: text/plain",
+                    "Content-Length: 1",
+                    "",
+                    "b");
+                consumed.set(true);
+              } finally {
+                barrier.await();
+              }
+              return null;
+            }
+          });
+      connector.connect(
+          new URL(String.format("http://127.0.0.1:%d", server.getLocalPort())),
+          ImmutableMap.<String, String>of());
+      fail();
+    } catch (IOException ignored) {
+      // ignored
+    } finally {
+      barrier.await();
+    }
+    assertThat(consumed.get()).isTrue();
+    assertThat(clock.currentTimeMillis()).isEqualTo(0L);
   }
 
   @Test
-  public void redirectToHttps_works() throws Exception {
-    when(connection.getURL()).thenReturn(new URL("http://lol.example"));
-    when(connection.getHeaderField("Location")).thenReturn("https://lol.example");
-    assertThat(HttpConnector.getLocation(connection)).isEqualTo(new URL("https://lol.example"));
+  public void always500_givesUpEventually() throws Exception {
+    final AtomicInteger tries = new AtomicInteger();
+    try (ServerSocket server = new ServerSocket(0, 1, InetAddress.getByName("127.0.0.1"))) {
+      executor.submit(new Callable<Object>() {
+            @Override
+            public Object call() throws Exception {
+              while (true) {
+                try (Socket socket = server.accept()) {
+                  readHttpRequest(socket.getInputStream());
+                  sendLines(socket,
+                      "HTTP/1.1 500 Oh My",
+                      "Date: Fri, 31 Dec 1999 23:59:59 GMT",
+                      "Connection: close",
+                      "Content-Type: text/plain",
+                      "Content-Length: 0",
+                      "",
+                      "");
+                  tries.incrementAndGet();
+                }
+              }
+            }
+          });
+      thrown.expect(IOException.class);
+      thrown.expectMessage("500 Oh My");
+      try {
+        connector.connect(
+            new URL(String.format("http://127.0.0.1:%d", server.getLocalPort())),
+            ImmutableMap.<String, String>of());
+      } finally {
+        assertThat(tries.get()).isGreaterThan(2);
+      }
+    }
   }
 
   @Test
-  public void testNormalRequest() throws Exception {
-    try (final ServerSocket server = new ServerSocket(0, 1, InetAddress.getByName("127.0.0.1"))) {
-      Future<Void> thread =
-          executor.submit(
-              new Callable<Void>() {
-                @Override
-                public Void call() throws Exception {
-                  try (Socket socket = server.accept()) {
-                    send(socket,
-                        "HTTP/1.1 200 OK\r\n"
-                            + "Date: Fri, 31 Dec 1999 23:59:59 GMT\r\n"
-                            + "Content-Type: text/plain\r\n"
-                            + "Content-Length: 5\r\n"
-                            + "\r\n"
-                            + "hello");
-                  }
-                  return null;
+  public void serverSays403_clientRetriesAnyway() throws Exception {
+    final AtomicInteger tries = new AtomicInteger();
+    try (ServerSocket server = new ServerSocket(0, 1, InetAddress.getByName("127.0.0.1"))) {
+      executor.submit(
+          new Callable<Object>() {
+            @Override
+            public Object call() throws Exception {
+              while (true) {
+                try (Socket socket = server.accept()) {
+                  readHttpRequest(socket.getInputStream());
+                  sendLines(socket,
+                      "HTTP/1.1 403 Forbidden",
+                      "Date: Fri, 31 Dec 1999 23:59:59 GMT",
+                      "Connection: close",
+                      "Content-Type: text/plain",
+                      "Content-Length: 0",
+                      "",
+                      "");
+                  tries.incrementAndGet();
                 }
-              });
-      try (Reader payload =
-              new InputStreamReader(
-                  HttpConnector.connect(
-                      new URL(String.format("http://127.0.0.1:%d", server.getLocalPort())),
-                      Proxy.NO_PROXY,
-                      eventHandler),
-                  ISO_8859_1)) {
-        assertThat(CharStreams.toString(payload)).isEqualTo("hello");
+              }
+            }
+          });
+      thrown.expect(IOException.class);
+      thrown.expectMessage("403 Forbidden");
+      try {
+        connector.connect(
+            new URL(String.format("http://127.0.0.1:%d", server.getLocalPort())),
+            ImmutableMap.<String, String>of());
+      } finally {
+        assertThat(tries.get()).isGreaterThan(2);
       }
-      thread.get();
     }
   }
 
   @Test
-  public void testRetry() throws Exception {
-    try (final ServerSocket server = new ServerSocket(0, 1, InetAddress.getByName("127.0.0.1"))) {
-      Future<Void> thread =
-          executor.submit(
-              new Callable<Void>() {
-                @Override
-                public Void call() throws Exception {
-                  try (Socket socket = server.accept()) {
-                    send(socket,
-                        "HTTP/1.1 500 Incredible Catastrophe\r\n"
-                            + "Date: Fri, 31 Dec 1999 23:59:59 GMT\r\n"
-                            + "Content-Type: text/plain\r\n"
-                            + "Content-Length: 8\r\n"
-                            + "\r\n"
-                            + "nononono");
-                  }
-                  try (Socket socket = server.accept()) {
-                    send(socket,
-                        "HTTP/1.1 200 OK\r\n"
-                            + "Date: Fri, 31 Dec 1999 23:59:59 GMT\r\n"
-                            + "Content-Type: text/plain\r\n"
-                            + "Content-Length: 5\r\n"
-                            + "\r\n"
-                            + "hello");
-                  }
-                  return null;
-                }
-              });
-      try (Reader payload =
-              new InputStreamReader(
-                  HttpConnector.connect(
-                      new URL(String.format("http://127.0.0.1:%d", server.getLocalPort())),
-                      Proxy.NO_PROXY,
-                      eventHandler),
-                  ISO_8859_1)) {
-        assertThat(CharStreams.toString(payload)).isEqualTo("hello");
+  public void redirectToDifferentPath_works() throws Exception {
+    final Map<String, String> headers1 = new ConcurrentHashMap<>();
+    final Map<String, String> headers2 = new ConcurrentHashMap<>();
+    try (ServerSocket server = new ServerSocket(0, 1, InetAddress.getByName("127.0.0.1"))) {
+      executor.submit(
+          new Callable<Object>() {
+            @Override
+            public Object call() throws Exception {
+              try (Socket socket = server.accept()) {
+                readHttpRequest(socket.getInputStream(), headers1);
+                sendLines(socket,
+                    "HTTP/1.1 301 Redirect",
+                    "Date: Fri, 31 Dec 1999 23:59:59 GMT",
+                    "Connection: close",
+                    "Location: /doodle.tar.gz",
+                    "Content-Length: 0",
+                    "",
+                    "");
+              }
+              try (Socket socket = server.accept()) {
+                readHttpRequest(socket.getInputStream(), headers2);
+                sendLines(socket,
+                    "HTTP/1.1 200 OK",
+                    "Date: Fri, 31 Dec 1999 23:59:59 GMT",
+                    "Connection: close",
+                    "Content-Type: text/plain",
+                    "Content-Length: 0",
+                    "",
+                    "");
+              }
+              return null;
+            }
+          });
+      URLConnection connection =
+          connector.connect(
+              new URL(String.format("http://127.0.0.1:%d", server.getLocalPort())),
+              ImmutableMap.<String, String>of());
+      assertThat(connection.getURL()).isEqualTo(
+          new URL(String.format("http://127.0.0.1:%d/doodle.tar.gz", server.getLocalPort())));
+      try (InputStream input = connection.getInputStream()) {
+        assertThat(ByteStreams.toByteArray(input)).isEmpty();
       }
-      thread.get();
     }
+    assertThat(headers1).containsEntry("x-request-uri", "/");
+    assertThat(headers2).containsEntry("x-request-uri", "/doodle.tar.gz");
   }
 
-  private static void send(Socket socket, String data) throws IOException {
-    ByteStreams.copy(
-        ByteSource.wrap(data.getBytes(ISO_8859_1)).openStream(),
-        socket.getOutputStream());
+  @Test
+  public void redirectToDifferentServer_works() throws Exception {
+    try (ServerSocket server1 = new ServerSocket(0, 1, InetAddress.getByName("127.0.0.1"));
+        ServerSocket server2 = new ServerSocket(0, 1, InetAddress.getByName("127.0.0.1"))) {
+      executor.submit(
+          new Callable<Object>() {
+            @Override
+            public Object call() throws Exception {
+              try (Socket socket = server1.accept()) {
+                readHttpRequest(socket.getInputStream());
+                sendLines(socket,
+                    "HTTP/1.1 301 Redirect",
+                    "Date: Fri, 31 Dec 1999 23:59:59 GMT",
+                    "Connection: close",
+                    String.format("Location: http://127.0.0.1:%d/doodle.tar.gz",
+                        server2.getLocalPort()),
+                    "Content-Length: 0",
+                    "",
+                    "");
+              }
+              return null;
+            }
+          });
+      executor.submit(
+          new Callable<Object>() {
+            @Override
+            public Object call() throws Exception {
+              try (Socket socket = server2.accept()) {
+                readHttpRequest(socket.getInputStream());
+                sendLines(socket,
+                    "HTTP/1.1 200 OK",
+                    "Date: Fri, 31 Dec 1999 23:59:59 GMT",
+                    "Connection: close",
+                    "Content-Type: text/plain",
+                    "Content-Length: 5",
+                    "",
+                    "hello");
+              }
+              return null;
+            }
+          });
+      URLConnection connection =
+          connector.connect(
+              new URL(String.format("http://127.0.0.1:%d", server1.getLocalPort())),
+              ImmutableMap.<String, String>of());
+      assertThat(connection.getURL()).isEqualTo(
+          new URL(String.format("http://127.0.0.1:%d/doodle.tar.gz", server2.getLocalPort())));
+      try (InputStream input = connection.getInputStream()) {
+        assertThat(ByteStreams.toByteArray(input)).isEqualTo("hello".getBytes(US_ASCII));
+      }
+    }
   }
 
   private File createTempFile(byte[] fileContents) throws IOException {
diff --git a/src/test/java/com/google/devtools/build/lib/bazel/repository/downloader/HttpParser.java b/src/test/java/com/google/devtools/build/lib/bazel/repository/downloader/HttpParser.java
new file mode 100644
index 0000000000..c901367aca
--- /dev/null
+++ b/src/test/java/com/google/devtools/build/lib/bazel/repository/downloader/HttpParser.java
@@ -0,0 +1,154 @@
+// Copyright 2016 The Bazel Authors. All rights reserved.
+//
+// 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.devtools.build.lib.bazel.repository.downloader;
+
+import com.google.common.base.Ascii;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.HashMap;
+import java.util.Map;
+
+/** Utility class for parsing HTTP messages. */
+final class HttpParser {
+
+  /** Exhausts request line and headers of HTTP request. */
+  static void readHttpRequest(InputStream stream) throws IOException {
+    readHttpRequest(stream, new HashMap<String, String>());
+  }
+
+  /**
+   * Parses request line and headers of HTTP request.
+   *
+   * <p>This parser is correct and extremely lax. This implementation is Θ(n) and the stream should
+   * be buffered. All decoding is ISO-8859-1. A 1mB upper bound on memory is enforced.
+   *
+   * @throws IOException if reading failed or premature end of stream encountered
+   * @throws HttpParserError if 400 error should be sent to client and connection must be closed
+   */
+  static void readHttpRequest(InputStream stream, Map<String, String> output) throws IOException {
+    StringBuilder builder = new StringBuilder(256);
+    State state = State.METHOD;
+    String key = "";
+    int toto = 0;
+    while (true) {
+      int c = stream.read();
+      if (c == -1) {
+        throw new IOException();  // RFC7230 § 3.4
+      }
+      if (++toto == 1024 * 1024) {
+        throw new HttpParserError();  // RFC7230 § 3.2.5
+      }
+      switch (state) {
+        case METHOD:
+          if (c == ' ') {
+            if (builder.length() == 0) {
+              throw new HttpParserError();
+            }
+            output.put("x-method", builder.toString());
+            builder.setLength(0);
+            state = State.URI;
+          } else if (c == '\r' || c == '\n') {
+            break;  // RFC7230 § 3.5
+          } else {
+            builder.append(Ascii.toUpperCase((char) c));
+          }
+          break;
+        case URI:
+          if (c == ' ') {
+            if (builder.length() == 0) {
+              throw new HttpParserError();
+            }
+            output.put("x-request-uri", builder.toString());
+            builder.setLength(0);
+            state = State.VERSION;
+          } else {
+            builder.append((char) c);
+          }
+          break;
+        case VERSION:
+          if (c == '\r' || c == '\n') {
+            output.put("x-version", builder.toString());
+            builder.setLength(0);
+            state = c == '\r' ? State.CR1 : State.LF1;
+          } else {
+            builder.append(Ascii.toUpperCase((char) c));
+          }
+          break;
+        case CR1:
+          if (c == '\n') {
+            state = State.LF1;
+            break;
+          }
+          throw new HttpParserError();
+        case LF1:
+          if (c == '\r') {
+            state = State.LF2;
+            break;
+          } else if (c == '\n') {
+            return;
+          } else if (c == ' ' || c == '\t') {
+            throw new HttpParserError("Line folding unacceptable");  // RFC7230 § 3.2.4
+          }
+          state = State.HKEY;
+          // epsilon transition
+        case HKEY:
+          if (c == ':') {
+            key = builder.toString();
+            builder.setLength(0);
+            state = State.HSEP;
+          } else {
+            builder.append(Ascii.toLowerCase((char) c));
+          }
+          break;
+        case HSEP:
+          if (c == ' ' || c == '\t') {
+            break;
+          }
+          state = State.HVAL;
+          // epsilon transition
+        case HVAL:
+          if (c == '\r' || c == '\n') {
+            output.put(key, builder.toString());
+            builder.setLength(0);
+            state = c == '\r' ? State.CR1 : State.LF1;
+          } else {
+            builder.append((char) c);
+          }
+          break;
+        case LF2:
+          if (c == '\n') {
+            return;
+          }
+          throw new HttpParserError();
+        default:
+          throw new AssertionError();
+      }
+    }
+  }
+
+  static final class HttpParserError extends IOException {
+    HttpParserError() {
+      this("Malformed Request");
+    }
+
+    HttpParserError(String messageForClient) {
+      super(messageForClient);
+    }
+  }
+
+  private enum State { METHOD, URI, VERSION, HKEY, HSEP, HVAL, CR1, LF1, LF2 }
+
+  private HttpParser() {}
+}
diff --git a/src/test/java/com/google/devtools/build/lib/bazel/repository/downloader/HttpStreamTest.java b/src/test/java/com/google/devtools/build/lib/bazel/repository/downloader/HttpStreamTest.java
new file mode 100644
index 0000000000..ac0351ed18
--- /dev/null
+++ b/src/test/java/com/google/devtools/build/lib/bazel/repository/downloader/HttpStreamTest.java
@@ -0,0 +1,195 @@
+// Copyright 2016 The Bazel Authors. All rights reserved.
+//
+// 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.devtools.build.lib.bazel.repository.downloader;
+
+import static com.google.common.io.ByteStreams.toByteArray;
+import static com.google.common.truth.Truth.assertThat;
+import static com.google.devtools.build.lib.bazel.repository.downloader.DownloaderTestUtils.makeUrl;
+import static java.nio.charset.StandardCharsets.UTF_8;
+import static org.junit.Assert.fail;
+import static org.mockito.Matchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import com.google.common.hash.Hashing;
+import com.google.common.io.ByteStreams;
+import com.google.devtools.build.lib.bazel.repository.downloader.RetryingInputStream.Reconnector;
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InterruptedIOException;
+import java.io.OutputStream;
+import java.net.HttpURLConnection;
+import java.net.URL;
+import java.util.Random;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.zip.GZIPOutputStream;
+import java.util.zip.ZipException;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.ExpectedException;
+import org.junit.rules.Timeout;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.stubbing.Answer;
+
+/** Integration tests for {@link HttpStream.Factory} and friends. */
+@RunWith(JUnit4.class)
+public class HttpStreamTest {
+
+  private static final Random randoCalrissian = new Random();
+  private static final byte[] data = "hello".getBytes(UTF_8);
+  private static final String GOOD_CHECKSUM =
+      "2cf24dba5fb0a30e26e83b2ac5b9e29e1b161e5c1fa7425e73043362938b9824";
+  private static final String BAD_CHECKSUM =
+      "0000000000000000000000000000000000000000000000000000000000000000";
+  private static final URL AURL = makeUrl("http://doodle.example");
+
+  @Rule
+  public final ExpectedException thrown = ExpectedException.none();
+
+  @Rule
+  public final Timeout globalTimeout = new Timeout(10000);
+
+  private final HttpURLConnection connection = mock(HttpURLConnection.class);
+  private final Reconnector reconnector = mock(Reconnector.class);
+  private final ProgressInputStream.Factory progress = mock(ProgressInputStream.Factory.class);
+  private final HttpStream.Factory streamFactory = new HttpStream.Factory(progress);
+
+  @Before
+  public void before() throws Exception {
+    when(connection.getInputStream()).thenReturn(new ByteArrayInputStream(data));
+    when(progress.create(any(InputStream.class), any(URL.class), any(URL.class))).thenAnswer(
+        new Answer<InputStream>() {
+          @Override
+          public InputStream answer(InvocationOnMock invocation) throws Throwable {
+            return (InputStream) invocation.getArguments()[0];
+          }
+        });
+  }
+
+  @Test
+  public void noChecksum_readsOk() throws Exception {
+    try (HttpStream stream = streamFactory.create(connection, AURL, "", reconnector)) {
+      assertThat(toByteArray(stream)).isEqualTo(data);
+    }
+  }
+
+  @Test
+  public void smallDataWithValidChecksum_readsOk() throws Exception {
+    try (HttpStream stream = streamFactory.create(connection, AURL, GOOD_CHECKSUM, reconnector)) {
+      assertThat(toByteArray(stream)).isEqualTo(data);
+    }
+  }
+
+  @Test
+  public void smallDataWithInvalidChecksum_throwsIOExceptionInCreatePhase() throws Exception {
+    thrown.expect(IOException.class);
+    thrown.expectMessage("Checksum");
+    streamFactory.create(connection, AURL, BAD_CHECKSUM, reconnector);
+  }
+
+  @Test
+  public void bigDataWithValidChecksum_readsOk() throws Exception {
+    // at google, we know big data
+    byte[] bigData = new byte[HttpStream.PRECHECK_BYTES + 70001];
+    randoCalrissian.nextBytes(bigData);
+    when(connection.getInputStream()).thenReturn(new ByteArrayInputStream(bigData));
+    try (HttpStream stream =
+            streamFactory.create(
+                connection, AURL, Hashing.sha256().hashBytes(bigData).toString(), reconnector)) {
+      assertThat(toByteArray(stream)).isEqualTo(bigData);
+    }
+  }
+
+  @Test
+  public void bigDataWithInvalidChecksum_throwsIOExceptionAfterCreateOnEof() throws Exception {
+    // the probability of this test flaking is 8.6361686e-78
+    byte[] bigData = new byte[HttpStream.PRECHECK_BYTES + 70001];
+    randoCalrissian.nextBytes(bigData);
+    when(connection.getInputStream()).thenReturn(new ByteArrayInputStream(bigData));
+    try (HttpStream stream = streamFactory.create(connection, AURL, BAD_CHECKSUM, reconnector)) {
+      thrown.expect(IOException.class);
+      thrown.expectMessage("Checksum");
+      toByteArray(stream);
+      fail("Should have thrown error before close()");
+    }
+  }
+
+  @Test
+  public void httpServerSaidGzippedButNotGzipped_throwsZipExceptionInCreate() throws Exception {
+    when(connection.getURL()).thenReturn(AURL);
+    when(connection.getContentEncoding()).thenReturn("gzip");
+    thrown.expect(ZipException.class);
+    streamFactory.create(connection, AURL, "", reconnector);
+  }
+
+  @Test
+  public void javascriptGzippedInTransit_automaticallyGunzips() throws Exception {
+    when(connection.getURL()).thenReturn(AURL);
+    when(connection.getContentEncoding()).thenReturn("x-gzip");
+    when(connection.getInputStream()).thenReturn(new ByteArrayInputStream(gzipData(data)));
+    try (HttpStream stream = streamFactory.create(connection, AURL, "", reconnector)) {
+      assertThat(toByteArray(stream)).isEqualTo(data);
+    }
+  }
+
+  @Test
+  public void serverSaysTarballPathIsGzipped_doesntAutomaticallyGunzip() throws Exception {
+    byte[] gzData = gzipData(data);
+    when(connection.getURL()).thenReturn(new URL("http://doodle.example/foo.tar.gz"));
+    when(connection.getContentEncoding()).thenReturn("gzip");
+    when(connection.getInputStream()).thenReturn(new ByteArrayInputStream(gzData));
+    try (HttpStream stream = streamFactory.create(connection, AURL, "", reconnector)) {
+      assertThat(toByteArray(stream)).isEqualTo(gzData);
+    }
+  }
+
+  @Test
+  public void threadInterrupted_haltsReadingAndThrowsInterrupt() throws Exception {
+    final AtomicBoolean wasInterrupted = new AtomicBoolean();
+    Thread thread = new Thread(
+        new Runnable() {
+          @Override
+          public void run() {
+            try (HttpStream stream = streamFactory.create(connection, AURL, "", reconnector)) {
+              stream.read();
+              Thread.currentThread().interrupt();
+              stream.read();
+              fail();
+            } catch (InterruptedIOException expected) {
+              wasInterrupted.set(true);
+            } catch (IOException ignored) {
+              // ignored
+            }
+          }
+        });
+    thread.start();
+    thread.join();
+    assertThat(wasInterrupted.get()).isTrue();
+  }
+
+  private static byte[] gzipData(byte[] bytes) throws IOException {
+    ByteArrayOutputStream baos = new ByteArrayOutputStream();
+    try (InputStream input = new ByteArrayInputStream(bytes);
+        OutputStream output = new GZIPOutputStream(baos)) {
+      ByteStreams.copy(input, output);
+    }
+    return baos.toByteArray();
+  }
+}
diff --git a/src/test/java/com/google/devtools/build/lib/bazel/repository/downloader/HttpUtilsTest.java b/src/test/java/com/google/devtools/build/lib/bazel/repository/downloader/HttpUtilsTest.java
new file mode 100644
index 0000000000..b79cbc5244
--- /dev/null
+++ b/src/test/java/com/google/devtools/build/lib/bazel/repository/downloader/HttpUtilsTest.java
@@ -0,0 +1,131 @@
+// Copyright 2016 The Bazel Authors. All rights reserved.
+//
+// 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.devtools.build.lib.bazel.repository.downloader;
+
+import static com.google.common.truth.Truth.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import java.io.IOException;
+import java.net.HttpURLConnection;
+import java.net.URL;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.ExpectedException;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+/** Unit tests for {@link HttpUtils}. */
+@RunWith(JUnit4.class)
+public class HttpUtilsTest {
+
+  @Rule
+  public final ExpectedException thrown = ExpectedException.none();
+
+  private final HttpURLConnection connection = mock(HttpURLConnection.class);
+
+  @Test
+  public void getExtension_twoExtensions_returnsLast() throws Exception {
+    assertThat(HttpUtils.getExtension("doodle.tar.gz")).isEqualTo("gz");
+  }
+
+  @Test
+  public void getExtension_isUppercase_returnsLowered() throws Exception {
+    assertThat(HttpUtils.getExtension("DOODLE.TXT")).isEqualTo("txt");
+  }
+
+  @Test
+  public void getLocation_missingInRedirect_throwsIOException() throws Exception {
+    thrown.expect(IOException.class);
+    when(connection.getURL()).thenReturn(new URL("http://lol.example"));
+    HttpUtils.getLocation(connection);
+  }
+
+  @Test
+  public void getLocation_absoluteInRedirect_returnsNewUrl() throws Exception {
+    when(connection.getURL()).thenReturn(new URL("http://lol.example"));
+    when(connection.getHeaderField("Location")).thenReturn("http://new.example/hi");
+    assertThat(HttpUtils.getLocation(connection)).isEqualTo(new URL("http://new.example/hi"));
+  }
+
+  @Test
+  public void getLocation_redirectOnlyHasPath_mergesHostFromOriginalUrl() throws Exception {
+    when(connection.getURL()).thenReturn(new URL("http://lol.example"));
+    when(connection.getHeaderField("Location")).thenReturn("/hi");
+    assertThat(HttpUtils.getLocation(connection)).isEqualTo(new URL("http://lol.example/hi"));
+  }
+
+  @Test
+  public void getLocation_onlyHasPathWithoutSlash_failsToMerge() throws Exception {
+    thrown.expect(IOException.class);
+    thrown.expectMessage("Could not merge");
+    when(connection.getURL()).thenReturn(new URL("http://lol.example"));
+    when(connection.getHeaderField("Location")).thenReturn("omg");
+    HttpUtils.getLocation(connection);
+  }
+
+  @Test
+  public void getLocation_hasFragment_prefersNewFragment() throws Exception {
+    when(connection.getURL()).thenReturn(new URL("http://lol.example#a"));
+    when(connection.getHeaderField("Location")).thenReturn("http://new.example/hi#b");
+    assertThat(HttpUtils.getLocation(connection)).isEqualTo(new URL("http://new.example/hi#b"));
+  }
+
+  @Test
+  public void getLocation_hasNoFragmentButOriginalDoes_mergesOldFragment() throws Exception {
+    when(connection.getURL()).thenReturn(new URL("http://lol.example#a"));
+    when(connection.getHeaderField("Location")).thenReturn("http://new.example/hi");
+    assertThat(HttpUtils.getLocation(connection)).isEqualTo(new URL("http://new.example/hi#a"));
+  }
+
+  @Test
+  public void getLocation_oldUrlHasPassRedirectingToSameDomain_mergesPassword() throws Exception {
+    when(connection.getURL()).thenReturn(new URL("http://a:b@lol.example"));
+    when(connection.getHeaderField("Location")).thenReturn("http://lol.example/hi");
+    assertThat(HttpUtils.getLocation(connection))
+        .isEqualTo(new URL("http://a:b@lol.example/hi"));
+    when(connection.getURL()).thenReturn(new URL("http://a:b@lol.example"));
+    when(connection.getHeaderField("Location")).thenReturn("/hi");
+    assertThat(HttpUtils.getLocation(connection))
+        .isEqualTo(new URL("http://a:b@lol.example/hi"));
+  }
+
+  @Test
+  public void getLocation_oldUrlHasPasswordRedirectingToNewServer_doesntMerge() throws Exception {
+    when(connection.getURL()).thenReturn(new URL("http://a:b@lol.example"));
+    when(connection.getHeaderField("Location")).thenReturn("http://new.example/hi");
+    assertThat(HttpUtils.getLocation(connection)).isEqualTo(new URL("http://new.example/hi"));
+    when(connection.getURL()).thenReturn(new URL("http://a:b@lol.example"));
+    when(connection.getHeaderField("Location")).thenReturn("http://lol.example:81/hi");
+    assertThat(HttpUtils.getLocation(connection))
+        .isEqualTo(new URL("http://lol.example:81/hi"));
+  }
+
+  @Test
+  public void getLocation_redirectToFtp_throwsIOException() throws Exception {
+    thrown.expect(IOException.class);
+    thrown.expectMessage("Bad Location");
+    when(connection.getURL()).thenReturn(new URL("http://lol.example"));
+    when(connection.getHeaderField("Location")).thenReturn("ftp://lol.example");
+    HttpUtils.getLocation(connection);
+  }
+
+  @Test
+  public void getLocation_redirectToHttps_works() throws Exception {
+    when(connection.getURL()).thenReturn(new URL("http://lol.example"));
+    when(connection.getHeaderField("Location")).thenReturn("https://lol.example");
+    assertThat(HttpUtils.getLocation(connection)).isEqualTo(new URL("https://lol.example"));
+  }
+}
diff --git a/src/test/java/com/google/devtools/build/lib/bazel/repository/downloader/ProgressInputStreamTest.java b/src/test/java/com/google/devtools/build/lib/bazel/repository/downloader/ProgressInputStreamTest.java
new file mode 100644
index 0000000000..2e047f773f
--- /dev/null
+++ b/src/test/java/com/google/devtools/build/lib/bazel/repository/downloader/ProgressInputStreamTest.java
@@ -0,0 +1,149 @@
+// Copyright 2016 The Bazel Authors. All rights reserved.
+//
+// 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.devtools.build.lib.bazel.repository.downloader;
+
+import static com.google.common.truth.Truth.assertThat;
+import static com.google.devtools.build.lib.bazel.repository.downloader.DownloaderTestUtils.makeUrl;
+import static org.junit.Assert.fail;
+import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.anyInt;
+import static org.mockito.Matchers.eq;
+import static org.mockito.Matchers.same;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoMoreInteractions;
+import static org.mockito.Mockito.when;
+
+import com.google.devtools.build.lib.events.Event;
+import com.google.devtools.build.lib.events.EventHandler;
+import com.google.devtools.build.lib.testutil.ManualClock;
+import java.io.IOException;
+import java.io.InputStream;
+import java.net.URL;
+import java.util.Locale;
+import org.junit.After;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+/** Unit tests for {@link ProgressInputStream}. */
+@RunWith(JUnit4.class)
+public class ProgressInputStreamTest {
+
+  private final ManualClock clock = new ManualClock();
+  private final EventHandler eventHandler = mock(EventHandler.class);
+  private final InputStream delegate = mock(InputStream.class);
+  private final URL url = makeUrl("http://lol.example");
+  private ProgressInputStream stream =
+      new ProgressInputStream(Locale.US, clock, eventHandler, 1, delegate, url, url);
+
+  @After
+  public void after() throws Exception {
+    verifyNoMoreInteractions(eventHandler, delegate);
+  }
+
+  @Test
+  public void close_callsDelegate() throws Exception {
+    stream.close();
+    verify(delegate).close();
+  }
+
+  @Test
+  public void available_callsDelegate() throws Exception {
+    stream.available();
+    verify(delegate).available();
+  }
+
+  @Test
+  public void read_callsdelegate() throws Exception {
+    stream.read();
+    verify(delegate).read();
+  }
+
+  @Test
+  public void readThrowsException_passesThrough() throws Exception {
+    when(delegate.read()).thenThrow(new IOException());
+    try {
+      stream.read();
+      fail("Expected IOException");
+    } catch (IOException expected) {
+      verify(delegate).read();
+    }
+  }
+
+  @Test
+  public void readsAfterInterval_emitsProgressOnce() throws Exception {
+    when(delegate.read()).thenReturn(42);
+    assertThat(stream.read()).isEqualTo(42);
+    clock.advanceMillis(1);
+    assertThat(stream.read()).isEqualTo(42);
+    assertThat(stream.read()).isEqualTo(42);
+    verify(delegate, times(3)).read();
+    verify(eventHandler).handle(Event.progress("Downloading http://lol.example: 2 bytes"));
+  }
+
+  @Test
+  public void multipleIntervalsElapsed_showsMultipleProgress() throws Exception {
+    stream.read();
+    stream.read();
+    clock.advanceMillis(1);
+    stream.read();
+    stream.read();
+    clock.advanceMillis(1);
+    stream.read();
+    stream.read();
+    verify(delegate, times(6)).read();
+    verify(eventHandler).handle(Event.progress("Downloading http://lol.example: 3 bytes"));
+    verify(eventHandler).handle(Event.progress("Downloading http://lol.example: 5 bytes"));
+  }
+
+  @Test
+  public void bufferReadsAfterInterval_emitsProgressOnce() throws Exception {
+    byte[] buffer = new byte[1024];
+    when(delegate.read(any(byte[].class), anyInt(), anyInt())).thenReturn(1024);
+    assertThat(stream.read(buffer)).isEqualTo(1024);
+    clock.advanceMillis(1);
+    assertThat(stream.read(buffer)).isEqualTo(1024);
+    assertThat(stream.read(buffer)).isEqualTo(1024);
+    verify(delegate, times(3)).read(same(buffer), eq(0), eq(1024));
+    verify(eventHandler).handle(Event.progress("Downloading http://lol.example: 2,048 bytes"));
+  }
+
+  @Test
+  public void bufferReadsAfterIntervalInGermany_usesPeriodAsSeparator() throws Exception {
+    stream = new ProgressInputStream(Locale.GERMANY, clock, eventHandler, 1, delegate, url, url);
+    byte[] buffer = new byte[1024];
+    when(delegate.read(any(byte[].class), anyInt(), anyInt())).thenReturn(1024);
+    clock.advanceMillis(1);
+    stream.read(buffer);
+    verify(delegate).read(same(buffer), eq(0), eq(1024));
+    verify(eventHandler).handle(Event.progress("Downloading http://lol.example: 1.024 bytes"));
+  }
+
+  @Test
+  public void redirectedToDifferentServer_showsOriginalUrlWithVia() throws Exception {
+    stream = new ProgressInputStream(
+        Locale.US, clock, eventHandler, 1, delegate, new URL("http://cdn.example/foo"), url);
+    when(delegate.read()).thenReturn(42);
+    assertThat(stream.read()).isEqualTo(42);
+    clock.advanceMillis(1);
+    assertThat(stream.read()).isEqualTo(42);
+    assertThat(stream.read()).isEqualTo(42);
+    verify(delegate, times(3)).read();
+    verify(eventHandler).handle(
+        Event.progress("Downloading http://lol.example via cdn.example: 2 bytes"));
+  }
+}
diff --git a/src/test/java/com/google/devtools/build/lib/bazel/repository/downloader/ProxyHelperTest.java b/src/test/java/com/google/devtools/build/lib/bazel/repository/downloader/ProxyHelperTest.java
index 2a725c195a..6063ae5054 100644
--- a/src/test/java/com/google/devtools/build/lib/bazel/repository/downloader/ProxyHelperTest.java
+++ b/src/test/java/com/google/devtools/build/lib/bazel/repository/downloader/ProxyHelperTest.java
@@ -21,46 +21,43 @@ import static org.junit.Assert.fail;
 import com.google.common.collect.ImmutableMap;
 import java.io.IOException;
 import java.net.Proxy;
+import java.net.URL;
 import java.util.Map;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.junit.runners.JUnit4;
 
 /**
- * Tests for @{link ProxyHelper}.
+ * Tests for {@link ProxyHelper}.
  */
 @RunWith(JUnit4.class)
 public class ProxyHelperTest {
 
   @Test
   public void testCreateIfNeededHttpLowerCase() throws Exception {
-    Map<String, String> env = ImmutableMap.<String, String>builder()
-        .put("http_proxy", "http://my.example.com").build();
-    Proxy proxy = ProxyHelper.createProxyIfNeeded("http://www.something.com", env);
+    ProxyHelper helper = new ProxyHelper(ImmutableMap.of("http_proxy", "http://my.example.com"));
+    Proxy proxy = helper.createProxyIfNeeded(new URL("http://www.something.com"));
     assertThat(proxy.toString()).endsWith("my.example.com:80");
   }
 
   @Test
   public void testCreateIfNeededHttpUpperCase() throws Exception {
-    Map<String, String> env = ImmutableMap.<String, String>builder()
-        .put("HTTP_PROXY", "http://my.example.com").build();
-    Proxy proxy = ProxyHelper.createProxyIfNeeded("http://www.something.com", env);
+    ProxyHelper helper = new ProxyHelper(ImmutableMap.of("HTTP_PROXY", "http://my.example.com"));
+    Proxy proxy = helper.createProxyIfNeeded(new URL("http://www.something.com"));
     assertThat(proxy.toString()).endsWith("my.example.com:80");
   }
 
   @Test
   public void testCreateIfNeededHttpsLowerCase() throws Exception {
-    Map<String, String> env = ImmutableMap.<String, String>builder()
-        .put("https_proxy", "https://my.example.com").build();
-    Proxy proxy = ProxyHelper.createProxyIfNeeded("https://www.something.com", env);
+    ProxyHelper helper = new ProxyHelper(ImmutableMap.of("https_proxy", "https://my.example.com"));
+    Proxy proxy = helper.createProxyIfNeeded(new URL("https://www.something.com"));
     assertThat(proxy.toString()).endsWith("my.example.com:443");
   }
 
   @Test
   public void testCreateIfNeededHttpsUpperCase() throws Exception {
-    Map<String, String> env = ImmutableMap.<String, String>builder()
-        .put("HTTPS_PROXY", "https://my.example.com").build();
-    Proxy proxy = ProxyHelper.createProxyIfNeeded("https://www.something.com", env);
+    ProxyHelper helper = new ProxyHelper(ImmutableMap.of("HTTPS_PROXY", "https://my.example.com"));
+    Proxy proxy = helper.createProxyIfNeeded(new URL("https://www.something.com"));
     assertThat(proxy.toString()).endsWith("my.example.com:443");
   }
 
@@ -72,7 +69,8 @@ public class ProxyHelperTest {
     proxy = ProxyHelper.createProxy("");
     assertEquals(Proxy.NO_PROXY, proxy);
     Map<String, String> env = ImmutableMap.of();
-    proxy = ProxyHelper.createProxyIfNeeded("https://www.something.com", env);
+    ProxyHelper helper = new ProxyHelper(env);
+    proxy = helper.createProxyIfNeeded(new URL("https://www.something.com"));
     assertEquals(Proxy.NO_PROXY, proxy);
   }
 
diff --git a/src/test/java/com/google/devtools/build/lib/bazel/repository/downloader/RetryingInputStreamTest.java b/src/test/java/com/google/devtools/build/lib/bazel/repository/downloader/RetryingInputStreamTest.java
new file mode 100644
index 0000000000..10f0923dcb
--- /dev/null
+++ b/src/test/java/com/google/devtools/build/lib/bazel/repository/downloader/RetryingInputStreamTest.java
@@ -0,0 +1,173 @@
+// Copyright 2016 The Bazel Authors. All rights reserved.
+//
+// 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.devtools.build.lib.bazel.repository.downloader;
+
+import static com.google.common.truth.Truth.assertThat;
+import static org.junit.Assert.fail;
+import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.eq;
+import static org.mockito.Matchers.same;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoMoreInteractions;
+import static org.mockito.Mockito.when;
+
+import com.google.common.collect.ImmutableMap;
+import com.google.devtools.build.lib.bazel.repository.downloader.RetryingInputStream.Reconnector;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InterruptedIOException;
+import java.net.SocketTimeoutException;
+import java.net.URLConnection;
+import org.junit.After;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+/** Unit tests for {@link RetryingInputStream}. */
+@RunWith(JUnit4.class)
+public class RetryingInputStreamTest {
+
+  private final InputStream delegate = mock(InputStream.class);
+  private final InputStream newDelegate = mock(InputStream.class);
+  private final Reconnector reconnector = mock(Reconnector.class);
+  private final URLConnection connection = mock(URLConnection.class);
+  private final RetryingInputStream stream = new RetryingInputStream(delegate, reconnector);
+
+  @After
+  public void after() throws Exception {
+    verifyNoMoreInteractions(delegate, newDelegate, reconnector);
+  }
+
+  @Test
+  public void close_callsDelegate() throws Exception {
+    stream.close();
+    verify(delegate).close();
+  }
+
+  @Test
+  public void available_callsDelegate() throws Exception {
+    stream.available();
+    verify(delegate).available();
+  }
+
+  @Test
+  public void read_callsdelegate() throws Exception {
+    stream.read();
+    verify(delegate).read();
+  }
+
+  @Test
+  public void bufferRead_callsdelegate() throws Exception {
+    byte[] buffer = new byte[1024];
+    stream.read(buffer);
+    verify(delegate).read(same(buffer), eq(0), eq(1024));
+  }
+
+  @Test
+  public void readThrowsExceptionWhenDisabled_passesThrough() throws Exception {
+    stream.disabled = true;
+    when(delegate.read()).thenThrow(new IOException());
+    try {
+      stream.read();
+      fail("Expected IOException");
+    } catch (IOException expected) {
+      verify(delegate).read();
+    }
+  }
+
+  @Test
+  public void readInterrupted_alwaysPassesThrough() throws Exception {
+    when(delegate.read()).thenThrow(new InterruptedIOException());
+    try {
+      stream.read();
+      fail("Expected InterruptedIOException");
+    } catch (InterruptedIOException expected) {
+      verify(delegate).read();
+    }
+  }
+
+  @Test
+  @SuppressWarnings("unchecked")
+  public void readTimesOut_retries() throws Exception {
+    when(delegate.read()).thenReturn(1).thenThrow(new SocketTimeoutException());
+    when(reconnector.connect(any(Throwable.class), any(ImmutableMap.class))).thenReturn(connection);
+    when(connection.getInputStream()).thenReturn(newDelegate);
+    when(newDelegate.read()).thenReturn(2);
+    when(connection.getHeaderField("Content-Range")).thenReturn("bytes 1-42/42");
+    assertThat(stream.read()).isEqualTo(1);
+    assertThat(stream.read()).isEqualTo(2);
+    verify(reconnector).connect(any(Throwable.class), eq(ImmutableMap.of("Range", "bytes 1-")));
+    verify(delegate, times(2)).read();
+    verify(delegate).close();
+    verify(newDelegate).read();
+  }
+
+  @Test
+  @SuppressWarnings("unchecked")
+  public void failureWhenNoBytesAreRead_doesntUseRange() throws Exception {
+    when(delegate.read()).thenThrow(new SocketTimeoutException());
+    when(newDelegate.read()).thenReturn(1);
+    when(reconnector.connect(any(Throwable.class), any(ImmutableMap.class))).thenReturn(connection);
+    when(connection.getInputStream()).thenReturn(newDelegate);
+    assertThat(stream.read()).isEqualTo(1);
+    verify(reconnector).connect(any(Throwable.class), eq(ImmutableMap.<String, String>of()));
+    verify(delegate).read();
+    verify(delegate).close();
+    verify(newDelegate).read();
+  }
+
+  @Test
+  @SuppressWarnings("unchecked")
+  public void reconnectFails_alwaysPassesThrough() throws Exception {
+    when(delegate.read()).thenThrow(new IOException());
+    when(reconnector.connect(any(Throwable.class), any(ImmutableMap.class)))
+        .thenThrow(new IOException());
+    try {
+      stream.read();
+      fail("Expected IOException");
+    } catch (IOException expected) {
+      verify(delegate).read();
+      verify(delegate).close();
+      verify(reconnector).connect(any(Throwable.class), any(ImmutableMap.class));
+    }
+  }
+
+  @Test
+  @SuppressWarnings("unchecked")
+  public void maxRetries_givesUp() throws Exception {
+    when(delegate.read())
+        .thenReturn(1)
+        .thenThrow(new IOException())
+        .thenThrow(new IOException())
+        .thenThrow(new IOException())
+        .thenThrow(new SocketTimeoutException());
+    when(reconnector.connect(any(Throwable.class), any(ImmutableMap.class))).thenReturn(connection);
+    when(connection.getInputStream()).thenReturn(delegate);
+    when(connection.getHeaderField("Content-Range")).thenReturn("bytes 1-42/42");
+    stream.read();
+    try {
+      stream.read();
+      fail("Expected SocketTimeoutException");
+    } catch (SocketTimeoutException e) {
+      assertThat(e.getSuppressed()).hasLength(3);
+      verify(reconnector, times(3))
+          .connect(any(Throwable.class), eq(ImmutableMap.of("Range", "bytes 1-")));
+      verify(delegate, times(5)).read();
+      verify(delegate, times(3)).close();
+    }
+  }
+}
diff --git a/src/test/java/com/google/devtools/build/lib/testutil/ManualClock.java b/src/test/java/com/google/devtools/build/lib/testutil/ManualClock.java
index 15752cc2ef..fe500d0464 100644
--- a/src/test/java/com/google/devtools/build/lib/testutil/ManualClock.java
+++ b/src/test/java/com/google/devtools/build/lib/testutil/ManualClock.java
@@ -15,18 +15,18 @@
 package com.google.devtools.build.lib.testutil;
 
 import com.google.devtools.build.lib.util.Clock;
-
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
 
 /**
  * A fake clock for testing.
  */
 public final class ManualClock implements Clock {
-  private long currentTimeMillis = 0L;
+  private final AtomicLong currentTimeMillis = new AtomicLong();
 
   @Override
   public long currentTimeMillis() {
-    return currentTimeMillis;
+    return currentTimeMillis.get();
   }
 
   /**
@@ -36,11 +36,11 @@ public final class ManualClock implements Clock {
    */
   @Override
   public long nanoTime() {
-    return TimeUnit.MILLISECONDS.toNanos(currentTimeMillis)
+    return TimeUnit.MILLISECONDS.toNanos(currentTimeMillis.get())
         + TimeUnit.SECONDS.toNanos(1000);
   }
 
-  public void advanceMillis(long time) {
-    currentTimeMillis += time;
+  public long advanceMillis(long time) {
+    return currentTimeMillis.addAndGet(time);
   }
 }
diff --git a/src/test/java/com/google/devtools/build/lib/testutil/ManualSleeper.java b/src/test/java/com/google/devtools/build/lib/testutil/ManualSleeper.java
new file mode 100644
index 0000000000..6e16253499
--- /dev/null
+++ b/src/test/java/com/google/devtools/build/lib/testutil/ManualSleeper.java
@@ -0,0 +1,36 @@
+// Copyright 2014 The Bazel Authors. All rights reserved.
+//
+// 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.devtools.build.lib.testutil;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import static com.google.common.base.Preconditions.checkNotNull;
+
+import com.google.devtools.build.lib.util.Sleeper;
+
+/** Fake sleeper for testing. */
+public final class ManualSleeper implements Sleeper {
+
+  private final ManualClock clock;
+
+  public ManualSleeper(ManualClock clock) {
+    this.clock = checkNotNull(clock);
+  }
+
+  @Override
+  public void sleepMillis(long milliseconds) throws InterruptedException {
+    checkArgument(milliseconds >= 0, "sleeper can't time travel");
+    clock.advanceMillis(milliseconds);
+  }
+}