fix android http cancelation

valdi/src/java/com/snap/valdi/network/DefaultHTTPRequestManager.kt

@@ -1,17 +1,60 @@
 package com.snap.valdi.network
 
 import android.content.Context
-import com.snap.valdi.utils.ExecutorsUtil
+import java.io.IOException
 import java.io.InputStream
 import java.net.HttpURLConnection
 import java.net.URL
-import java.util.concurrent.Executors
+import java.net.URLConnection
+import java.util.concurrent.ExecutorService
+import java.util.concurrent.LinkedBlockingQueue
+import java.util.concurrent.ThreadPoolExecutor
+import java.util.concurrent.TimeUnit
+import java.util.concurrent.atomic.AtomicInteger
 import com.snapchat.client.valdi.*
 import com.snapchat.client.valdi_core.*
 
-class DefaultHTTPRequestManager(context: Context): HTTPRequestManager() {
+class DefaultHTTPRequestManager(
+    context: Context,
+    private val openConnection: (URL) -> URLConnection = { it.openConnection() },
+): HTTPRequestManager() {
 
-    private class RequestTask(val url: URL, val method: String, val body: ByteArray?, val headers: Map<String, String>, completion: HTTPRequestManagerCompletion): HTTPRequestTask(completion), Runnable {
+    private class RequestTask(val url: URL, val method: String, val body: ByteArray?, val headers: Map<String, String>, val openConnection: (URL) -> URLConnection, completion: HTTPRequestManagerCompletion): HTTPRequestTask(completion), Runnable {
+
+        private var connection: HttpURLConnection? = null
+        private var cancelled = false
+
+        override fun cancel() {
+            super.cancel()
+
+            val connectionToClose = synchronized(this) {
+                cancelled = true
+                connection
+            }
+
+            // Closing from another thread makes the worker's blocked read throw, which is
+            // how the thread gets reclaimed. Do it outside the lock so a slow close cannot stall
+            // the task binding its connection. Swallow like the worker's own teardown does: this
+            // races that teardown, and native is the only caller left to hand a throw to.
+            try {
+                connectionToClose?.disconnect()
+            } catch (exc: Exception) {}
+        }
+
+        private fun isCancelled(): Boolean = synchronized(this) { cancelled }
+
+        /**
+         * Hands the connection to [cancel] so it can be torn down. Returns false when the request
+         * was already cancelled, in which case the caller must not go on to perform it.
+         */
+        private fun bindConnection(urlConnection: HttpURLConnection): Boolean = synchronized(this) {
+            if (cancelled) {
+                false
+            } else {
+                connection = urlConnection
+                true
+            }
+        }
 
         private fun doPerformRequestWithURLConnection(urlConnection: HttpURLConnection): HTTPResponse {
             urlConnection.instanceFollowRedirects = true
@@ -25,6 +68,14 @@ class DefaultHTTPRequestManager(context: Context): HTTPRequestManager() {
 
                 urlConnection.doInput = true
 
+                // Nothing above this point touches the network. Writing the body connects, and so
+                // does reading responseCode when there is no body, and disconnect() cannot tear
+                // down a connection that has not connected yet -- so this is the last point a
+                // cancel that arrived during setup can be honoured.
+                if (isCancelled()) {
+                    throw IOException("Request was cancelled")
+                }
+
                 if (body != null) {
                     urlConnection.doOutput = true
                     urlConnection.outputStream.write(body)
@@ -53,9 +104,14 @@ class DefaultHTTPRequestManager(context: Context): HTTPRequestManager() {
         }
 
         private fun performRequest(): HTTPResponse {
-            val urlConnection = url.openConnection()
+            val urlConnection = openConnection(url)
 
             if (urlConnection is HttpURLConnection) {
+                if (!bindConnection(urlConnection)) {
+                    urlConnection.disconnect()
+                    throw IOException("Request was cancelled")
+                }
+
                 return doPerformRequestWithURLConnection(urlConnection)
             } else {
                 urlConnection.doInput = true
@@ -66,17 +122,24 @@ class DefaultHTTPRequestManager(context: Context): HTTPRequestManager() {
         }
 
         override fun run() {
+            // Cancelled while queued: opening the connection at all would be wasted work.
+            if (isCancelled()) {
+                return
+            }
+
             try {
                 val response = performRequest()
                 notifySuccess(response)
             } catch (error: Exception) {
+                // Once cancelled the completion is already gone, so a teardown IOException lands
+                // here and goes nowhere -- matching how iOS swallows NSURLErrorCancelled.
                 notifyFailure("HTTP Request failed: ${error.message}")
             }
         }
 
         companion object {
 
-            fun from(request: HTTPRequest, completion: HTTPRequestManagerCompletion): RequestTask {
+            fun from(request: HTTPRequest, openConnection: (URL) -> URLConnection, completion: HTTPRequestManagerCompletion): RequestTask {
                 val url = URL(request.url)
                 val method = request.method
                 val body = request.body
@@ -93,22 +156,35 @@ class DefaultHTTPRequestManager(context: Context): HTTPRequestManager() {
                     }
                 }
 
-                return RequestTask(url, method, body, headers, completion)
+                return RequestTask(url, method, body, headers, openConnection, completion)
             }
         }
     }
 
-    private var executors = ExecutorsUtil.newSingleThreadCachedExecutor { r ->
+    private val threadCount = AtomicInteger(0)
+
+    private val executors: ExecutorService = ThreadPoolExecutor(
+        MAX_CONCURRENT_REQUESTS,
+        MAX_CONCURRENT_REQUESTS,
+        60L,
+        TimeUnit.SECONDS,
+        LinkedBlockingQueue(),
+    ) { r ->
         Thread(r).apply {
-            name = "Valdi Network Thread"
+            name = "Valdi Network Thread ${threadCount.incrementAndGet()}"
             priority = Thread.NORM_PRIORITY
         }
+    }.apply {
+        // An unbounded queue never rejects, so a pool only ever grows to its core size and
+        // maximumPoolSize is inert. Concurrency has to come from the core size, and this restores
+        // the idle reaping that the previous core size of zero provided.
+        allowCoreThreadTimeOut(true)
     }
 
     override fun performRequest(request: HTTPRequest, completion: HTTPRequestManagerCompletion): Cancelable {
         val task: RequestTask
         try {
-            task = RequestTask.from(request, completion)
+            task = RequestTask.from(request, openConnection, completion)
         } catch (exception: Exception) {
             completion.onFail("Failed to build request: ${exception.message}")
 
@@ -123,4 +199,8 @@ class DefaultHTTPRequestManager(context: Context): HTTPRequestManager() {
         return task
     }
 
+    companion object {
+        const val MAX_CONCURRENT_REQUESTS = 4
+    }
+
 }