Make CronetDataSource interruptable.

-------------
Created by MOE: https://github.com/google/moe
MOE_MIGRATED_REVID=173262660

extensions/cronet/src/androidTest/java/com/google/android/exoplayer2/ext/cronet/CronetDataSourceTest.java

@@ -75,12 +75,13 @@ import org.mockito.stubbing.Answer;
 public final class CronetDataSourceTest {
 
   private static final int TEST_CONNECT_TIMEOUT_MS = 100;
-  private static final int TEST_READ_TIMEOUT_MS = 50;
+  private static final int TEST_READ_TIMEOUT_MS = 100;
   private static final String TEST_URL = "http://google.com";
   private static final String TEST_CONTENT_TYPE = "test/test";
   private static final byte[] TEST_POST_BODY = "test post body".getBytes();
   private static final long TEST_CONTENT_LENGTH = 16000L;
   private static final int TEST_CONNECTION_STATUS = 5;
+  private static final int TEST_INVALID_CONNECTION_STATUS = -1;
 
   private DataSpec testDataSpec;
   private DataSpec testPostDataSpec;
@@ -576,6 +577,45 @@ public final class CronetDataSourceTest {
     verify(mockTransferListener, never()).onTransferStart(dataSourceUnderTest, testDataSpec);
   }
 
+  @Test
+  public void testConnectInterrupted() {
+    when(mockClock.elapsedRealtime()).thenReturn(0L);
+    final ConditionVariable startCondition = buildUrlRequestStartedCondition();
+    final ConditionVariable timedOutCondition = new ConditionVariable();
+
+    Thread thread =
+        new Thread() {
+          @Override
+          public void run() {
+            try {
+              dataSourceUnderTest.open(testDataSpec);
+              fail();
+            } catch (HttpDataSourceException e) {
+              // Expected.
+              assertTrue(e instanceof CronetDataSource.OpenException);
+              assertTrue(e.getCause() instanceof CronetDataSource.InterruptedIOException);
+              assertEquals(
+                  TEST_INVALID_CONNECTION_STATUS,
+                  ((CronetDataSource.OpenException) e).cronetConnectionStatus);
+              timedOutCondition.open();
+            }
+          }
+        };
+    thread.start();
+    startCondition.block();
+
+    // We should still be trying to open.
+    assertFalse(timedOutCondition.block(50));
+    // We should still be trying to open as we approach the timeout.
+    when(mockClock.elapsedRealtime()).thenReturn((long) TEST_CONNECT_TIMEOUT_MS - 1);
+    assertFalse(timedOutCondition.block(50));
+    // Now we interrupt.
+    thread.interrupt();
+    timedOutCondition.block();
+
+    verify(mockTransferListener, never()).onTransferStart(dataSourceUnderTest, testDataSpec);
+  }
+
   @Test
   public void testConnectResponseBeforeTimeout() {
     when(mockClock.elapsedRealtime()).thenReturn(0L);
@@ -717,6 +757,38 @@ public final class CronetDataSourceTest {
     }
   }
 
+  @Test
+  public void testReadInterrupted() throws HttpDataSourceException {
+    when(mockClock.elapsedRealtime()).thenReturn(0L);
+    mockResponseStartSuccess();
+    dataSourceUnderTest.open(testDataSpec);
+
+    final ConditionVariable startCondition = buildReadStartedCondition();
+    final ConditionVariable timedOutCondition = new ConditionVariable();
+    byte[] returnedBuffer = new byte[8];
+    Thread thread =
+        new Thread() {
+          @Override
+          public void run() {
+            try {
+              dataSourceUnderTest.read(returnedBuffer, 0, 8);
+              fail();
+            } catch (HttpDataSourceException e) {
+              // Expected.
+              assertTrue(e.getCause() instanceof CronetDataSource.InterruptedIOException);
+              timedOutCondition.open();
+            }
+          }
+        };
+    thread.start();
+    startCondition.block();
+
+    assertFalse(timedOutCondition.block(50));
+    // Now we interrupt.
+    thread.interrupt();
+    timedOutCondition.block();
+  }
+
   @Test
   public void testAllowDirectExecutor() throws HttpDataSourceException {
     testDataSpec = new DataSpec(Uri.parse(TEST_URL), 1000, 5000, null);
@@ -834,16 +906,34 @@ public final class CronetDataSourceTest {
   }
 
   private void mockReadFailure() {
-    doAnswer(new Answer<Object>() {
+    doAnswer(
-      @Override
+            new Answer<Object>() {
-      public Object answer(InvocationOnMock invocation) throws Throwable {
+              @Override
-        dataSourceUnderTest.onFailed(
+              public Object answer(InvocationOnMock invocation) throws Throwable {
-            mockUrlRequest,
+                dataSourceUnderTest.onFailed(
-            createUrlResponseInfo(500), // statusCode
+                    mockUrlRequest,
-            mockNetworkException);
+                    createUrlResponseInfo(500), // statusCode
-        return null;
+                    mockNetworkException);
-      }
+                return null;
-    }).when(mockUrlRequest).read(any(ByteBuffer.class));
+              }
+            })
+        .when(mockUrlRequest)
+        .read(any(ByteBuffer.class));
+  }
+
+  private ConditionVariable buildReadStartedCondition() {
+    final ConditionVariable startedCondition = new ConditionVariable();
+    doAnswer(
+            new Answer<Object>() {
+              @Override
+              public Object answer(InvocationOnMock invocation) throws Throwable {
+                startedCondition.open();
+                return null;
+              }
+            })
+        .when(mockUrlRequest)
+        .read(any(ByteBuffer.class));
+    return startedCondition;
   }
 
   private ConditionVariable buildUrlRequestStartedCondition() {

extensions/cronet/src/main/java/com/google/android/exoplayer2/ext/cronet/CronetDataSource.java

@@ -16,7 +16,6 @@
 package com.google.android.exoplayer2.ext.cronet;
 
 import android.net.Uri;
-import android.os.ConditionVariable;
 import android.text.TextUtils;
 import android.util.Log;
 import com.google.android.exoplayer2.C;
@@ -27,6 +26,7 @@ import com.google.android.exoplayer2.upstream.HttpDataSource;
 import com.google.android.exoplayer2.upstream.TransferListener;
 import com.google.android.exoplayer2.util.Assertions;
 import com.google.android.exoplayer2.util.Clock;
+import com.google.android.exoplayer2.util.ConditionVariable;
 import com.google.android.exoplayer2.util.Predicate;
 import java.io.IOException;
 import java.net.CookieManager;
@@ -78,6 +78,14 @@ public class CronetDataSource extends UrlRequest.Callback implements HttpDataSou
 
   }
 
+  /** Thrown on catching an InterruptedException. */
+  public static final class InterruptedIOException extends IOException {
+
+    public InterruptedIOException(InterruptedException e) {
+      super(e);
+    }
+  }
+
   static {
     ExoPlayerLibraryInfo.registerModule("goog.exo.cronet");
   }
@@ -266,13 +274,18 @@ public class CronetDataSource extends UrlRequest.Callback implements HttpDataSou
       throw new OpenException(e, dataSpec, Status.IDLE);
     }
     currentUrlRequest.start();
-    boolean requestStarted = blockUntilConnectTimeout();
 
-    if (exception != null) {
+    try {
-      throw new OpenException(exception, currentDataSpec, getStatus(currentUrlRequest));
+      boolean connectionOpened = blockUntilConnectTimeout();
-    } else if (!requestStarted) {
+      if (exception != null) {
-      // The timeout was reached before the connection was opened.
+        throw new OpenException(exception, currentDataSpec, getStatus(currentUrlRequest));
-      throw new OpenException(new SocketTimeoutException(), dataSpec, getStatus(currentUrlRequest));
+      } else if (!connectionOpened) {
+        // The timeout was reached before the connection was opened.
+        throw new OpenException(
+            new SocketTimeoutException(), dataSpec, getStatus(currentUrlRequest));
+      }
+    } catch (InterruptedException e) {
+      throw new OpenException(new InterruptedIOException(e), dataSpec, Status.INVALID);
     }
 
     // Check for a valid response code.
@@ -340,14 +353,24 @@ public class CronetDataSource extends UrlRequest.Callback implements HttpDataSou
       operation.close();
       readBuffer.clear();
       currentUrlRequest.read(readBuffer);
-      if (!operation.block(readTimeoutMs)) {
+      try {
-        // We're timing out, but since the operation is still ongoing we'll need to replace
+        if (!operation.block(readTimeoutMs)) {
-        // readBuffer to avoid the possibility of it being written to by this operation during a
+          throw new SocketTimeoutException();
-        // subsequent request.
+        }
+      } catch (InterruptedException | SocketTimeoutException e) {
+        // If we're timing out or getting interrupted, the operation is still ongoing.
+        // So we'll need to replace readBuffer to avoid the possibility of it being written to by
+        // this operation during a subsequent request.
         readBuffer = null;
         throw new HttpDataSourceException(
-            new SocketTimeoutException(), currentDataSpec, HttpDataSourceException.TYPE_READ);
+            e instanceof InterruptedException
-      } else if (exception != null) {
+                ? new InterruptedIOException((InterruptedException) e)
+                : (SocketTimeoutException) e,
+            currentDataSpec,
+            HttpDataSourceException.TYPE_READ);
+      }
+
+      if (exception != null) {
         throw new HttpDataSourceException(exception, currentDataSpec,
             HttpDataSourceException.TYPE_READ);
       } else if (finished) {
@@ -541,7 +564,7 @@ public class CronetDataSource extends UrlRequest.Callback implements HttpDataSou
     return requestBuilder.build();
   }
 
-  private boolean blockUntilConnectTimeout() {
+  private boolean blockUntilConnectTimeout() throws InterruptedException {
     long now = clock.elapsedRealtime();
     boolean opened = false;
     while (!opened && now < currentConnectTimeoutMs) {
@@ -639,7 +662,7 @@ public class CronetDataSource extends UrlRequest.Callback implements HttpDataSou
     return contentLength;
   }
 
-  private static int getStatus(UrlRequest request) {
+  private static int getStatus(UrlRequest request) throws InterruptedException {
     final ConditionVariable conditionVariable = new ConditionVariable();
     final int[] statusHolder = new int[1];
     request.getStatus(new UrlRequest.StatusListener() {

library/core/src/main/java/com/google/android/exoplayer2/util/ConditionVariable.java

@@ -66,7 +66,7 @@ public final class ConditionVariable {
    * @return true If the condition was opened, false if the call returns because of the timeout.
    * @throws InterruptedException If the thread is interrupted.
    */
-  public synchronized boolean block(int timeout) throws InterruptedException {
+  public synchronized boolean block(long timeout) throws InterruptedException {
     long now = System.currentTimeMillis();
     long end = now + timeout;
     while (!isOpen && now < end) {