Initial CronetEngine (#421)

* Initial CronetEngine without UploadDataProvider support

* Support ByteArrayContent

* Cleanup

---------

Co-authored-by: GG Bond <whatever@tagkong.com>

app/src/gms/kotlin/com/hippo/ehviewer/cronet/CronetKt.kt

@@ -13,13 +13,9 @@ import kotlin.coroutines.resumeWithException
 import kotlinx.coroutines.coroutineScope
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.suspendCancellableCoroutine
-import okhttp3.RequestBody
-import okio.Buffer
 import okio.Path.Companion.toOkioPath
 import org.chromium.net.CronetEngine
 import org.chromium.net.CronetException
-import org.chromium.net.UploadDataProvider
-import org.chromium.net.UploadDataSink
 import org.chromium.net.UrlRequest
 import org.chromium.net.UrlResponseInfo
 import splitties.init.appCtx
@@ -85,21 +81,5 @@ suspend inline fun <R> CronetRequest.execute(crossinline callback: suspend Crone
     }
 }
 
-fun UrlRequest.Builder.withRequestBody(body: RequestBody) {
-    addHeader("Content-Type", body.contentType().toString())
-    val buffer = Buffer().apply { body.writeTo(this) }
-    val provider = object : UploadDataProvider() {
-        override fun getLength() = body.contentLength()
-        override fun read(uploadDataSink: UploadDataSink, byteBuffer: ByteBuffer) {
-            buffer.read(byteBuffer)
-            uploadDataSink.onReadSucceeded(false)
-        }
-        override fun rewind(uploadDataSink: UploadDataSink) {
-            error("OneShot!")
-        }
-    }
-    setUploadDataProvider(provider, cronetHttpClientExecutor)
-}
-
 fun UrlRequest.Builder.noCache(): UrlRequest.Builder = disableCache()
 fun UrlResponseInfo.getHeadersMap(): Map<String, List<String>> = allHeaders

app/src/gms/kotlin/com/hippo/ehviewer/ktor/CronetEngine.kt

