From 29e33d6e69c9bf88b36f7d529a5d43fe5816ce93 Mon Sep 17 00:00:00 2001 From: Vladimir Dostoyevskiy <99822064+VDostoyevskiy@users.noreply.github.com> Date: Sat, 2 Dec 2023 15:50:47 +0800 Subject: [PATCH] Initial CronetEngine (#421) * Initial CronetEngine without UploadDataProvider support * Support ByteArrayContent * Cleanup --------- Co-authored-by: GG Bond --- .../com/hippo/ehviewer/cronet/CronetKt.kt | 20 --- .../com/hippo/ehviewer/ktor/CronetEngine.kt | 159 +++++++++++++++++ .../com/hippo/ehviewer/EhApplication.kt | 19 +- .../hippo/ehviewer/cronet/CronetExtensions.kt | 13 -- .../com/hippo/ehviewer/cronet/CronetKt.kt | 20 --- .../com/hippo/ehviewer/ktor/CronetEngine.kt | 165 ++++++++++++++++++ 6 files changed, 338 insertions(+), 58 deletions(-) create mode 100644 app/src/gms/kotlin/com/hippo/ehviewer/ktor/CronetEngine.kt create mode 100644 app/src/oss/kotlin/com/hippo/ehviewer/ktor/CronetEngine.kt diff --git a/app/src/gms/kotlin/com/hippo/ehviewer/cronet/CronetKt.kt b/app/src/gms/kotlin/com/hippo/ehviewer/cronet/CronetKt.kt index 813361f1e..ca71d75c0 100644 --- a/app/src/gms/kotlin/com/hippo/ehviewer/cronet/CronetKt.kt +++ b/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 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> = allHeaders diff --git a/app/src/gms/kotlin/com/hippo/ehviewer/ktor/CronetEngine.kt b/app/src/gms/kotlin/com/hippo/ehviewer/ktor/CronetEngine.kt new file mode 100644 index 000000000..4e29fa683 --- /dev/null +++ b/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 + 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") +} diff --git a/app/src/main/kotlin/com/hippo/ehviewer/EhApplication.kt b/app/src/main/kotlin/com/hippo/ehviewer/EhApplication.kt index 970206670..7799a22ae 100644 --- a/app/src/main/kotlin/com/hippo/ehviewer/EhApplication.kt +++ b/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 + } } } } diff --git a/app/src/main/kotlin/com/hippo/ehviewer/cronet/CronetExtensions.kt b/app/src/main/kotlin/com/hippo/ehviewer/cronet/CronetExtensions.kt index 402232910..39e66ad86 100644 --- a/app/src/main/kotlin/com/hippo/ehviewer/cronet/CronetExtensions.kt +++ b/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) diff --git a/app/src/oss/kotlin/com/hippo/ehviewer/cronet/CronetKt.kt b/app/src/oss/kotlin/com/hippo/ehviewer/cronet/CronetKt.kt index 5ab2befaa..664ee0e0b 100644 --- a/app/src/oss/kotlin/com/hippo/ehviewer/cronet/CronetKt.kt +++ b/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 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> = headers.asMap diff --git a/app/src/oss/kotlin/com/hippo/ehviewer/ktor/CronetEngine.kt b/app/src/oss/kotlin/com/hippo/ehviewer/ktor/CronetEngine.kt new file mode 100644 index 000000000..076fdff03 --- /dev/null +++ b/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 + 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") +}