diff --git a/app/src/main/java/io/legado/app/model/CacheBook.kt b/app/src/main/java/io/legado/app/model/CacheBook.kt index c5f8f9407..e5ee4f337 100644 --- a/app/src/main/java/io/legado/app/model/CacheBook.kt +++ b/app/src/main/java/io/legado/app/model/CacheBook.kt @@ -11,18 +11,29 @@ import io.legado.app.data.entities.BookSource import io.legado.app.exception.ConcurrentException import io.legado.app.help.book.BookHelp import io.legado.app.help.book.isLocal +import io.legado.app.help.config.AppConfig import io.legado.app.help.coroutine.CompositeCoroutine import io.legado.app.help.coroutine.Coroutine import io.legado.app.model.webBook.WebBook import io.legado.app.service.CacheBookService +import io.legado.app.utils.onEachParallel import io.legado.app.utils.postEvent import io.legado.app.utils.startService import kotlinx.coroutines.CancellationException import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.CoroutineStart import kotlinx.coroutines.Dispatchers.IO +import kotlinx.coroutines.coroutineScope +import kotlinx.coroutines.currentCoroutineContext import kotlinx.coroutines.delay +import kotlinx.coroutines.flow.MutableStateFlow +import kotlinx.coroutines.flow.collect +import kotlinx.coroutines.flow.first +import kotlinx.coroutines.flow.flow +import kotlinx.coroutines.isActive +import kotlinx.coroutines.sync.Mutex import kotlinx.coroutines.sync.Semaphore +import kotlinx.coroutines.sync.withLock import java.util.concurrent.ConcurrentHashMap import kotlin.coroutines.CoroutineContext @@ -30,6 +41,9 @@ object CacheBook { val cacheBookMap = ConcurrentHashMap() + private val workingState = MutableStateFlow(true) + private val mutex = Mutex() + @Synchronized fun getOrCreate(bookUrl: String): CacheBookModel? { val book = appDb.bookDao.getBook(bookUrl) ?: return null @@ -104,6 +118,36 @@ object CacheBook { errorDownloadMap.clear() } + fun setWorkingState(value: Boolean) { + workingState.value = value + } + + suspend fun startProcessJob(context: CoroutineContext) = mutex.withLock { + setWorkingState(true) + flow { + while (currentCoroutineContext().isActive && cacheBookMap.isNotEmpty()) { + var emitted = false + + cacheBookMap.forEach { (_, model) -> + if (!model.isLoading()) { + emit(model) + emitted = true + } + workingState.first { it } + } + + if (!emitted) { + delay(1000) + } + } + }.onEachParallel(AppConfig.threadCount) { + coroutineScope { + it.download(this, context) + } + }.collect() + } + + val downloadSummary: String get() { return "正在下载:${onDownloadCount}|等待中:${waitCount}|失败:${errorDownloadMap.count()}|成功:${successDownloadSet.size}" @@ -111,11 +155,12 @@ object CacheBook { val isRun: Boolean get() { - var isRun = false cacheBookMap.forEach { - isRun = isRun || it.value.isRun() + if (it.value.isRun()) { + return true + } } - return isRun + return false } private val waitCount: Int @@ -146,6 +191,7 @@ object CacheBook { private val tasks = CompositeCoroutine() private var isStopped = false private var waitingRetry = false + private var isLoading = true val waitCount get() = waitDownloadSet.size val onDownloadCount get() = onDownloadSet.size @@ -156,7 +202,7 @@ object CacheBook { @Synchronized fun isRun(): Boolean { - return waitDownloadSet.isNotEmpty() || onDownloadSet.isNotEmpty() + return waitDownloadSet.isNotEmpty() || onDownloadSet.isNotEmpty() || isLoading } @Synchronized @@ -164,6 +210,11 @@ object CacheBook { return isStopped || (!isRun() && !waitingRetry) } + @Synchronized + fun isLoading(): Boolean { + return isLoading + } + @Synchronized fun stop() { waitDownloadSet.clear() @@ -181,6 +232,7 @@ object CacheBook { } } cacheBookMap[book.bookUrl] = this + isLoading = false postEvent(EventBus.UP_DOWNLOAD, book.bookUrl) } @@ -243,7 +295,7 @@ object CacheBook { postEvent(EventBus.UP_DOWNLOAD, book.bookUrl) val chapterIndex = waitDownloadSet.firstOrNull() if (chapterIndex == null) { - if (onDownloadSet.isEmpty()) { + if (!isLoading && onDownloadSet.isEmpty()) { cacheBookMap.remove(book.bookUrl) } return diff --git a/app/src/main/java/io/legado/app/service/CacheBookService.kt b/app/src/main/java/io/legado/app/service/CacheBookService.kt index f45ae634c..e807ceeea 100644 --- a/app/src/main/java/io/legado/app/service/CacheBookService.kt +++ b/app/src/main/java/io/legado/app/service/CacheBookService.kt @@ -161,23 +161,8 @@ class CacheBookService : BaseService() { private fun download() { downloadJob?.cancel() downloadJob = lifecycleScope.launch(cachePool) { - while (isActive) { - if (!CacheBook.isRun) { - stopSelf() - return@launch - } - CacheBook.cacheBookMap.forEach { - val cacheBookModel = it.value - while (cacheBookModel.waitCount > 0) { - if (CacheBook.onDownloadCount < threadCount) { - cacheBookModel.download(this, cachePool) - } else { - delay(100) - } - } - } - delay(100) - } + CacheBook.startProcessJob(cachePool) + stopSelf() } } diff --git a/app/src/main/java/io/legado/app/ui/main/MainViewModel.kt b/app/src/main/java/io/legado/app/ui/main/MainViewModel.kt index 16d833abd..17540fe17 100644 --- a/app/src/main/java/io/legado/app/ui/main/MainViewModel.kt +++ b/app/src/main/java/io/legado/app/ui/main/MainViewModel.kt @@ -213,28 +213,14 @@ class MainViewModel(application: Application) : BaseViewModel(application) { if (AppConfig.preDownloadNum == 0) return cacheBookJob?.cancel() cacheBookJob = viewModelScope.launch(upTocPool) { - while (isActive) { - if (CacheBookService.isRun || !CacheBook.isRun) { - cacheBookJob?.cancel() - cacheBookJob = null - return@launch + launch { + while (isActive && CacheBook.isRun) { + //有目录更新是不缓存,优先更新目录,现在更多网站限制并发 + CacheBook.setWorkingState(waitUpTocBooks.isEmpty() && onUpTocBooks.isEmpty()) + delay(1000) } - CacheBook.cacheBookMap.forEach { - val cacheBookModel = it.value - while (cacheBookModel.waitCount > 0) { - //有目录更新是不缓存,优先更新目录,现在更多网站限制并发 - if (waitUpTocBooks.isEmpty() - && onUpTocBooks.isEmpty() - && CacheBook.onDownloadCount < threadCount - ) { - cacheBookModel.download(this, upTocPool) - } else { - delay(100) - } - } - } - delay(100) } + CacheBook.startProcessJob(upTocPool) } } diff --git a/app/src/main/java/io/legado/app/utils/FlowExtensions.kt b/app/src/main/java/io/legado/app/utils/FlowExtensions.kt index 494ff0a5e..40a7d71db 100644 --- a/app/src/main/java/io/legado/app/utils/FlowExtensions.kt +++ b/app/src/main/java/io/legado/app/utils/FlowExtensions.kt @@ -5,6 +5,7 @@ import androidx.lifecycle.repeatOnLifecycle import io.legado.app.data.appDb import kotlinx.coroutines.ExperimentalCoroutinesApi import kotlinx.coroutines.async +import kotlinx.coroutines.currentCoroutineContext import kotlinx.coroutines.ensureActive import kotlinx.coroutines.flow.Flow import kotlinx.coroutines.flow.FlowCollector @@ -20,7 +21,6 @@ import kotlinx.coroutines.flow.map import kotlinx.coroutines.flow.onEach import kotlinx.coroutines.flow.produceIn import kotlinx.coroutines.sync.Semaphore -import kotlin.coroutines.coroutineContext @OptIn(ExperimentalCoroutinesApi::class) inline fun Flow.onEachParallel( @@ -42,7 +42,7 @@ inline fun Flow.onEachParallelSafe( try { action(value) } catch (e: Throwable) { - coroutineContext.ensureActive() + currentCoroutineContext().ensureActive() } emit(value) } @@ -64,7 +64,7 @@ inline fun Flow.mapParallelSafe( try { emit(transform(value)) } catch (_: Throwable) { - coroutineContext.ensureActive() + currentCoroutineContext().ensureActive() } } }.buffer(0) @@ -78,7 +78,7 @@ inline fun Flow.transformParallelSafe( try { transform(value) } catch (e: Throwable) { - coroutineContext.ensureActive() + currentCoroutineContext().ensureActive() } } }.buffer(0)