@@ -0,0 +1,159 @@
+package com.hippo.ehviewer.ktor
+
+import com.hippo.ehviewer.cronet.cronetHttpClient
+import com.hippo.ehviewer.cronet.cronetHttpClientExecutor
+import com.hippo.ehviewer.cronet.pool
+import io.ktor.client.engine.HttpClientEngineBase
+import io.ktor.client.engine.HttpClientEngineConfig
+import io.ktor.client.engine.callContext
+import io.ktor.client.request.HttpRequestData
+import io.ktor.client.request.HttpResponseData
+import io.ktor.http.HttpHeaders
+import io.ktor.http.HttpProtocolVersion
+import io.ktor.http.HttpStatusCode
+import io.ktor.http.content.OutgoingContent
+import io.ktor.http.headers
+import io.ktor.util.date.GMTDate
+import io.ktor.util.flattenForEach
+import io.ktor.utils.io.ByteReadChannel
+import io.ktor.utils.io.InternalAPI
+import io.ktor.utils.io.pool.useInstance
+import io.ktor.utils.io.writer
+import java.nio.ByteBuffer
+import kotlin.coroutines.Continuation
+import kotlin.coroutines.CoroutineContext
+import kotlin.coroutines.resume
+import kotlin.coroutines.resumeWithException
+import kotlinx.coroutines.DelicateCoroutinesApi
+import kotlinx.coroutines.GlobalScope
+import kotlinx.coroutines.Job
+import kotlinx.coroutines.isActive
+import kotlinx.coroutines.suspendCancellableCoroutine
+import org.chromium.net.CronetException
+import org.chromium.net.UploadDataProvider
+import org.chromium.net.UploadDataSink
+import org.chromium.net.UrlRequest
+import org.chromium.net.UrlResponseInfo
+
+object CronetEngine : HttpClientEngineBase("Cronet") {
+    override val config = HttpClientEngineConfig()
+
+    @InternalAPI
+    override suspend fun execute(data: HttpRequestData): HttpResponseData {
+        val callContext = callContext()
+
+        return executeHttpRequest(callContext, data)
+    }
+
+    @OptIn(DelicateCoroutinesApi::class)
+    private suspend fun executeHttpRequest(
+        callContext: CoroutineContext,
+        data: HttpRequestData,
+    ): HttpResponseData = suspendCancellableCoroutine { continuation ->
+        val requestTime = GMTDate()
+
+        val callback = object : UrlRequest.Callback() {
+            lateinit var readerCont: Continuation<Boolean>
+            override fun onRedirectReceived(request: UrlRequest, info: UrlResponseInfo, newLocationUrl: String) {
+                continuation.resume(
+                    info.toHttpResponseData(
+                        requestTime = requestTime,
+                        callContext = callContext,
+                    ),
+                )
+            }
+
+            override fun onResponseStarted(request: UrlRequest, info: UrlResponseInfo) {
+                val channel = GlobalScope.writer(callContext) {
+                    pool.useInstance { buffer ->
+                        while (callContext.isActive) {
+                            val done = suspendCancellableCoroutine {
+                                readerCont = it
+                                request.read(buffer)
+                            }
+                            buffer.flip()
+                            channel.writeFully(buffer)
+                            buffer.clear()
+                            if (done) break
+                        }
+                    }
+                }.channel
+                continuation.resume(
+                    info.toHttpResponseData(
+                        requestTime = requestTime,
+                        callContext = callContext,
+                        responseBody = channel,
+                    ),
+                )
+            }
+
+            override fun onReadCompleted(request: UrlRequest, info: UrlResponseInfo, byteBuffer: ByteBuffer) {
+                readerCont.resume(false)
+            }
+
+            override fun onSucceeded(request: UrlRequest, info: UrlResponseInfo) {
+                readerCont.resume(true)
+            }
+
+            override fun onFailed(request: UrlRequest, info: UrlResponseInfo, error: CronetException) {
+                if (::readerCont.isInitialized) {
+                    readerCont.resumeWithException(error)
+                } else {
+                    continuation.resumeWithException(error)
+                }
+            }
+        }
+
+        val request = cronetHttpClient.newUrlRequestBuilder(data.url.toString(), callback, cronetHttpClientExecutor).apply {
+            setHttpMethod(data.method.value)
+            data.headers.flattenForEach { key, value -> addHeader(key, value) }
+            data.body.contentType?.let { addHeader(HttpHeaders.ContentType, it.toString()) }
+            data.body.contentLength?.let { addHeader(HttpHeaders.ContentLength, it.toString()) }
+            data.body.toUploadDataProvider()?.let { setUploadDataProvider(it, cronetHttpClientExecutor) }
+        }.build()
+        request.start()
+        callContext[Job]!!.invokeOnCompletion { request.cancel() }
+        continuation.invokeOnCancellation { request.cancel() }
+    }
+}
+
+private fun UrlResponseInfo.toHttpResponseData(
+    requestTime: GMTDate,
+    callContext: CoroutineContext,
+    responseBody: ByteReadChannel = ByteReadChannel.Empty,
+): HttpResponseData {
+    return HttpResponseData(
+        statusCode = HttpStatusCode.fromValue(httpStatusCode),
+        requestTime = requestTime,
+        headers = headers {
+            allHeaders.forEach { (key, value) ->
+                appendAll(key, value)
+            }
+        },
+        version = when (negotiatedProtocol) {
+            "h2" -> HttpProtocolVersion.HTTP_2_0
+            "h3" -> HttpProtocolVersion.QUIC
+            "quic/1+spdy/3" -> HttpProtocolVersion.SPDY_3
+            else -> HttpProtocolVersion.HTTP_1_1
+        },
+        body = responseBody,
+        callContext = callContext,
+    )
+}
+
+private fun OutgoingContent.toUploadDataProvider(): UploadDataProvider? = when (this) {
+    is OutgoingContent.NoContent -> null
+    is OutgoingContent.ByteArrayContent -> object : UploadDataProvider() {
+        val buffer = ByteBuffer.wrap(bytes()).slice()
+        override fun getLength() = buffer.limit().toLong()
+        override fun read(uploadDataSink: UploadDataSink, byteBuffer: ByteBuffer) {
+            byteBuffer.put(buffer)
+            uploadDataSink.onReadSucceeded(false)
+        }
+        override fun rewind(uploadDataSink: UploadDataSink) {
+            buffer.position(0)
+            uploadDataSink.onRewindSucceeded()
+        }
+    }
+    else -> error("UnsupportedContentType $this")
+}

