Improve page loading progress displaying

app/src/main/java/org/koitharu/kotatsu/local/data/PagesCache.kt

@@ -4,10 +4,6 @@ import android.content.Context
 import com.tomclaw.cache.DiskLruCache
 import dagger.hilt.android.qualifiers.ApplicationContext
 import kotlinx.coroutines.Dispatchers
-import kotlinx.coroutines.Job
-import kotlinx.coroutines.currentCoroutineContext
-import kotlinx.coroutines.ensureActive
-import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.withContext
 import org.koitharu.kotatsu.utils.FileSize
 import org.koitharu.kotatsu.utils.ext.copyToSuspending
@@ -43,40 +39,6 @@ class PagesCache @Inject constructor(@ApplicationContext context: Context) {
 			file.delete()
 		}
 	}
-
-	suspend fun put(
-		url: String,
-		inputStream: InputStream,
-		contentLength: Long,
-		progress: MutableStateFlow<Float>,
-	): File = withContext(Dispatchers.IO) {
-		val job = currentCoroutineContext()[Job]
-		val file = File(cacheDir, url.longHashCode().toString())
-		try {
-			file.outputStream().use { out ->
-				var bytesCopied: Long = 0
-				val buffer = ByteArray(DEFAULT_BUFFER_SIZE)
-				var bytes = inputStream.read(buffer)
-				while (bytes >= 0) {
-					out.write(buffer, 0, bytes)
-					bytesCopied += bytes
-					job?.ensureActive()
-					publishProgress(contentLength, bytesCopied, progress)
-					bytes = inputStream.read(buffer)
-					job?.ensureActive()
-				}
-			}
-			lruCache.put(url, file)
-		} finally {
-			file.delete()
-		}
-	}
-
-	private fun publishProgress(contentLength: Long, bytesCopied: Long, progress: MutableStateFlow<Float>) {
-		if (contentLength > 0) {
-			progress.value = (bytesCopied.toDouble() / contentLength.toDouble()).toFloat()
-		}
-	}
 }
 
 private fun createDiskLruCacheSafe(dir: File, size: Long): DiskLruCache {

app/src/main/java/org/koitharu/kotatsu/reader/domain/PageLoader.kt

@@ -33,6 +33,7 @@ import org.koitharu.kotatsu.parsers.util.await
 import org.koitharu.kotatsu.reader.ui.pager.ReaderPage
 import org.koitharu.kotatsu.utils.ext.connectivityManager
 import org.koitharu.kotatsu.utils.ext.printStackTraceDebug
+import org.koitharu.kotatsu.utils.ext.withProgress
 import org.koitharu.kotatsu.utils.progress.ProgressDeferred
 import java.io.File
 import java.util.LinkedList
@@ -203,8 +204,8 @@ class PageLoader @Inject constructor(
 				val body = checkNotNull(response.body) {
 					"Null response"
 				}
-				body.byteStream().use {
-					cache.put(pageUrl, it, body.contentLength(), progress)
+				body.withProgress(progress).byteStream().use {
+					cache.put(pageUrl, it)
 				}
 			}
 		}

app/src/main/java/org/koitharu/kotatsu/reader/ui/pager/PageHolderDelegate.kt

@@ -143,7 +143,7 @@ class PageHolderDelegate(
 	}
 
 	private fun observeProgress(scope: CoroutineScope, progress: Flow<Float>) = progress
-		.debounce(500)
+		.debounce(250)
 		.onEach { callback.onProgressChanged((100 * it).toInt()) }
 		.launchIn(scope)
 

app/src/main/java/org/koitharu/kotatsu/utils/ext/IO.kt

@@ -4,7 +4,10 @@ import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.Job
 import kotlinx.coroutines.currentCoroutineContext
 import kotlinx.coroutines.ensureActive
+import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.withContext
+import okhttp3.ResponseBody
+import org.koitharu.kotatsu.utils.progress.ProgressResponseBody
 import java.io.InputStream
 import java.io.OutputStream
 
@@ -25,3 +28,7 @@ suspend fun InputStream.copyToSuspending(
 	}
 	bytesCopied
 }
+
+fun ResponseBody.withProgress(progressState: MutableStateFlow<Float>): ResponseBody {
+	return ProgressResponseBody(this, progressState)
+}

app/src/main/java/org/koitharu/kotatsu/utils/progress/ProgressResponseBody.kt

@@ -0,0 +1,51 @@
+package org.koitharu.kotatsu.utils.progress
+
+import kotlinx.coroutines.flow.MutableStateFlow
+import okhttp3.MediaType
+import okhttp3.ResponseBody
+import okio.Buffer
+import okio.BufferedSource
+import okio.ForwardingSource
+import okio.Source
+import okio.buffer
+
+class ProgressResponseBody(
+	private val delegate: ResponseBody,
+	private val progressState: MutableStateFlow<Float>,
+) : ResponseBody() {
+
+	private var bufferedSource: BufferedSource? = null
+
+	override fun close() {
+		super.close()
+		delegate.close()
+	}
+
+	override fun contentLength(): Long = delegate.contentLength()
+
+	override fun contentType(): MediaType? = delegate.contentType()
+
+	override fun source(): BufferedSource {
+		return bufferedSource ?: ProgressSource(delegate.source(), contentLength(), progressState).buffer().also {
+			bufferedSource = it
+		}
+	}
+
+	private class ProgressSource(
+		delegate: Source,
+		private val contentLength: Long,
+		private val progressState: MutableStateFlow<Float>,
+	) : ForwardingSource(delegate) {
+
+		private var totalBytesRead = 0L
+
+		override fun read(sink: Buffer, byteCount: Long): Long {
+			val bytesRead = super.read(sink, byteCount)
+			if (contentLength > 0) {
+				totalBytesRead += if (bytesRead != -1L) bytesRead else 0
+				progressState.value = (totalBytesRead.toDouble() / contentLength.toDouble()).toFloat()
+			}
+			return bytesRead
+		}
+	}
+}