diff --git a/app/src/gms/kotlin/com/hippo/ehviewer/ktor/CronetEngine.kt b/app/src/gms/kotlin/com/hippo/ehviewer/ktor/CronetEngine.kt index e3792962f..465c0f615 100644 --- a/app/src/gms/kotlin/com/hippo/ehviewer/ktor/CronetEngine.kt +++ b/app/src/gms/kotlin/com/hippo/ehviewer/ktor/CronetEngine.kt @@ -29,6 +29,7 @@ import kotlinx.coroutines.GlobalScope import kotlinx.coroutines.Job import kotlinx.coroutines.channels.Channel import kotlinx.coroutines.channels.consumeEach +import kotlinx.coroutines.launch import kotlinx.coroutines.suspendCancellableCoroutine import org.chromium.net.CronetException import org.chromium.net.UploadDataProvider @@ -109,7 +110,7 @@ object CronetEngine : HttpClientEngineBase("Cronet") { 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) } + data.body.toUploadDataProvider(callContext)?.let { setUploadDataProvider(it, cronetHttpClientExecutor) } }.build() request.start() callContext[Job]!!.invokeOnCompletion { request.cancel() } @@ -141,7 +142,8 @@ private fun UrlResponseInfo.toHttpResponseData( ) } -private fun OutgoingContent.toUploadDataProvider(): UploadDataProvider? = when (this) { +@OptIn(DelicateCoroutinesApi::class) +private fun OutgoingContent.toUploadDataProvider(callContext: CoroutineContext): UploadDataProvider? = when (this) { is OutgoingContent.NoContent -> null is OutgoingContent.ByteArrayContent -> object : UploadDataProvider() { val buffer = ByteBuffer.wrap(bytes()).slice() @@ -155,5 +157,20 @@ private fun OutgoingContent.toUploadDataProvider(): UploadDataProvider? = when ( uploadDataSink.onRewindSucceeded() } } + is OutgoingContent.WriteChannelContent -> object : UploadDataProvider() { + fun launch() = GlobalScope.writer(callContext) { writeTo(channel) }.channel + var chan = launch() + override fun getLength() = contentLength!! + override fun read(uploadDataSink: UploadDataSink, byteBuffer: ByteBuffer) { + GlobalScope.launch(callContext) { + chan.readAvailable(byteBuffer) + uploadDataSink.onReadSucceeded(false) + } + } + override fun rewind(uploadDataSink: UploadDataSink) { + chan = launch() + uploadDataSink.onRewindSucceeded() + } + } else -> error("UnsupportedContentType $this") } diff --git a/app/src/oss/kotlin/com/hippo/ehviewer/ktor/CronetEngine.kt b/app/src/oss/kotlin/com/hippo/ehviewer/ktor/CronetEngine.kt index 95f267c59..df74f1c7a 100644 --- a/app/src/oss/kotlin/com/hippo/ehviewer/ktor/CronetEngine.kt +++ b/app/src/oss/kotlin/com/hippo/ehviewer/ktor/CronetEngine.kt @@ -36,6 +36,7 @@ import kotlinx.coroutines.GlobalScope import kotlinx.coroutines.Job import kotlinx.coroutines.channels.Channel import kotlinx.coroutines.channels.consumeEach +import kotlinx.coroutines.launch import kotlinx.coroutines.suspendCancellableCoroutine object CronetEngine : HttpClientEngineBase("Cronet") { @@ -115,7 +116,7 @@ object CronetEngine : HttpClientEngineBase("Cronet") { 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) } + data.body.toUploadDataProvider(callContext)?.let { setUploadDataProvider(it, cronetHttpClientExecutor) } }.build() request.start() callContext[Job]!!.invokeOnCompletion { request.cancel() } @@ -147,7 +148,8 @@ private fun UrlResponseInfo.toHttpResponseData( ) } -private fun OutgoingContent.toUploadDataProvider(): UploadDataProvider? = when (this) { +@OptIn(DelicateCoroutinesApi::class) +private fun OutgoingContent.toUploadDataProvider(callContext: CoroutineContext): UploadDataProvider? = when (this) { is OutgoingContent.NoContent -> null is OutgoingContent.ByteArrayContent -> object : UploadDataProvider() { val buffer = ByteBuffer.wrap(bytes()).slice() @@ -161,5 +163,20 @@ private fun OutgoingContent.toUploadDataProvider(): UploadDataProvider? = when ( uploadDataSink.onRewindSucceeded() } } + is OutgoingContent.WriteChannelContent -> object : UploadDataProvider() { + fun launch() = GlobalScope.writer(callContext) { writeTo(channel) }.channel + var chan = launch() + override fun getLength() = contentLength!! + override fun read(uploadDataSink: UploadDataSink, byteBuffer: ByteBuffer) { + GlobalScope.launch(callContext) { + chan.readAvailable(byteBuffer) + uploadDataSink.onReadSucceeded(false) + } + } + override fun rewind(uploadDataSink: UploadDataSink) { + chan = launch() + uploadDataSink.onRewindSucceeded() + } + } else -> error("UnsupportedContentType $this") }