app/src/main/kotlin/com/hippo/ehviewer/EhApplication.kt

@@ -37,6 +37,7 @@ import com.hippo.ehviewer.download.DownloadManager
 import com.hippo.ehviewer.ktbuilder.diskCache
 import com.hippo.ehviewer.ktbuilder.httpClient
 import com.hippo.ehviewer.ktbuilder.imageLoader
+import com.hippo.ehviewer.ktor.CronetEngine
 import com.hippo.ehviewer.legacy.cleanObsoleteCache
 import com.hippo.ehviewer.ui.keepNoMediaFileStatus
 import com.hippo.ehviewer.ui.lockObserver
@@ -177,12 +178,20 @@ class EhApplication : Application(), ImageLoaderFactory {
 
     companion object {
         val ktorClient by lazy {
-            HttpClient(OkHttp) {
-                engine {
-                    preconfigured = baseOkHttpClient
+            if (isCronetSupported) {
+                HttpClient(CronetEngine) {
+                    install(HttpCookies) {
+                        storage = EhCookieStore
+                    }
                 }
-                install(HttpCookies) {
-                    storage = EhCookieStore
+            } else {
+                HttpClient(OkHttp) {
+                    engine {
+                        preconfigured = baseOkHttpClient
+                    }
+                    install(HttpCookies) {
+                        storage = EhCookieStore
+                    }
                 }
             }
         }

app/src/main/kotlin/com/hippo/ehviewer/cronet/CronetExtensions.kt

@@ -31,19 +31,6 @@ suspend inline fun CronetRequest.awaitBodyFully(crossinline callback: (ByteBuffe
     }
 }
 
-@Suppress("NewApi")
-suspend inline fun CronetRequest.awaitBodyFully(buffer: ByteBuffer) {
-    return suspendCancellableCoroutine { cont ->
-        consumer = {
-            request.read(buffer)
-            check(buffer.hasRemaining())
-        }
-        onError = { readerCont.resumeWithException(it) }
-        readerCont = cont
-        request.read(buffer)
-    }
-}
-
 suspend inline fun CronetRequest.copyToChannel(chan: FileChannel, crossinline listener: ((Int) -> Unit) = {}) = awaitBodyFully {
     val bytes = chan.write(it)
     listener(bytes)

app/src/oss/kotlin/com/hippo/ehviewer/cronet/CronetKt.kt

@@ -4,8 +4,6 @@ package com.hippo.ehviewer.cronet
 
 import android.net.http.HttpEngine
 import android.net.http.HttpException
-import android.net.http.UploadDataProvider
-import android.net.http.UploadDataSink
 import android.net.http.UrlRequest
 import android.net.http.UrlResponseInfo
 import com.hippo.ehviewer.client.CHROME_ACCEPT
@@ -21,8 +19,6 @@ import kotlin.coroutines.resumeWithException
 import kotlinx.coroutines.coroutineScope
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.suspendCancellableCoroutine
-import okhttp3.RequestBody
-import okio.Buffer
 import okio.Path.Companion.toOkioPath
 import splitties.init.appCtx
 
@@ -91,21 +87,5 @@ suspend inline fun <R> CronetRequest.execute(crossinline callback: suspend Crone
     }
 }
 
-fun UrlRequest.Builder.withRequestBody(body: RequestBody) {
-    addHeader("Content-Type", body.contentType().toString())
-    val buffer = Buffer().apply { body.writeTo(this) }
-    val provider = object : UploadDataProvider() {
-        override fun getLength() = body.contentLength()
-        override fun read(uploadDataSink: UploadDataSink, byteBuffer: ByteBuffer) {
-            buffer.read(byteBuffer)
-            uploadDataSink.onReadSucceeded(false)
-        }
-        override fun rewind(uploadDataSink: UploadDataSink) {
-            error("OneShot!")
-        }
-    }
-    setUploadDataProvider(provider, cronetHttpClientExecutor)
-}
-
 fun UrlRequest.Builder.noCache(): UrlRequest.Builder = setCacheDisabled(true)
 fun UrlResponseInfo.getHeadersMap(): Map<String, List<String>> = headers.asMap

app/src/oss/kotlin/com/hippo/ehviewer/ktor/CronetEngine.kt

@@ -0,0 +1,165 @@
+@file:Suppress("NewApi")
+
+package com.hippo.ehviewer.ktor
+
+import android.net.http.HttpException
+import android.net.http.UploadDataProvider
+import android.net.http.UploadDataSink
+import android.net.http.UrlRequest
+import android.net.http.UrlResponseInfo
+import com.hippo.ehviewer.cronet.cronetHttpClient
+import com.hippo.ehviewer.cronet.cronetHttpClientExecutor
+import com.hippo.ehviewer.cronet.pool
+import io.ktor.client.engine.HttpClientEngineBase
+import io.ktor.client.engine.HttpClientEngineConfig
+import io.ktor.client.engine.callContext
+import io.ktor.client.request.HttpRequestData
+import io.ktor.client.request.HttpResponseData
+import io.ktor.http.HttpHeaders
+import io.ktor.http.HttpProtocolVersion
+import io.ktor.http.HttpStatusCode
+import io.ktor.http.content.OutgoingContent
+import io.ktor.http.headers
+import io.ktor.util.date.GMTDate
+import io.ktor.util.flattenForEach
+import io.ktor.utils.io.ByteReadChannel
+import io.ktor.utils.io.InternalAPI
+import io.ktor.utils.io.pool.useInstance
+import io.ktor.utils.io.writer
+import java.nio.ByteBuffer
+import kotlin.coroutines.Continuation
+import kotlin.coroutines.CoroutineContext
+import kotlin.coroutines.resume
+import kotlin.coroutines.resumeWithException
+import kotlinx.coroutines.DelicateCoroutinesApi
+import kotlinx.coroutines.GlobalScope
+import kotlinx.coroutines.Job
+import kotlinx.coroutines.isActive
+import kotlinx.coroutines.suspendCancellableCoroutine
+
+object CronetEngine : HttpClientEngineBase("Cronet") {
+    override val config = HttpClientEngineConfig()
+
+    @InternalAPI
+    override suspend fun execute(data: HttpRequestData): HttpResponseData {
+        val callContext = callContext()
+
+        return executeHttpRequest(callContext, data)
+    }
+
+    @OptIn(DelicateCoroutinesApi::class)
+    private suspend fun executeHttpRequest(
+        callContext: CoroutineContext,
+        data: HttpRequestData,
+    ): HttpResponseData = suspendCancellableCoroutine { continuation ->
+        val requestTime = GMTDate()
+
+        val callback = object : UrlRequest.Callback {
+            lateinit var readerCont: Continuation<Boolean>
+            override fun onRedirectReceived(request: UrlRequest, info: UrlResponseInfo, newLocationUrl: String) {
+                continuation.resume(
+                    info.toHttpResponseData(
+                        requestTime = requestTime,
+                        callContext = callContext,
+                    ),
+                )
+            }
+
+            override fun onResponseStarted(request: UrlRequest, info: UrlResponseInfo) {
+                val channel = GlobalScope.writer(callContext) {
+                    pool.useInstance { buffer ->
+                        while (callContext.isActive) {
+                            val done = suspendCancellableCoroutine {
+                                readerCont = it
+                                request.read(buffer)
+                            }
+                            buffer.flip()
+                            channel.writeFully(buffer)
+                            buffer.clear()
+                            if (done) break
+                        }
+                    }
+                }.channel
+                continuation.resume(
+                    info.toHttpResponseData(
+                        requestTime = requestTime,
+                        callContext = callContext,
+                        responseBody = channel,
+                    ),
+                )
+            }
+
+            override fun onReadCompleted(request: UrlRequest, info: UrlResponseInfo, byteBuffer: ByteBuffer) {
+                readerCont.resume(false)
+            }
+
+            override fun onSucceeded(request: UrlRequest, info: UrlResponseInfo) {
+                readerCont.resume(true)
+            }
+
+            override fun onFailed(request: UrlRequest, info: UrlResponseInfo?, error: HttpException) {
+                if (::readerCont.isInitialized) {
+                    readerCont.resumeWithException(error)
+                } else {
+                    continuation.resumeWithException(error)
+                }
+            }
+
+            override fun onCanceled(p0: UrlRequest, p1: UrlResponseInfo?) {
+                // No-op
+            }
+        }
+
+        val request = cronetHttpClient.newUrlRequestBuilder(data.url.toString(), cronetHttpClientExecutor, callback).apply {
+            setHttpMethod(data.method.value)
+            data.headers.flattenForEach { key, value -> addHeader(key, value) }
+            data.body.contentType?.let { addHeader(HttpHeaders.ContentType, it.toString()) }
+            data.body.contentLength?.let { addHeader(HttpHeaders.ContentLength, it.toString()) }
+            data.body.toUploadDataProvider()?.let { setUploadDataProvider(it, cronetHttpClientExecutor) }
+        }.build()
+        request.start()
+        callContext[Job]!!.invokeOnCompletion { request.cancel() }
+        continuation.invokeOnCancellation { request.cancel() }
+    }
+}
+
+private fun UrlResponseInfo.toHttpResponseData(
+    requestTime: GMTDate,
+    callContext: CoroutineContext,
+    responseBody: ByteReadChannel = ByteReadChannel.Empty,
+): HttpResponseData {
+    return HttpResponseData(
+        statusCode = HttpStatusCode.fromValue(httpStatusCode),
+        requestTime = requestTime,
+        headers = headers {
+            headers.asMap.forEach { (key, value) ->
+                appendAll(key, value)
+            }
+        },
+        version = when (negotiatedProtocol) {
+            "h2" -> HttpProtocolVersion.HTTP_2_0
+            "h3" -> HttpProtocolVersion.QUIC
+            "quic/1+spdy/3" -> HttpProtocolVersion.SPDY_3
+            else -> HttpProtocolVersion.HTTP_1_1
+        },
+        body = responseBody,
+        callContext = callContext,
+    )
+}
+
+private fun OutgoingContent.toUploadDataProvider(): UploadDataProvider? = when (this) {
+    is OutgoingContent.NoContent -> null
+    is OutgoingContent.ByteArrayContent -> object : UploadDataProvider() {
+        val buffer = ByteBuffer.wrap(bytes()).slice()
+        override fun getLength() = buffer.limit().toLong()
+        override fun read(uploadDataSink: UploadDataSink, byteBuffer: ByteBuffer) {
+            byteBuffer.put(buffer)
+            uploadDataSink.onReadSucceeded(false)
+        }
+        override fun rewind(uploadDataSink: UploadDataSink) {
+            buffer.position(0)
+            uploadDataSink.onRewindSucceeded()
+        }
+    }
+    else -> error("UnsupportedContentType $this")
+}