Workaround OkHttp lack of clean interrupt support

libraries/datasource_okhttp/build.gradle

@@ -11,9 +11,15 @@
 // 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.
+
 apply from: "$gradle.ext.androidxMediaSettingsDir/common_library_config.gradle"
+apply plugin: 'kotlin-android'
 
-android.defaultConfig.minSdkVersion 21
+android {
+    defaultConfig {
+        minSdkVersion 21
+    }
+}
 
 dependencies {
     implementation project(modulePrefix + 'lib-common')
@@ -25,6 +31,14 @@ dependencies {
     testImplementation 'com.squareup.okhttp3:mockwebserver:' + okhttpVersion
     testImplementation 'org.robolectric:robolectric:' + robolectricVersion
     api 'com.squareup.okhttp3:okhttp:' + okhttpVersion
+
+    androidTestImplementation 'androidx.test:rules:' + androidxTestRulesVersion
+    androidTestImplementation 'androidx.test:runner:' + androidxTestRunnerVersion
+    androidTestImplementation 'androidx.multidex:multidex:' + androidxMultidexVersion
+    androidTestImplementation 'com.linkedin.dexmaker:dexmaker-mockito:' + dexmakerVersion
+    androidTestImplementation project(modulePrefix + 'test-utils')
+    androidTestImplementation project(modulePrefix + 'lib-exoplayer')
+    androidTestImplementation 'androidx.media:media:' + androidxMediaVersion
 }
 
 ext {

libraries/datasource_okhttp/src/androidTest/AndroidManifest.xml

@@ -0,0 +1,25 @@
+<?xml version="1.0" encoding="utf-8"?>
+<!-- Copyright (C) 2021 The Android Open Source Project
+
+     Licensed under the Apache License, Version 2.0 (the "License");
+     you may not use this file except in compliance with the License.
+     You may obtain a copy of the License at
+
+          http://www.apache.org/licenses/LICENSE-2.0
+
+     Unless required by applicable law or agreed to in writing, software
+     distributed under the License is distributed on an "AS IS" BASIS,
+     WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+     See the License for the specific language governing permissions and
+     limitations under the License.
+-->
+
+<manifest xmlns:android="http://schemas.android.com/apk/res/android"
+    package="androidx.media3.datasource.okhttp.test">
+
+  <uses-permission android:name="android.permission.INTERNET"/>
+  <uses-permission android:name="android.permission.ACCESS_NETWORK_STATE"/>
+
+  <application android:usesCleartextTraffic="true" />
+
+</manifest>

libraries/datasource_okhttp/src/androidTest/java/androidx/media3/datasource/okhttp/OkHttpDataSourceCancellationTest.kt

@@ -0,0 +1,145 @@
+/*
+ * Copyright (C) 2020 The Android Open Source Project
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package androidx.media3.datasource.okhttp
+
+import androidx.media3.datasource.DataSource
+import androidx.media3.datasource.DataSpec
+import androidx.media3.datasource.HttpDataSource.HttpDataSourceException
+import androidx.test.ext.junit.runners.AndroidJUnit4
+import com.google.common.truth.Truth.assertThat
+import okhttp3.Call
+import okhttp3.OkHttpClient
+import org.junit.After
+import org.junit.Assert.fail
+import org.junit.Assume.assumeTrue
+import org.junit.Test
+import org.junit.runner.RunWith
+import java.io.Closeable
+import java.net.InetAddress
+import java.net.UnknownHostException
+import java.util.concurrent.CountDownLatch
+import java.util.logging.Handler
+import java.util.logging.LogRecord
+
+/** [DataSource] contract tests for [OkHttpDataSource].  */
+@RunWith(AndroidJUnit4::class)
+class OkHttpDataSourceCancellationTest {
+    val logs: MutableList<String> = mutableListOf()
+
+    val okHttpClient: Call.Factory = OkHttpClient.Builder()
+            .build()
+
+    private var logging: Closeable? = null
+
+    @After
+    fun disableLogging() {
+        logging?.close()
+    }
+
+    fun createDataSource(): DataSource {
+        return OkHttpDataSource.Factory(okHttpClient).createDataSource()
+    }
+
+    @Test
+    fun handlesFastCancellations() {
+        assumeInternet()
+
+        val interruptPoint = CountDownLatch(1)
+        val interruptedPoint = CountDownLatch(1)
+
+        logging = OkHttpDebugLogging.enableHttp2(handler = TestLogHandler { message ->
+            if (message.matches(">>.*HEADERS".toRegex())) {
+                interruptPoint.countDown()
+                interruptedPoint.awaitUninterruptible()
+            }
+        })
+
+        val testThread = Thread.currentThread()
+
+        Thread {
+            interruptPoint.await()
+
+            testThread.interrupt()
+
+            interruptedPoint.countDown()
+        }.start()
+
+        val dataSource = createDataSource()
+        val content = DataSpec.Builder()
+                .setUri("https://storage.googleapis.com/exoplayer-test-media-1/gen-3/screens/dash-vod-single-segment/video-avc-baseline-480.mp4")
+                .build()
+
+        try {
+            dataSource.open(content)
+            fail()
+        } catch (hdse: HttpDataSourceException) {
+            // expected
+        }
+
+//        logs.forEach {
+//            println("'$it'")
+//        }
+
+        // Check we started a request
+        assertThat(logs).contains(">> 0x00000003   102 HEADERS       END_STREAM|END_HEADERS")
+
+        // If execute is used in OkHttpDataSource.open() then this RST_STREAM is not sent.
+        assertThat(logs).contains(">> 0x00000003     4 RST_STREAM    ")
+    }
+
+    private fun assumeInternet() {
+        try {
+            InetAddress.getByName("www.google.com")
+        } catch (uhe: UnknownHostException) {
+            assumeTrue("requires network", false)
+        }
+    }
+
+    inner class TestLogHandler(val onMessage: (message: String) -> Unit) : Handler() {
+        override fun publish(log: LogRecord) {
+            val message = log.message
+
+            logs.add(message)
+
+            onMessage(message)
+        }
+
+        override fun flush() {
+        }
+
+        override fun close() {
+        }
+    }
+}
+
+private fun CountDownLatch.awaitUninterruptible() {
+    var interrupted = false
+
+    try {
+        while (true) {
+            try {
+                await()
+                return
+            } catch (ie: InterruptedException) {
+                interrupted = true
+            }
+        }
+    } finally {
+        if (interrupted) {
+            Thread.currentThread().interrupt()
+        }
+    }
+}

libraries/datasource_okhttp/src/androidTest/java/androidx/media3/datasource/okhttp/OkHttpDataSourceContractTest.kt

@@ -0,0 +1,49 @@
+/*
+ * Copyright (C) 2020 The Android Open Source Project
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package androidx.media3.datasource.okhttp
+
+import android.net.Uri
+import androidx.media3.datasource.DataSource
+import androidx.media3.test.utils.DataSourceContractTest
+import androidx.media3.test.utils.HttpDataSourceTestEnv
+import androidx.test.ext.junit.runners.AndroidJUnit4
+import com.google.common.collect.ImmutableList
+import okhttp3.Call
+import okhttp3.OkHttpClient
+import org.junit.Rule
+import org.junit.runner.RunWith
+
+/** [DataSource] contract tests for [OkHttpDataSource].  */
+@RunWith(AndroidJUnit4::class)
+class OkHttpDataSourceContractTest : DataSourceContractTest() {
+    @get:Rule
+    var httpDataSourceTestEnv = HttpDataSourceTestEnv()
+
+    override fun createDataSource(): DataSource {
+        val okHttpClient: Call.Factory = OkHttpClient.Builder()
+                .build()
+
+        return OkHttpDataSource.Factory(okHttpClient).createDataSource()
+    }
+
+    override fun getTestResources(): ImmutableList<TestResource> {
+        return httpDataSourceTestEnv.servedResources
+    }
+
+    override fun getNotFoundUri(): Uri {
+        return Uri.parse(httpDataSourceTestEnv.nonexistentUrl)
+    }
+}

libraries/datasource_okhttp/src/androidTest/java/androidx/media3/datasource/okhttp/logging.kt

@@ -0,0 +1,59 @@
+/*
+ * Copyright (C) 2019 Square, Inc.
+ *
+ * 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 androidx.media3.datasource.okhttp
+
+import okhttp3.internal.concurrent.TaskRunner
+import okhttp3.internal.http2.Http2
+import java.io.Closeable
+import java.util.concurrent.CopyOnWriteArraySet
+import java.util.logging.ConsoleHandler
+import java.util.logging.Handler
+import java.util.logging.Level
+import java.util.logging.LogRecord
+import java.util.logging.Logger
+import java.util.logging.SimpleFormatter
+import kotlin.reflect.KClass
+
+object OkHttpDebugLogging {
+    // Keep references to loggers to prevent their configuration from being GC'd.
+    private val configuredLoggers = CopyOnWriteArraySet<Logger>()
+
+    fun enableHttp2(handler: Handler = logHandler()) = enable(Http2::class, handler)
+
+    fun enableTaskRunner(handler: Handler = logHandler()) = enable(TaskRunner::class, handler)
+
+    fun logHandler() = ConsoleHandler().apply {
+        level = Level.FINE
+        formatter = object : SimpleFormatter() {
+            override fun format(record: LogRecord) =
+                    String.format("[%1\$tF %1\$tT] %2\$s %n", record.millis, record.message)
+        }
+    }
+
+    fun enable(loggerClass: String, handler: Handler = logHandler()): Closeable {
+        val logger = Logger.getLogger(loggerClass)
+        if (configuredLoggers.add(logger)) {
+            logger.addHandler(handler)
+            logger.level = Level.FINEST
+        }
+        return Closeable {
+            logger.removeHandler(handler)
+        }
+    }
+
+    fun enable(loggerClass: KClass<*>, handler: Handler = logHandler()) =
+            enable(loggerClass.java.name, handler)
+}

libraries/datasource_okhttp/src/main/java/androidx/media3/datasource/okhttp/OkHttpDataSource.java

@@ -36,15 +36,18 @@
 import androidx.media3.datasource.TransferListener;
 import com.google.common.base.Predicate;
 import com.google.common.net.HttpHeaders;
+import com.google.common.util.concurrent.SettableFuture;
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.InterruptedIOException;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.ExecutionException;
 import okhttp3.CacheControl;
 import okhttp3.Call;
+import okhttp3.Callback;
 import okhttp3.HttpUrl;
 import okhttp3.MediaType;
 import okhttp3.OkHttpClient;
@@ -300,7 +303,8 @@ public long open(DataSpec dataSpec) throws HttpDataSourceException {
     Response response;
     ResponseBody responseBody;
     try {
-      this.response = callFactory.newCall(request).execute();
+      this.response = executeAsync(callFactory.newCall(request));
+
       response = this.response;
       responseBody = Assertions.checkNotNull(response.body());
       responseByteStream = responseBody.byteStream();
@@ -374,6 +378,30 @@ public long open(DataSpec dataSpec) throws HttpDataSourceException {
     return bytesToRead;
   }
 
+  private Response executeAsync(Call call) throws IOException {
+    SettableFuture<Response> future = SettableFuture.create();
+    call.enqueue(new Callback() {
+      @Override
+      public void onFailure(Call call, IOException e) {
+        future.setException(e);
+      }
+
+      @Override
+      public void onResponse(Call call, Response response) {
+        future.set(response);
+      }
+    });
+
+    try {
+      return future.get();
+    } catch (InterruptedException e) {
+      call.cancel();
+      throw new InterruptedIOException();
+    } catch (ExecutionException e) {
+      throw (IOException) e.getCause();
+    }
+  }
+
   @UnstableApi
   @Override
   public int read(byte[] buffer, int offset, int length) throws HttpDataSourceException {