[优化] 继续干下载逻辑,现在小米设备应该大概可能不会再有主线程锁等待导致的ANR

This commit is contained in:
HapeLee
2026-04-30 02:08:24 +08:00
parent bc670629b4
commit bab3faedef
17 changed files with 1669 additions and 582 deletions
@@ -11,6 +11,7 @@ import io.legado.app.constant.BookType
import io.legado.app.data.entities.Book
import io.legado.app.data.entities.BookGroup
import io.legado.app.data.entities.BookSource
import io.legado.app.domain.model.CacheableBook
import io.legado.app.help.book.isNotShelf
import io.legado.app.ui.main.bookshelf.BookShelfItem
import kotlinx.coroutines.flow.Flow
@@ -516,6 +517,20 @@ interface BookDao {
@Query("SELECT * FROM books WHERE bookUrl = :bookUrl")
fun getBook(bookUrl: String): Book?
@Query(
"""
SELECT
bookUrl,
type & ${BookType.local} > 0 AS isLocal,
type & ${BookType.audio} > 0 AS isAudio,
durChapterIndex,
totalChapterNum - 1 AS lastChapterIndex
FROM books
WHERE bookUrl IN (:bookUrls)
"""
)
fun getCacheableBooks(bookUrls: Set<String>): List<CacheableBook>
@Query("SELECT * FROM books WHERE bookUrl = :bookUrl")
fun flowGetBook(bookUrl: String): Flow<Book?>
@@ -7,7 +7,6 @@ import io.legado.app.domain.model.BookGroupAssignment
import io.legado.app.domain.model.CacheableBook
import io.legado.app.domain.model.DeletableBook
import io.legado.app.domain.repository.BookDomainRepository
import io.legado.app.help.book.isAudio
import io.legado.app.help.book.isLocal
class BookDomainRepositoryImpl(
@@ -21,15 +20,8 @@ class BookDomainRepositoryImpl(
}
override suspend fun getCacheableBooks(bookUrls: Set<String>): List<CacheableBook> {
return getBooks(bookUrls).map { book ->
CacheableBook(
bookUrl = book.bookUrl,
isLocal = book.isLocal,
isAudio = book.isAudio,
durChapterIndex = book.durChapterIndex,
lastChapterIndex = book.lastChapterIndex
)
}
if (bookUrls.isEmpty()) return emptyList()
return bookDao.getCacheableBooks(bookUrls)
}
override suspend fun getDeletableBooks(bookUrls: Set<String>): List<DeletableBook> {
@@ -17,6 +17,11 @@ class CacheBookDownloadRepository(
CacheBook.start(appCtx, request, isLocal = book.isLocal)
}
override suspend fun start(requests: List<CacheDownloadRequest>) {
if (requests.isEmpty()) return
CacheBook.start(appCtx, requests)
}
override suspend fun start(bookUrl: String, chapterIndices: List<Int>) {
start(
CacheDownloadRequest(
@@ -4,6 +4,7 @@ import io.legado.app.model.cache.CacheDownloadRequest
interface BookCacheDownloadGateway {
suspend fun start(request: CacheDownloadRequest)
suspend fun start(requests: List<CacheDownloadRequest>)
suspend fun start(bookUrl: String, chapterIndices: List<Int>)
suspend fun start(bookUrl: String, startIndex: Int, endIndex: Int)
}
@@ -18,32 +18,27 @@ class BatchCacheDownloadUseCase(
skipAudioBooks: Boolean = false
): Int {
if (bookUrls.isEmpty()) return 0
var count = 0
bookRepository.getCacheableBooks(bookUrls).forEach { book ->
if (startIfNeeded(book, downloadAllChapters, skipAudioBooks)) {
count++
}
val requests = bookRepository.getCacheableBooks(bookUrls).mapNotNull { book ->
createRequestIfNeeded(book, downloadAllChapters, skipAudioBooks)
}
return count
bookCacheDownloadGateway.start(requests)
return requests.size
}
private suspend fun startIfNeeded(
private fun createRequestIfNeeded(
book: CacheableBook,
downloadAllChapters: Boolean,
skipAudioBooks: Boolean
): Boolean {
if (book.isLocal) return false
if (skipAudioBooks && book.isAudio) return false
): CacheDownloadRequest? {
if (book.isLocal) return null
if (skipAudioBooks && book.isAudio) return null
val startIndex = if (downloadAllChapters) 0 else book.durChapterIndex
val endIndex = book.lastChapterIndex
if (endIndex < startIndex) return false
bookCacheDownloadGateway.start(
CacheDownloadRequest(
bookUrl = book.bookUrl,
selection = ChapterSelection.Range(startIndex, endIndex),
source = CacheDownloadSource.Batch,
)
if (endIndex < startIndex) return null
return CacheDownloadRequest(
bookUrl = book.bookUrl,
selection = ChapterSelection.Range(startIndex, endIndex),
source = CacheDownloadSource.Batch,
)
return true
}
}
+344 -430
View File
@@ -1,32 +1,21 @@
package io.legado.app.model
import android.content.Context
import io.legado.app.constant.AppLog
import io.legado.app.constant.IntentAction
import io.legado.app.data.appDb
import io.legado.app.data.entities.Book
import io.legado.app.data.entities.BookChapter
import io.legado.app.data.entities.BookSource
import io.legado.app.exception.ConcurrentException
import io.legado.app.help.book.isLocal
import io.legado.app.help.coroutine.CompositeCoroutine
import io.legado.app.help.coroutine.Coroutine
import io.legado.app.model.cache.CacheDownloadQueue
import io.legado.app.model.cache.CacheDownloadRepository
import io.legado.app.model.cache.CacheDownloadRequest
import io.legado.app.model.cache.CacheDownloadSource
import io.legado.app.model.cache.CacheDownloadStateStore
import io.legado.app.model.cache.ChapterSelection
import io.legado.app.model.cache.ReadingCacheEvent
import io.legado.app.model.cache.ReadingCacheEvents
import io.legado.app.service.CacheBookService
import io.legado.app.ui.config.otherConfig.OtherConfig
import io.legado.app.utils.onEachParallel
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.CompletableDeferred
import kotlinx.coroutines.TimeoutCancellationException
import kotlinx.coroutines.coroutineScope
import kotlinx.coroutines.currentCoroutineContext
import kotlinx.coroutines.delay
@@ -39,11 +28,13 @@ import kotlinx.coroutines.flow.first
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.flow.onCompletion
import kotlinx.coroutines.flow.onStart
import kotlinx.coroutines.flow.update
import kotlinx.coroutines.isActive
import kotlinx.coroutines.sync.Mutex
import kotlinx.coroutines.sync.Semaphore
import kotlinx.coroutines.sync.withLock
import kotlinx.coroutines.withTimeout
import java.util.concurrent.ConcurrentHashMap
import java.util.concurrent.atomic.AtomicInteger
import java.util.concurrent.atomic.AtomicLong
import kotlin.coroutines.CoroutineContext
@@ -51,16 +42,20 @@ object CacheBook {
const val maxDownloadConcurrency = 8
data class Diagnostics(
val activeBookCount: Int,
val waitingChapterCount: Int,
val runningChapterCount: Int,
val trackedChapterTaskCount: Int,
val loadingBookCount: Int,
val retryingBookCount: Int,
)
private data class QueueStats(
val waitingCount: Int,
val downloadingCount: Int
)
private data class ChapterKey(
val bookUrl: String,
val index: Int,
)
private class CacheBookCoordinator {
val taskMap = ConcurrentHashMap<String, CacheBookModel>()
private val processMutex = Mutex()
@@ -73,13 +68,13 @@ object CacheBook {
suspend fun startProcessJob(context: CoroutineContext) = processMutex.withLock {
setWorkingState(true)
flow {
while (currentCoroutineContext().isActive && taskMap.isNotEmpty()) {
while (currentCoroutineContext().isActive && taskMap.isNotEmpty() && !isPaused) {
if (!workingState.value) {
workingState.first { it }
}
var emitted = false
taskMap.forEach { (_, model) ->
if (!model.isLoading()) {
if (model.hasRunnableDownloads()) {
emit(model)
emitted = true
}
@@ -98,10 +93,13 @@ object CacheBook {
}
}
private val modelHost = ModelHostImpl()
private val coordinator = CacheBookCoordinator()
private val stateStore = CacheDownloadStateStore()
private val pendingRequests = ConcurrentHashMap<Long, CacheDownloadRequest>()
private val pendingRemoveRequests = ConcurrentHashMap<Long, CompletableDeferred<Boolean>>()
private val pendingRequestId = AtomicLong(0)
@Volatile
private var isPaused = false
val downloadStateFlow = stateStore.stateFlow
private val _cacheSuccessFlow = MutableSharedFlow<BookChapter>(extraBufferCapacity = 64)
@@ -110,6 +108,9 @@ object CacheBook {
private val _downloadSummaryFlow = MutableStateFlow("")
val downloadSummaryFlow = _downloadSummaryFlow.asStateFlow()
private val _pendingAdmissionFlow = MutableStateFlow<Map<String, Int>>(emptyMap())
val pendingAdmissionFlow = _pendingAdmissionFlow.asStateFlow()
private val _downloadingIndicesFlow =
MutableStateFlow<Pair<String, Set<Int>>>("" to emptySet())
val downloadingIndicesFlow = _downloadingIndicesFlow.asStateFlow()
@@ -123,9 +124,7 @@ object CacheBook {
@Volatile
private var lastQueueStats = QueueStats(0, 0)
@Volatile
private var successDownloadCount = 0
private val errorRetryMap = ConcurrentHashMap<ChapterKey, Int>()
private val successDownloadCount = AtomicInteger(0)
val cacheBookMap: ConcurrentHashMap<String, CacheBookModel>
get() = coordinator.taskMap
@@ -135,27 +134,48 @@ object CacheBook {
}
fun markBookFailed(bookUrl: String, message: String) {
removePendingAdmission(bookUrl)
stateStore.markBookFailed(bookUrl, message)
updateSummary()
_queueChangedFlow.tryEmit(bookUrl)
}
private fun collectQueueStats(): QueueStats {
fun diagnostics(): Diagnostics {
var waiting = 0
var downloading = 0
var running = 0
var trackedTasks = 0
var loading = 0
var retrying = 0
cacheBookMap.forEach { (_, model) ->
val (w, d) = model.queueCounts()
waiting += w
downloading += d
val item = model.diagnostics()
waiting += item.waitingChapterCount
running += item.runningChapterCount
trackedTasks += item.trackedChapterTaskCount
if (item.isLoading) loading++
if (item.waitingRetry) retrying++
}
return QueueStats(waiting, downloading)
return Diagnostics(
activeBookCount = cacheBookMap.size,
waitingChapterCount = waiting,
runningChapterCount = running,
trackedChapterTaskCount = trackedTasks,
loadingBookCount = loading,
retryingBookCount = retrying,
)
}
private fun collectQueueStats(): QueueStats {
val state = stateStore.state
return QueueStats(
waitingCount = state.totalWaiting + _pendingAdmissionFlow.value.values.sum(),
downloadingCount = state.totalRunning,
)
}
private fun updateSummary() {
val stats = collectQueueStats()
lastQueueStats = stats
_downloadSummaryFlow.value =
"正在下载:${stats.downloadingCount}|等待中:${stats.waitingCount}|失败:${stateStore.state.totalFailure}|成功:$successDownloadCount"
_downloadSummaryFlow.value = buildSummary(stats)
}
@Synchronized
@@ -173,7 +193,7 @@ object CacheBook {
model.book = book
return model
}
val model = CacheBookModel(bookSource, book)
val model = CacheBookModel(bookSource, book, modelHost)
cacheBookMap[book.bookUrl] = model
updateSummary()
return model
@@ -211,56 +231,109 @@ object CacheBook {
fun start(context: Context, request: CacheDownloadRequest, isLocal: Boolean = false) {
if (isLocal) return
when (val selection = request.selection) {
is ChapterSelection.Range -> {
if (selection.end < selection.start) return
}
is ChapterSelection.Indices -> {
if (selection.values.isEmpty()) return
}
is ChapterSelection.Single -> Unit
}
val requestId = pendingRequestId.incrementAndGet()
pendingRequests[requestId] = request
if (!request.hasValidSelection()) return
isPaused = false
context.startService<CacheBookService> {
action = IntentAction.start
putExtra("requestId", requestId)
putExtra("bookUrl", request.bookUrl)
putExtra("source", request.source.name)
when (val selection = request.selection) {
is ChapterSelection.Range -> {
putExtra("start", selection.start)
putExtra("end", selection.end)
}
is ChapterSelection.Indices -> Unit
is ChapterSelection.Single -> {
putExtra("start", selection.index)
putExtra("end", selection.index)
}
}
putRequestExtras(request)
}
}
fun takePendingRequest(requestId: Long): CacheDownloadRequest? {
return pendingRequests.remove(requestId)
fun start(context: Context, requests: List<CacheDownloadRequest>) {
requests.asSequence()
.filter { it.hasValidSelection() }
.filter { request ->
appDb.bookDao.getBook(request.bookUrl)?.isLocal != true
}
.forEach { request ->
isPaused = false
context.startService<CacheBookService> {
action = IntentAction.start
putRequestExtras(request)
}
}
}
private fun android.content.Intent.putRequestExtras(request: CacheDownloadRequest) {
putExtra("bookUrl", request.bookUrl)
putExtra("source", request.source.name)
when (val selection = request.selection) {
is ChapterSelection.Range -> {
putExtra("start", selection.start)
putExtra("end", selection.end)
}
is ChapterSelection.Indices -> {
putExtra("indices", selection.values.toIntArray())
}
is ChapterSelection.Single -> {
putExtra("start", selection.index)
putExtra("end", selection.index)
}
}
}
fun remove(context: Context, bookUrl: String) {
if (!CacheBookService.isRun) {
removeBookFromService(bookUrl)
return
}
context.startService<CacheBookService> {
action = IntentAction.remove
putExtra("bookUrl", bookUrl)
}
}
fun removeBook(bookUrl: String): Boolean {
suspend fun removeAwait(context: Context, bookUrl: String): Boolean {
if (!CacheBookService.isRun) {
return removeBookFromService(bookUrl)
}
val requestId = pendingRequestId.incrementAndGet()
val removeRequest = CompletableDeferred<Boolean>()
pendingRemoveRequests[requestId] = removeRequest
runCatching {
context.startService<CacheBookService> {
action = IntentAction.remove
putExtra("bookUrl", bookUrl)
putExtra("removeRequestId", requestId)
}
}.onFailure {
pendingRemoveRequests.remove(requestId)
removeRequest.completeExceptionally(it)
}
return try {
withTimeout(30_000L) {
removeRequest.await()
}
} catch (_: TimeoutCancellationException) {
pendingRemoveRequests.remove(requestId)
false
}
}
internal fun completePendingRemoveRequest(requestId: Long, removed: Boolean) {
pendingRemoveRequests.remove(requestId)?.complete(removed)
}
internal fun removeBookFromService(bookUrl: String): Boolean {
val model = cacheBookMap.remove(bookUrl)
model?.stop()
removePendingAdmission(bookUrl)
stateStore.removeBook(bookUrl)
updateSummary()
_queueChangedFlow.tryEmit(bookUrl)
return model != null
}
internal fun removeModelFromService(bookUrl: String, model: CacheBookModel): Boolean {
val removed = cacheBookMap.remove(bookUrl, model)
if (!removed) return false
model.stop()
stateStore.removeBook(bookUrl)
updateSummary()
_queueChangedFlow.tryEmit(bookUrl)
return true
}
fun removeChapter(bookUrl: String, chapterIndex: Int): Boolean {
return cacheBookMap[bookUrl]?.removeDownload(chapterIndex) == true
}
@@ -273,13 +346,101 @@ object CacheBook {
}
}
fun close() {
fun pause(context: Context) {
if (CacheBookService.isRun) {
context.startService<CacheBookService> {
action = IntentAction.pause
}
} else {
pauseAllFromService()
}
}
fun resume(context: Context): Boolean {
if (!hasQueuedDownloads) return false
isPaused = false
context.startService<CacheBookService> {
action = IntentAction.resume
}
return true
}
internal fun pauseAllFromService(): Boolean {
val hadTasks = hasQueuedDownloads
if (!hadTasks) return false
isPaused = true
cacheBookMap.forEach { (bookUrl, model) ->
model.pause()
_queueChangedFlow.tryEmit(bookUrl)
}
updateSummary()
return true
}
internal fun resumeFromService() {
isPaused = false
cacheBookMap.values.forEach { it.resume() }
updateSummary()
cacheBookMap.keys.forEach { _queueChangedFlow.tryEmit(it) }
}
fun pauseBook(context: Context, bookUrl: String): Boolean {
val paused = cacheBookMap[bookUrl]?.pause() == true
if (paused) {
updateSummary()
_queueChangedFlow.tryEmit(bookUrl)
}
return paused
}
fun resumeBook(context: Context, bookUrl: String): Boolean {
val resumed = cacheBookMap[bookUrl]?.resume() == true
if (resumed) {
isPaused = false
updateSummary()
_queueChangedFlow.tryEmit(bookUrl)
context.startService<CacheBookService> {
action = IntentAction.resume
}
}
return resumed
}
fun pauseChapter(bookUrl: String, chapterIndex: Int): Boolean {
val paused = cacheBookMap[bookUrl]?.pauseDownload(chapterIndex) == true
if (paused) {
updateSummary()
_queueChangedFlow.tryEmit(bookUrl)
}
return paused
}
fun resumeChapter(context: Context, bookUrl: String, chapterIndex: Int): Boolean {
val resumed = cacheBookMap[bookUrl]?.resumeDownload(chapterIndex) == true
if (resumed) {
isPaused = false
updateSummary()
_queueChangedFlow.tryEmit(bookUrl)
context.startService<CacheBookService> {
action = IntentAction.resume
}
}
return resumed
}
fun close(clearFailureState: Boolean = false) {
isPaused = false
cacheBookMap.forEach { (_, model) -> model.stop() }
cacheBookMap.clear()
successDownloadCount = 0
errorRetryMap.clear()
pendingRequests.clear()
stateStore.clear()
successDownloadCount.set(0)
pendingRemoveRequests.values.forEach { it.complete(false) }
pendingRemoveRequests.clear()
clearPendingAdmissions()
if (clearFailureState) {
stateStore.clear()
} else {
stateStore.clearRuntimeState()
}
updateSummary()
}
@@ -294,27 +455,119 @@ object CacheBook {
val totalCount: Int
get() {
val stats = collectQueueStats()
return stats.waitingCount + stats.downloadingCount + successDownloadCount + stateStore.state.totalFailure
return stats.waitingCount + stats.downloadingCount + successDownloadCount.get() + stateStore.state.totalFailure
}
val completedCount: Int
get() = successDownloadCount + stateStore.state.totalFailure
get() = successDownloadCount.get() + stateStore.state.totalFailure
val downloadSummary: String
get() {
val stats = collectQueueStats()
return "正在下载:${stats.downloadingCount} | 等待中:${stats.waitingCount} | 失败:${stateStore.state.totalFailure} | 成功:$successDownloadCount"
return buildSummary(stats)
}
val isRun: Boolean
get() = lastQueueStats.waitingCount > 0 || lastQueueStats.downloadingCount > 0
get() = !isPaused && (
lastQueueStats.waitingCount > 0 ||
lastQueueStats.downloadingCount > 0 ||
cacheBookMap.values.any { it.hasQueuedDownloads() }
)
private fun onTaskQueuesChanged(bookUrl: String) {
val hasQueuedDownloads: Boolean
get() = cacheBookMap.values.any { it.hasQueuedDownloads() } ||
_pendingAdmissionFlow.value.isNotEmpty()
val hasPausedDownloads: Boolean
get() = (isPaused && hasQueuedDownloads) || cacheBookMap.values.any { it.isPaused() }
val isGloballyPaused: Boolean
get() = isPaused && hasQueuedDownloads
private fun buildSummary(stats: QueueStats = collectQueueStats()): String {
val hasGlobalPause = isPaused && (stats.waitingCount > 0 || stats.downloadingCount > 0)
val downloadingCount = if (hasGlobalPause) 0 else stats.downloadingCount
val waitingCount = if (hasGlobalPause) 0 else stats.waitingCount
val modelPausedCount = cacheBookMap.values.sumOf {
it.pausedCount()
}
val pausedCount = maxOf(stateStore.state.totalPaused, modelPausedCount) + if (hasGlobalPause) {
stats.waitingCount + stats.downloadingCount
} else {
0
}
return "下载中:$downloadingCount | 等待:$waitingCount | 暂停:$pausedCount | 失败:${stateStore.state.totalFailure} | 已缓存:${successDownloadCount.get()}"
}
private fun CacheDownloadRequest.hasValidSelection(): Boolean {
return when (val selection = selection) {
is ChapterSelection.Range -> selection.end >= selection.start
is ChapterSelection.Indices -> selection.values.isNotEmpty()
is ChapterSelection.Single -> true
}
}
fun addPendingAdmissions(requests: Iterable<CacheDownloadRequest>) {
val counts = requests.groupingBy { it.bookUrl }
.fold(0) { count, request -> count + request.pendingChapterCount() }
.filterValues { it > 0 }
if (counts.isEmpty()) return
_pendingAdmissionFlow.update { pending ->
pending + counts.mapValues { (bookUrl, count) ->
pending[bookUrl].orZero() + count
}
}
updateSummary()
counts.keys.forEach { _queueChangedFlow.tryEmit(it) }
}
fun removePendingAdmission(request: CacheDownloadRequest) {
val chapterCount = request.pendingChapterCount()
if (chapterCount <= 0) return
_pendingAdmissionFlow.update { pending ->
val remaining = pending[request.bookUrl].orZero() - chapterCount
if (remaining > 0) {
pending + (request.bookUrl to remaining)
} else {
pending - request.bookUrl
}
}
updateSummary()
_queueChangedFlow.tryEmit(request.bookUrl)
}
fun removePendingAdmission(bookUrl: String) {
if (!_pendingAdmissionFlow.value.containsKey(bookUrl)) return
_pendingAdmissionFlow.update { it - bookUrl }
updateSummary()
_queueChangedFlow.tryEmit(bookUrl)
}
private fun clearPendingAdmissions() {
val bookUrls = _pendingAdmissionFlow.value.keys
if (bookUrls.isEmpty()) return
_pendingAdmissionFlow.value = emptyMap()
updateSummary()
bookUrls.forEach { _queueChangedFlow.tryEmit(it) }
}
private fun CacheDownloadRequest.pendingChapterCount(): Int {
return when (val selection = selection) {
is ChapterSelection.Range -> selection.end - selection.start + 1
is ChapterSelection.Indices -> selection.values.size
is ChapterSelection.Single -> 1
}
}
private fun Int?.orZero(): Int = this ?: 0
private fun notifyTaskQueuesChanged(bookUrl: String) {
cacheBookMap[bookUrl]?.let { model ->
stateStore.updateBookQueue(
bookUrl = bookUrl,
waitingCount = model.queueCounts().first,
runningIndices = model.downloadingIndices(),
pausedIndices = model.pausedIndices(),
)
_downloadingIndicesFlow.tryEmit(bookUrl to model.downloadingIndices())
_downloadErrorFlow.tryEmit(bookUrl to errorIndices(bookUrl))
@@ -323,7 +576,7 @@ object CacheBook {
_queueChangedFlow.tryEmit(bookUrl)
}
private fun onTaskRemoved(bookUrl: String, clearState: Boolean = false) {
private fun notifyTaskRemoved(bookUrl: String, clearState: Boolean = false) {
cacheBookMap.remove(bookUrl)
if (clearState) {
stateStore.removeBook(bookUrl)
@@ -332,368 +585,29 @@ object CacheBook {
_queueChangedFlow.tryEmit(bookUrl)
}
class CacheBookModel(
@Volatile var bookSource: BookSource,
@Volatile var book: Book
) {
private class ModelHostImpl : CacheBookModel.Host {
override val stateStore: CacheDownloadStateStore
get() = CacheBook.stateStore
override val cacheBookMap: ConcurrentHashMap<String, CacheBookModel>
get() = CacheBook.cacheBookMap
private val queue = CacheDownloadQueue()
private val onDownloadSet = linkedSetOf<Int>()
private val pausedDownloadSet = hashSetOf<Int>()
private val chapterTasks = hashMapOf<Int, Coroutine<*>>()
private val tasks = CompositeCoroutine()
private val repository = CacheDownloadRepository()
private var isStopped = false
private var waitingRetry = false
private var isLoading = false
private fun notifyDownloadSetChanged() {
_downloadingIndicesFlow.tryEmit(book.bookUrl to onDownloadSet.toSet())
if (cacheBookMap[book.bookUrl] === this) {
stateStore.updateBookQueue(
bookUrl = book.bookUrl,
waitingCount = queue.waitingCount(),
runningIndices = onDownloadSet.toSet(),
)
}
override fun incrementSuccessCount(): Int = CacheBook.successDownloadCount.incrementAndGet()
override fun onTaskQueuesChanged(bookUrl: String) {
CacheBook.notifyTaskQueuesChanged(bookUrl)
}
private fun notifyErrorChanged() {
val errors = errorIndices(book.bookUrl)
_downloadErrorFlow.tryEmit(book.bookUrl to errors)
override fun onTaskRemoved(bookUrl: String, clearState: Boolean) {
CacheBook.notifyTaskRemoved(bookUrl, clearState)
}
@Synchronized
fun queueCounts(): Pair<Int, Int> = queue.waitingCount() to onDownloadSet.size
@Synchronized
fun isWaiting(index: Int): Boolean = queue.isWaiting(index)
@Synchronized
fun isDownloading(index: Int): Boolean = onDownloadSet.contains(index)
@Synchronized
fun downloadingIndices(): Set<Int> = onDownloadSet.toSet()
@Synchronized
fun isRun(): Boolean {
return queue.waitingCount() > 0 || onDownloadSet.isNotEmpty() || isLoading
override fun emitDownloadingIndices(bookUrl: String, indices: Set<Int>) {
CacheBook._downloadingIndicesFlow.tryEmit(bookUrl to indices)
}
@Synchronized
fun isStop(): Boolean {
return isStopped || (!isRun() && !waitingRetry)
override fun emitDownloadError(bookUrl: String, indices: Set<Int>) {
CacheBook._downloadErrorFlow.tryEmit(bookUrl to indices)
}
@Synchronized
fun isLoading(): Boolean = isLoading
@Synchronized
fun setLoading() {
isLoading = true
CacheBook.onTaskQueuesChanged(book.bookUrl)
}
@Synchronized
fun stop() {
queue.clear()
pausedDownloadSet.clear()
chapterTasks.clear()
tasks.clear()
isStopped = true
isLoading = false
onDownloadSet.clear()
notifyDownloadSetChanged()
CacheBook.onTaskQueuesChanged(book.bookUrl)
}
@Synchronized
fun addDownload(start: Int, end: Int) {
addRequest(
CacheDownloadRequest(
book.bookUrl,
ChapterSelection.Range(start, end),
CacheDownloadSource.ReadPreload,
)
)
}
@Synchronized
fun addDownloads(indices: Iterable<Int>) {
val values = indices.toSet()
if (values.isEmpty()) return
addRequest(
CacheDownloadRequest(
book.bookUrl,
ChapterSelection.Indices(values),
CacheDownloadSource.Manual,
)
)
}
@Synchronized
fun addRequest(request: CacheDownloadRequest) {
isStopped = false
when (val selection = request.selection) {
is ChapterSelection.Range -> {
pausedDownloadSet.removeAll { it in selection.start..selection.end }
}
is ChapterSelection.Indices -> selection.values.forEach { pausedDownloadSet.remove(it) }
is ChapterSelection.Single -> pausedDownloadSet.remove(selection.index)
}
queue.enqueue(request)
cacheBookMap[book.bookUrl] = this
isLoading = false
notifyDownloadSetChanged()
CacheBook.onTaskQueuesChanged(book.bookUrl)
}
fun addDownload(index: Int) {
addDownload(index, index)
}
@Synchronized
private fun onSuccess(chapter: BookChapter) {
onDownloadSet.remove(chapter.index)
chapterTasks.remove(chapter.index)
val chapterKey = ChapterKey(book.bookUrl, chapter.index)
successDownloadCount++
errorRetryMap.remove(chapterKey)
stateStore.markSuccess(book.bookUrl, chapter.index)
notifyDownloadSetChanged()
notifyErrorChanged()
_cacheSuccessFlow.tryEmit(chapter)
}
@Synchronized
private fun onPreError(chapter: BookChapter, error: Throwable) {
waitingRetry = true
if (error !is ConcurrentException) {
errorRetryMap.merge(ChapterKey(book.bookUrl, chapter.index), 1) { old, inc -> old + inc }
stateStore.markFailed(book.bookUrl, chapter.index)
}
onDownloadSet.remove(chapter.index)
chapterTasks.remove(chapter.index)
}
@Synchronized
private fun onPostError(chapter: BookChapter, error: Throwable) {
val retryCount = errorRetryMap[ChapterKey(book.bookUrl, chapter.index)] ?: 0
if (retryCount < 3 && !isStopped) {
queue.enqueue(ChapterSelection.Single(chapter.index))
} else {
AppLog.put("下载${book.name}-${chapter.title}失败\n${error.localizedMessage}", error)
}
waitingRetry = false
}
@Synchronized
private fun onError(chapter: BookChapter, error: Throwable) {
onPreError(chapter, error)
onPostError(chapter, error)
notifyDownloadSetChanged()
notifyErrorChanged()
}
@Synchronized
private fun onCancel(index: Int) {
onDownloadSet.remove(index)
chapterTasks.remove(index)
if (!isStopped && !pausedDownloadSet.remove(index)) {
queue.enqueue(ChapterSelection.Single(index))
}
notifyDownloadSetChanged()
}
@Synchronized
private fun onFinally() {
val bookUrl = book.bookUrl
if (queue.waitingCount() == 0 && onDownloadSet.isEmpty()) {
CacheBook.onTaskRemoved(bookUrl)
} else {
CacheBook.onTaskQueuesChanged(bookUrl)
}
notifyDownloadSetChanged()
}
@Synchronized
fun removeDownload(index: Int): Boolean {
val removedWaiting = queue.removeChapter(index)
val task = chapterTasks.remove(index)
val removedRunning = onDownloadSet.contains(index) || task != null
if (removedRunning) {
pausedDownloadSet.add(index)
task?.let {
tasks.delete(it)
it.cancel()
}
}
if (!removedWaiting && !removedRunning) return false
notifyDownloadSetChanged()
if (queue.waitingCount() == 0 && onDownloadSet.isEmpty()) {
CacheBook.onTaskRemoved(book.bookUrl, clearState = true)
} else {
CacheBook.onTaskQueuesChanged(book.bookUrl)
}
return true
}
/**
* 从待下载列表内取第一条下载
*/
@Synchronized
fun download(scope: CoroutineScope, context: CoroutineContext) {
val candidate = queue.next(book.bookUrl, onDownloadSet)
if (candidate == null) {
if (!isLoading && onDownloadSet.isEmpty()) {
CacheBook.onTaskRemoved(book.bookUrl)
}
return
}
val chapterIndex = candidate.chapterIndex
if (onDownloadSet.contains(chapterIndex)) {
return
}
val chapter = repository.getChapter(book.bookUrl, chapterIndex) ?: run {
return
}
if (chapter.isVolume) {
_cacheSuccessFlow.tryEmit(chapter)
return
}
if (repository.hasImageContent(book, chapter)) {
return
}
onDownloadSet.add(chapterIndex)
notifyDownloadSetChanged()
if (repository.hasContent(book, chapter)) {
val task = repository.saveCachedImagesTask(
scope = scope,
context = context,
bookSource = bookSource,
book = book,
chapter = chapter,
).onSuccess {
onSuccess(chapter)
}.onError {
onPreError(chapter, it)
delay(1000)
onPostError(chapter, it)
}.onCancel {
onCancel(chapterIndex)
}.onFinally {
chapterTasks.remove(chapterIndex)?.let { tasks.delete(it) }
onFinally()
}
chapterTasks[chapterIndex] = task
tasks.add(task)
return
}
val task = repository.cacheContentTask(
scope = scope,
bookSource = bookSource,
book = book,
chapter = chapter,
context = context,
start = CoroutineStart.LAZY,
executeContext = context,
).onSuccess {
onSuccess(chapter)
}.onError {
onPreError(chapter, it)
delay(1000)
onPostError(chapter, it)
}.onCancel {
onCancel(chapterIndex)
}.onFinally {
chapterTasks.remove(chapterIndex)?.let { tasks.delete(it) }
onFinally()
}
chapterTasks[chapterIndex] = task
tasks.add(task)
task.start()
}
suspend fun downloadAwait(chapter: BookChapter): String {
synchronized(this) {
onDownloadSet.add(chapter.index)
queue.removeChapter(chapter.index)
notifyDownloadSetChanged()
}
try {
val content = repository.downloadContentAwait(bookSource, book, chapter)
onSuccess(chapter)
ReadBook.downloadedChapters.add(chapter.index)
ReadBook.downloadFailChapters.remove(chapter.index)
return content
} catch (e: Exception) {
if (e is CancellationException) onCancel(chapter.index)
onError(chapter, e)
ReadBook.downloadFailChapters[chapter.index] =
(ReadBook.downloadFailChapters[chapter.index] ?: 0) + 1
return "获取正文失败\n${e.localizedMessage}"
} finally {
CacheBook.onTaskQueuesChanged(book.bookUrl)
}
}
@Synchronized
fun download(
scope: CoroutineScope,
chapter: BookChapter,
semaphore: Semaphore?,
resetPageOffset: Boolean = false
) {
if (onDownloadSet.contains(chapter.index)) return
onDownloadSet.add(chapter.index)
queue.removeChapter(chapter.index)
notifyDownloadSetChanged()
repository.downloadContentTask(
scope = scope,
bookSource = bookSource,
book = book,
chapter = chapter,
start = CoroutineStart.LAZY,
context = IO,
executeContext = IO,
semaphore = semaphore
).onSuccess { content ->
onSuccess(chapter)
ReadBook.downloadedChapters.add(chapter.index)
ReadBook.downloadFailChapters.remove(chapter.index)
downloadFinish(chapter, content, resetPageOffset)
}.onError {
onError(chapter, it)
ReadBook.downloadFailChapters[chapter.index] =
(ReadBook.downloadFailChapters[chapter.index] ?: 0) + 1
downloadFinish(chapter, "获取正文失败\n${it.localizedMessage}", resetPageOffset)
}.onCancel {
onCancel(chapter.index)
downloadFinish(chapter, "download canceled", resetPageOffset, true)
}.onFinally {
CacheBook.onTaskQueuesChanged(book.bookUrl)
}.start()
}
private fun downloadFinish(
chapter: BookChapter,
content: String,
resetPageOffset: Boolean = false,
canceled: Boolean = false
) {
ReadingCacheEvents.emit(
ReadingCacheEvent.ContentReady(
book = book,
chapter = chapter,
content = content,
resetPageOffset = resetPageOffset,
canceled = canceled,
)
)
override fun emitChapterCached(chapter: BookChapter) {
CacheBook._cacheSuccessFlow.tryEmit(chapter)
}
override fun errorIndices(bookUrl: String): Set<Int> =
CacheBook.errorIndices(bookUrl)
}
}
@@ -0,0 +1,592 @@
package io.legado.app.model
import io.legado.app.constant.AppLog
import io.legado.app.data.entities.Book
import io.legado.app.data.entities.BookChapter
import io.legado.app.data.entities.BookSource
import io.legado.app.exception.ConcurrentException
import io.legado.app.help.coroutine.CompositeCoroutine
import io.legado.app.help.coroutine.Coroutine
import io.legado.app.model.cache.CacheDownloadCandidate
import io.legado.app.model.cache.CacheDownloadQueue
import io.legado.app.model.cache.CacheDownloadRepository
import io.legado.app.model.cache.CacheDownloadRequest
import io.legado.app.model.cache.CacheDownloadSource
import io.legado.app.model.cache.CacheDownloadStateStore
import io.legado.app.model.cache.ChapterSelection
import io.legado.app.model.cache.ReadingCacheEvent
import io.legado.app.model.cache.ReadingCacheEvents
import kotlinx.coroutines.CancellationException
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.CoroutineStart
import kotlinx.coroutines.Dispatchers.IO
import kotlinx.coroutines.delay
import kotlinx.coroutines.sync.Semaphore
import java.util.concurrent.ConcurrentHashMap
import kotlin.coroutines.CoroutineContext
class CacheBookModel(
@Volatile var bookSource: BookSource,
@Volatile var book: Book,
private val host: Host,
) {
interface Host {
val stateStore: CacheDownloadStateStore
val cacheBookMap: ConcurrentHashMap<String, CacheBookModel>
fun incrementSuccessCount(): Int
fun onTaskQueuesChanged(bookUrl: String)
fun onTaskRemoved(bookUrl: String, clearState: Boolean = false)
fun emitDownloadingIndices(bookUrl: String, indices: Set<Int>)
fun emitDownloadError(bookUrl: String, indices: Set<Int>)
fun emitChapterCached(chapter: BookChapter)
fun errorIndices(bookUrl: String): Set<Int>
}
data class Diagnostics(
val waitingChapterCount: Int,
val runningChapterCount: Int,
val trackedChapterTaskCount: Int,
val isLoading: Boolean,
val waitingRetry: Boolean,
)
private val queue = CacheDownloadQueue()
private val onDownloadSet = linkedSetOf<Int>()
private val canceledDownloadSet = hashSetOf<Int>()
private val pausedChapterSet = linkedSetOf<Int>()
private val chapterTasks = hashMapOf<Int, Coroutine<*>>()
private val tasks = CompositeCoroutine()
private val repository = CacheDownloadRepository()
private val retryCountMap = hashMapOf<Int, Int>()
private var isStopped = false
private var waitingRetry = false
private var isLoading = false
private var isPaused = false
private fun notifyDownloadSetChanged() {
host.emitDownloadingIndices(book.bookUrl, onDownloadSet.toSet())
if (host.cacheBookMap[book.bookUrl] === this) {
host.stateStore.updateBookQueue(
bookUrl = book.bookUrl,
waitingCount = queue.waitingCount(),
runningIndices = onDownloadSet.toSet(),
pausedIndices = pausedChapterSet.toSet(),
)
}
}
private fun notifyErrorChanged() {
val errors = host.errorIndices(book.bookUrl)
host.emitDownloadError(book.bookUrl, errors)
}
@Synchronized
fun queueCounts(): Pair<Int, Int> = queue.waitingCount() to onDownloadSet.size
@Synchronized
fun isWaiting(index: Int): Boolean = queue.isWaiting(index)
@Synchronized
fun isDownloading(index: Int): Boolean = onDownloadSet.contains(index)
@Synchronized
fun isPaused(): Boolean = isPaused
@Synchronized
fun isPaused(index: Int): Boolean {
return pausedChapterSet.contains(index) ||
(isPaused && (queue.isWaiting(index) || onDownloadSet.contains(index)))
}
@Synchronized
fun downloadingIndices(): Set<Int> = onDownloadSet.toSet()
@Synchronized
fun pausedIndices(): Set<Int> = pausedChapterSet.toSet()
@Synchronized
fun pausedCount(): Int {
return pausedChapterSet.size + if (isPaused) {
queue.waitingCount() + onDownloadSet.size
} else {
0
}
}
@Synchronized
fun isRun(): Boolean {
return queue.waitingCount() > 0 || onDownloadSet.isNotEmpty() || isLoading || chapterTasks.isNotEmpty()
}
@Synchronized
fun isStop(): Boolean {
return isStopped || (!isRun() && !waitingRetry)
}
@Synchronized
fun isLoading(): Boolean = isLoading
@Synchronized
fun hasQueuedDownloads(): Boolean {
return queue.waitingCount() > 0 ||
onDownloadSet.isNotEmpty() ||
pausedChapterSet.isNotEmpty() ||
isLoading ||
chapterTasks.isNotEmpty()
}
@Synchronized
fun hasRunnableDownloads(): Boolean {
return !isPaused && (
queue.waitingCount() > 0 ||
onDownloadSet.isNotEmpty() ||
isLoading ||
chapterTasks.isNotEmpty()
)
}
@Synchronized
fun diagnostics(): Diagnostics {
return Diagnostics(
waitingChapterCount = queue.waitingCount(),
runningChapterCount = onDownloadSet.size,
trackedChapterTaskCount = chapterTasks.size,
isLoading = isLoading,
waitingRetry = waitingRetry,
)
}
@Synchronized
fun setLoading() {
isLoading = true
host.onTaskQueuesChanged(book.bookUrl)
}
@Synchronized
fun pause(): Boolean {
if (!hasQueuedDownloads()) return false
isPaused = true
isLoading = false
waitingRetry = false
chapterTasks.values.toList().forEach { task ->
tasks.delete(task)
task.cancel()
}
notifyDownloadSetChanged()
host.onTaskQueuesChanged(book.bookUrl)
return true
}
@Synchronized
fun resume(): Boolean {
if (!isPaused && pausedChapterSet.isEmpty()) return false
isPaused = false
if (pausedChapterSet.isNotEmpty()) {
queue.enqueue(ChapterSelection.Indices(pausedChapterSet.toSet()))
pausedChapterSet.clear()
}
notifyDownloadSetChanged()
host.onTaskQueuesChanged(book.bookUrl)
return true
}
@Synchronized
fun stop() {
queue.clear()
canceledDownloadSet.clear()
pausedChapterSet.clear()
chapterTasks.clear()
tasks.clear()
retryCountMap.clear()
isStopped = true
isPaused = false
isLoading = false
onDownloadSet.clear()
notifyDownloadSetChanged()
host.onTaskQueuesChanged(book.bookUrl)
}
@Synchronized
fun addDownload(start: Int, end: Int) {
addRequest(
CacheDownloadRequest(
book.bookUrl,
ChapterSelection.Range(start, end),
CacheDownloadSource.ReadPreload,
)
)
}
@Synchronized
fun addDownloads(indices: Iterable<Int>) {
val values = indices.toSet()
if (values.isEmpty()) return
addRequest(
CacheDownloadRequest(
book.bookUrl,
ChapterSelection.Indices(values),
CacheDownloadSource.Manual,
)
)
}
@Synchronized
fun addRequest(request: CacheDownloadRequest) {
isStopped = false
isPaused = false
when (val selection = request.selection) {
is ChapterSelection.Range -> {
canceledDownloadSet.removeAll { it in selection.start..selection.end }
pausedChapterSet.removeAll { it in selection.start..selection.end }
}
is ChapterSelection.Indices -> selection.values.forEach {
canceledDownloadSet.remove(it)
pausedChapterSet.remove(it)
}
is ChapterSelection.Single -> {
canceledDownloadSet.remove(selection.index)
pausedChapterSet.remove(selection.index)
}
}
queue.enqueue(request)
host.cacheBookMap[book.bookUrl] = this
isLoading = false
notifyDownloadSetChanged()
host.onTaskQueuesChanged(book.bookUrl)
}
fun addDownload(index: Int) {
addDownload(index, index)
}
@Synchronized
private fun onSuccess(chapter: BookChapter) {
onDownloadSet.remove(chapter.index)
chapterTasks.remove(chapter.index)
host.incrementSuccessCount()
retryCountMap.remove(chapter.index)
host.stateStore.markSuccess(book.bookUrl, chapter.index)
notifyDownloadSetChanged()
notifyErrorChanged()
host.emitChapterCached(chapter)
}
@Synchronized
private fun onPreError(chapter: BookChapter, error: Throwable) {
waitingRetry = true
if (error !is ConcurrentException) {
retryCountMap[chapter.index] = (retryCountMap[chapter.index] ?: 0) + 1
host.stateStore.markFailed(book.bookUrl, chapter.index)
}
onDownloadSet.remove(chapter.index)
chapterTasks.remove(chapter.index)
}
@Synchronized
private fun onPostError(chapter: BookChapter, error: Throwable) {
val retryCount = retryCountMap[chapter.index] ?: 0
if (retryCount < 3 && !isStopped) {
queue.enqueue(ChapterSelection.Single(chapter.index))
} else {
AppLog.put("下载${book.name}-${chapter.title}失败\n${error.localizedMessage}", error)
}
waitingRetry = false
}
@Synchronized
private fun onError(chapter: BookChapter, error: Throwable) {
onPreError(chapter, error)
onPostError(chapter, error)
notifyDownloadSetChanged()
notifyErrorChanged()
}
@Synchronized
private fun onCancel(index: Int, requeue: Boolean = true) {
onDownloadSet.remove(index)
chapterTasks.remove(index)
val wasCanceled = canceledDownloadSet.remove(index)
if (requeue && !isStopped && !wasCanceled) {
queue.enqueue(ChapterSelection.Single(index))
}
notifyDownloadSetChanged()
}
@Synchronized
private fun onFinally() {
val bookUrl = book.bookUrl
if (queue.waitingCount() == 0 && onDownloadSet.isEmpty() && pausedChapterSet.isEmpty()) {
host.onTaskRemoved(bookUrl)
} else {
host.onTaskQueuesChanged(bookUrl)
}
notifyDownloadSetChanged()
}
@Synchronized
fun removeDownload(index: Int): Boolean {
val removedWaiting = queue.removeChapter(index)
val removedPaused = pausedChapterSet.remove(index)
val task = chapterTasks.remove(index)
val removedRunning = onDownloadSet.contains(index) || task != null
if (removedRunning) {
canceledDownloadSet.add(index)
onDownloadSet.remove(index)
task?.let {
tasks.delete(it)
it.cancel()
}
}
if (!removedWaiting && !removedRunning && !removedPaused) return false
notifyDownloadSetChanged()
if (queue.waitingCount() == 0 && onDownloadSet.isEmpty() && pausedChapterSet.isEmpty()) {
host.onTaskRemoved(book.bookUrl, clearState = true)
} else {
host.onTaskQueuesChanged(book.bookUrl)
}
return true
}
@Synchronized
fun pauseDownload(index: Int): Boolean {
val removedWaiting = queue.removeChapter(index)
val task = chapterTasks.remove(index)
val removedRunning = onDownloadSet.contains(index) || task != null
if (!removedWaiting && !removedRunning) return false
pausedChapterSet.add(index)
if (removedRunning) {
canceledDownloadSet.add(index)
onDownloadSet.remove(index)
task?.let {
tasks.delete(it)
it.cancel()
}
}
notifyDownloadSetChanged()
host.onTaskQueuesChanged(book.bookUrl)
return true
}
@Synchronized
fun resumeDownload(index: Int): Boolean {
if (!pausedChapterSet.remove(index)) return false
isPaused = false
queue.enqueue(ChapterSelection.Single(index))
notifyDownloadSetChanged()
host.onTaskQueuesChanged(book.bookUrl)
return true
}
/**
* 从待下载列表内取第一条下载
*/
fun download(scope: CoroutineScope, context: CoroutineContext) {
val candidate = nextDownloadCandidate() ?: return
val chapterIndex = candidate.chapterIndex
val chapter = repository.getChapter(book.bookUrl, chapterIndex) ?: run {
onSkipped(chapterIndex)
return
}
if (chapter.isVolume) {
host.emitChapterCached(chapter)
onSkipped(chapterIndex)
return
}
if (repository.hasImageContent(book, chapter)) {
onSkipped(chapterIndex)
return
}
if (repository.hasContent(book, chapter)) {
val task = repository.saveCachedImagesTask(
scope = scope,
context = context,
bookSource = bookSource,
book = book,
chapter = chapter,
start = CoroutineStart.LAZY,
)
if (!attachTaskIfActive(task, chapter, chapterIndex)) {
task.cancel()
return
}
task.start()
return
}
val task = repository.cacheContentTask(
scope = scope,
bookSource = bookSource,
book = book,
chapter = chapter,
context = context,
start = CoroutineStart.LAZY,
executeContext = context,
)
if (!attachTaskIfActive(task, chapter, chapterIndex)) {
task.cancel()
return
}
task.start()
}
@Synchronized
private fun nextDownloadCandidate(): CacheDownloadCandidate? {
if (isPaused) return null
val candidate = queue.next(book.bookUrl, onDownloadSet)
if (candidate == null) {
notifyDownloadSetChanged()
if (!isLoading && onDownloadSet.isEmpty() && pausedChapterSet.isEmpty()) {
host.onTaskRemoved(book.bookUrl)
} else {
host.onTaskQueuesChanged(book.bookUrl)
}
return null
}
val chapterIndex = candidate.chapterIndex
if (onDownloadSet.contains(chapterIndex)) {
return null
}
onDownloadSet.add(chapterIndex)
notifyDownloadSetChanged()
return candidate
}
@Synchronized
private fun onSkipped(index: Int) {
onDownloadSet.remove(index)
chapterTasks.remove(index)
notifyDownloadSetChanged()
if (queue.waitingCount() == 0 && onDownloadSet.isEmpty() && pausedChapterSet.isEmpty() && !isLoading) {
host.onTaskRemoved(book.bookUrl)
} else {
host.onTaskQueuesChanged(book.bookUrl)
}
}
@Synchronized
private fun <T> attachTaskIfActive(
task: Coroutine<T>,
chapter: BookChapter,
chapterIndex: Int,
): Boolean {
if (isStopped || isPaused || !onDownloadSet.contains(chapterIndex)) {
if (!isStopped && isPaused && onDownloadSet.remove(chapterIndex)) {
queue.enqueue(ChapterSelection.Single(chapterIndex))
notifyDownloadSetChanged()
host.onTaskQueuesChanged(book.bookUrl)
}
return false
}
attachCallbacks(task, chapter, chapterIndex)
chapterTasks[chapterIndex] = task
tasks.add(task)
return true
}
private fun <T> attachCallbacks(
task: Coroutine<T>,
chapter: BookChapter,
chapterIndex: Int,
) {
task.onSuccess {
onSuccess(chapter)
}.onError {
onPreError(chapter, it)
delay(1000)
onPostError(chapter, it)
}.onCancel {
onCancel(chapterIndex)
}.onFinally {
chapterTasks.remove(chapterIndex)?.let { tasks.delete(it) }
onFinally()
}
}
suspend fun downloadAwait(chapter: BookChapter): String {
synchronized(this) {
onDownloadSet.add(chapter.index)
queue.removeChapter(chapter.index)
notifyDownloadSetChanged()
}
try {
val content = repository.downloadContentAwait(bookSource, book, chapter)
onSuccess(chapter)
ReadBook.downloadedChapters.add(chapter.index)
ReadBook.downloadFailChapters.remove(chapter.index)
return content
} catch (e: Exception) {
if (e is CancellationException) {
onCancel(chapter.index, requeue = false)
return "download canceled"
}
onError(chapter, e)
ReadBook.downloadFailChapters[chapter.index] =
(ReadBook.downloadFailChapters[chapter.index] ?: 0) + 1
return "获取正文失败\n${e.localizedMessage}"
} finally {
host.onTaskQueuesChanged(book.bookUrl)
}
}
fun download(
scope: CoroutineScope,
chapter: BookChapter,
semaphore: Semaphore?,
resetPageOffset: Boolean = false
) {
if (!markChapterDownloadStarted(chapter.index)) return
repository.downloadContentTask(
scope = scope,
bookSource = bookSource,
book = book,
chapter = chapter,
start = CoroutineStart.LAZY,
context = IO,
executeContext = IO,
semaphore = semaphore
).onSuccess { content ->
onSuccess(chapter)
ReadBook.downloadedChapters.add(chapter.index)
ReadBook.downloadFailChapters.remove(chapter.index)
downloadFinish(chapter, content, resetPageOffset)
}.onError {
onError(chapter, it)
ReadBook.downloadFailChapters[chapter.index] =
(ReadBook.downloadFailChapters[chapter.index] ?: 0) + 1
downloadFinish(chapter, "获取正文失败\n${it.localizedMessage}", resetPageOffset)
}.onCancel {
onCancel(chapter.index, requeue = false)
downloadFinish(chapter, "download canceled", resetPageOffset, true)
}.onFinally {
host.onTaskQueuesChanged(book.bookUrl)
}.start()
}
@Synchronized
private fun markChapterDownloadStarted(index: Int): Boolean {
if (onDownloadSet.contains(index)) return false
onDownloadSet.add(index)
queue.removeChapter(index)
notifyDownloadSetChanged()
return true
}
private fun downloadFinish(
chapter: BookChapter,
content: String,
resetPageOffset: Boolean = false,
canceled: Boolean = false
) {
ReadingCacheEvents.emit(
ReadingCacheEvent.ContentReady(
book = book,
chapter = chapter,
content = content,
resetPageOffset = resetPageOffset,
canceled = canceled,
)
)
}
}
@@ -0,0 +1,51 @@
package io.legado.app.model.cache
class CacheDownloadAdmissionQueue(
private val maxActiveBooks: Int
) {
private val requests = ArrayDeque<CacheDownloadRequest>()
val size: Int
get() = requests.size
fun isEmpty(): Boolean = requests.isEmpty()
fun shouldQueue(request: CacheDownloadRequest, activeBookUrls: Set<String>): Boolean {
return request.bookUrl !in activeBookUrls && activeBookUrls.size >= maxActiveBooks
}
fun add(request: CacheDownloadRequest) {
requests.addLast(request)
}
fun pollReady(activeBookUrls: Set<String>): CacheDownloadRequest? {
if (requests.isEmpty()) return null
val activeBookRequestIndex = requests.indexOfFirst {
it.bookUrl in activeBookUrls
}
if (activeBookRequestIndex >= 0) {
return requests.removeAt(activeBookRequestIndex)
}
if (activeBookUrls.size >= maxActiveBooks) return null
return requests.removeFirst()
}
fun removeBook(bookUrl: String): Boolean {
var removed = false
val iterator = requests.iterator()
while (iterator.hasNext()) {
if (iterator.next().bookUrl == bookUrl) {
iterator.remove()
removed = true
}
}
return removed
}
fun clear() {
requests.clear()
}
}
@@ -31,6 +31,7 @@ data class CacheDownloadState(
val isRunning: Boolean = false,
val totalWaiting: Int = 0,
val totalRunning: Int = 0,
val totalPaused: Int = 0,
val totalSuccess: Int = 0,
val totalFailure: Int = 0,
val books: Map<String, CacheBookDownloadState> = emptyMap(),
@@ -40,6 +41,7 @@ data class CacheBookDownloadState(
val bookUrl: String,
val waitingCount: Int = 0,
val runningIndices: Set<Int> = emptySet(),
val pausedIndices: Set<Int> = emptySet(),
val failedIndices: Set<Int> = emptySet(),
val successCount: Int = 0,
val failureMessage: String? = null,
@@ -32,8 +32,9 @@ class CacheDownloadRepository {
bookSource: BookSource,
book: Book,
chapter: BookChapter,
start: CoroutineStart = CoroutineStart.LAZY,
): Coroutine<Unit> {
return Coroutine.async(scope, context, executeContext = context) {
return Coroutine.async(scope, context, start = start, executeContext = context) {
BookHelp.getContent(book, chapter)?.let {
BookHelp.saveImages(bookSource, book, chapter, it, 1)
}
@@ -16,12 +16,14 @@ class CacheDownloadStateStore {
bookUrl: String,
waitingCount: Int,
runningIndices: Set<Int>,
pausedIndices: Set<Int> = emptySet(),
) {
updateBook(bookUrl) { current ->
current.copy(
waitingCount = waitingCount,
runningIndices = runningIndices,
failureMessage = if (waitingCount > 0 || runningIndices.isNotEmpty()) {
pausedIndices = pausedIndices,
failureMessage = if (waitingCount > 0 || runningIndices.isNotEmpty() || pausedIndices.isNotEmpty()) {
null
} else {
current.failureMessage
@@ -34,6 +36,7 @@ class CacheDownloadStateStore {
updateBook(bookUrl) { current ->
current.copy(
runningIndices = current.runningIndices - chapterIndex,
pausedIndices = current.pausedIndices - chapterIndex,
failedIndices = current.failedIndices - chapterIndex,
successCount = current.successCount + 1,
failureMessage = null,
@@ -45,6 +48,7 @@ class CacheDownloadStateStore {
updateBook(bookUrl) { current ->
current.copy(
runningIndices = current.runningIndices - chapterIndex,
pausedIndices = current.pausedIndices - chapterIndex,
failedIndices = current.failedIndices + chapterIndex,
)
}
@@ -55,6 +59,8 @@ class CacheDownloadStateStore {
current.copy(
waitingCount = 0,
runningIndices = emptySet(),
pausedIndices = emptySet(),
failedIndices = emptySet(),
failureMessage = message,
)
}
@@ -76,6 +82,25 @@ class CacheDownloadStateStore {
_stateFlow.value = CacheDownloadState()
}
fun clearRuntimeState() {
_stateFlow.update { state ->
val failureBooks = state.books
.mapValues { (_, bookState) ->
bookState.copy(
waitingCount = 0,
runningIndices = emptySet(),
successCount = 0,
)
}
.filterValues { bookState ->
bookState.pausedIndices.isNotEmpty() ||
bookState.failedIndices.isNotEmpty() ||
bookState.failureMessage != null
}
state.copy(books = failureBooks).recalculate()
}
}
fun bookState(bookUrl: String): CacheBookDownloadState? {
return state.books[bookUrl]
}
@@ -93,13 +118,15 @@ class CacheDownloadStateStore {
private fun CacheDownloadState.recalculate(): CacheDownloadState {
val totalWaiting = books.values.sumOf { it.waitingCount }
val totalRunning = books.values.sumOf { it.runningIndices.size }
val totalPaused = books.values.sumOf { it.pausedIndices.size }
val totalFailure = books.values.sumOf { it.failedIndices.size } +
books.values.count { it.failureMessage != null }
val totalSuccess = books.values.sumOf { it.successCount }
return copy(
isRunning = totalWaiting > 0 || totalRunning > 0,
isRunning = totalWaiting > totalPaused || totalRunning > 0,
totalWaiting = totalWaiting,
totalRunning = totalRunning,
totalPaused = totalPaused,
totalFailure = totalFailure,
totalSuccess = totalSuccess,
)
@@ -12,16 +12,25 @@ import io.legado.app.constant.NotificationId
import io.legado.app.data.appDb
import io.legado.app.help.book.update
import io.legado.app.model.CacheBook
import io.legado.app.model.cache.CacheDownloadAdmissionQueue
import io.legado.app.model.cache.CacheDownloadRequest
import io.legado.app.model.cache.CacheDownloadSource
import io.legado.app.model.cache.ChapterSelection
import io.legado.app.model.webBook.WebBook
import io.legado.app.ui.config.otherConfig.OtherConfig
import io.legado.app.ui.main.MainActivity
import io.legado.app.utils.LogUtils
import io.legado.app.utils.activityPendingIntent
import io.legado.app.utils.servicePendingIntent
import kotlinx.coroutines.CompletableDeferred
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.ExecutorCoroutineDispatcher
import kotlinx.coroutines.Job
import kotlinx.coroutines.SupervisorJob
import kotlinx.coroutines.asCoroutineDispatcher
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.cancel
import kotlinx.coroutines.currentCoroutineContext
import kotlinx.coroutines.delay
import kotlinx.coroutines.isActive
import kotlinx.coroutines.launch
@@ -38,16 +47,29 @@ import kotlin.math.min
class CacheBookService : BaseService() {
companion object {
private const val MB = 1024L * 1024L
private const val DIAGNOSTICS_LOG_INTERVAL_MILLIS = 5_000L
var isRun = false
private set
}
private val threadCount = OtherConfig.cacheBookThreadCount.coerceIn(1, CacheBook.maxDownloadConcurrency)
private var cachePool =
Executors.newFixedThreadPool(min(threadCount, AppConst.MAX_THREAD)).asCoroutineDispatcher()
private val maxActiveBookCount = (threadCount * 2).coerceAtLeast(1)
private val admissionQueue = CacheDownloadAdmissionQueue(maxActiveBookCount)
private val admissionLock = Any()
private val admittingBookUrls = hashSetOf<String>()
private val admissionGenerations = hashMapOf<String, Long>()
private val admissionBuffers = hashMapOf<String, ArrayDeque<AdmissionRequest>>()
private val admissionIdleWaiters = hashMapOf<String, MutableList<CompletableDeferred<Unit>>>()
private lateinit var cachePool: ExecutorCoroutineDispatcher
private val serviceCommandScope = CoroutineScope(SupervisorJob() + Dispatchers.IO)
private val serviceCommandMutex = Mutex()
private val downloadJobLock = Any()
private var downloadJob: Job? = null
private var notificationContent = appCtx.getString(R.string.service_starting)
private var mutex = Mutex()
private var lastDiagnosticsLogTime = 0L
private val notificationBuilder by lazy {
val builder = NotificationCompat.Builder(this, AppConst.channelIdDownload)
.setSmallIcon(R.drawable.ic_download)
@@ -68,12 +90,24 @@ class CacheBookService : BaseService() {
builder.setVisibility(NotificationCompat.VISIBILITY_PUBLIC)
}
private data class AdmissionRequest(
val request: CacheDownloadRequest,
val fromAdmissionQueue: Boolean,
val generation: Long,
)
override fun onCreate() {
super.onCreate()
if (::cachePool.isInitialized) {
cachePool.close()
}
cachePool = Executors.newFixedThreadPool(min(threadCount, AppConst.MAX_THREAD)).asCoroutineDispatcher()
isRun = true
lifecycleScope.launch {
while (isActive) {
lifecycleScope.launch(Dispatchers.IO) {
while (currentCoroutineContext().isActive) {
delay(1000)
drainPendingDownloadRequests()
logDownloadDiagnostics()
notificationContent = CacheBook.downloadSummary
upCacheBookNotification()
}
@@ -81,119 +115,361 @@ class CacheBookService : BaseService() {
}
override fun onStartCommand(intent: Intent?, flags: Int, startId: Int): Int {
var startResult = START_REDELIVER_INTENT
intent?.action?.let { action ->
when (action) {
IntentAction.start -> {
val requestId = intent.getLongExtra("requestId", -1L)
val request = if (requestId >= 0) {
CacheBook.takePendingRequest(requestId)
} else {
null
}
if (request != null) {
addDownloadRequest(request)
} else {
val bookUrl = intent.getStringExtra("bookUrl") ?: return@let
addDownloadData(
bookUrl,
intent.getIntExtra("start", 0),
intent.getIntExtra("end", 0)
)
val request = reconstructRequestFromIntent(intent) ?: run {
stopIfIdle()
return@let
}
addDownloadRequest(request)
}
IntentAction.remove -> {
val bookUrl = intent.getStringExtra("bookUrl")
bookUrl?.let { CacheBook.removeBook(it) }
val removeRequestId = intent.getLongExtra("removeRequestId", -1L)
bookUrl?.let {
lifecycleScope.launch(Dispatchers.IO) {
val removed = removeBookCompletely(it)
if (removeRequestId >= 0L) {
CacheBook.completePendingRemoveRequest(removeRequestId, removed)
}
stopIfIdle()
}
}
}
IntentAction.stop -> {
startResult = START_NOT_STICKY
stopSelf()
}
IntentAction.pause -> {
serviceCommandScope.launch {
serviceCommandMutex.withLock {
CacheBook.pauseAllFromService()
notificationContent = CacheBook.downloadSummary
upCacheBookNotification()
}
}
}
IntentAction.resume -> {
serviceCommandScope.launch {
serviceCommandMutex.withLock {
CacheBook.resumeFromService()
ensureDownloadJob()
notificationContent = CacheBook.downloadSummary
upCacheBookNotification()
}
}
}
IntentAction.stop -> stopSelf()
}
}
return super.onStartCommand(intent, flags, startId)
super.onStartCommand(intent, flags, startId)
return startResult
}
override fun onDestroy() {
isRun = false
cachePool.close()
CacheBook.close()
if (::cachePool.isInitialized) {
cachePool.close()
}
synchronized(admissionQueue) {
admissionQueue.clear()
}
synchronized(admissionLock) {
admissionBuffers.clear()
admittingBookUrls.clear()
admissionIdleWaiters.values.flatten().forEach { it.complete(Unit) }
admissionIdleWaiters.clear()
}
serviceCommandScope.launch {
try {
serviceCommandMutex.withLock {
CacheBook.close(clearFailureState = false)
}
} finally {
serviceCommandScope.cancel()
}
}
super.onDestroy()
}
private fun addDownloadData(bookUrl: String?, start: Int, end: Int) {
bookUrl ?: return
if (end < start) return
addDownloadRequest(
CacheDownloadRequest(
private fun reconstructRequestFromIntent(intent: Intent): CacheDownloadRequest? {
val bookUrl = intent.getStringExtra("bookUrl") ?: return null
val sourceName = intent.getStringExtra("source")
val source = sourceName?.let { name ->
runCatching { CacheDownloadSource.valueOf(name) }.getOrDefault(CacheDownloadSource.Manual)
} ?: CacheDownloadSource.Manual
val indices = intent.getIntArrayExtra("indices")
if (indices != null && indices.isNotEmpty()) {
return CacheDownloadRequest(
bookUrl = bookUrl,
selection = ChapterSelection.Range(start, end),
source = CacheDownloadSource.Manual,
selection = ChapterSelection.Indices(indices.toSet()),
source = source,
)
}
val start = intent.getIntExtra("start", 0)
val end = intent.getIntExtra("end", 0)
if (end < start) return null
return CacheDownloadRequest(
bookUrl = bookUrl,
selection = ChapterSelection.Range(start, end),
source = source,
)
}
private fun addDownloadRequest(request: CacheDownloadRequest) {
execute {
val cacheBook = CacheBook.getOrCreate(request.bookUrl) ?: run {
CacheBook.markBookFailed(request.bookUrl, getString(R.string.error_no_source))
return@execute
addDownloadRequestsToQueue(listOf(request))
}
private fun addDownloadRequestsToQueue(requests: List<CacheDownloadRequest>) {
if (requests.isEmpty()) return
val queuedRequests = mutableListOf<CacheDownloadRequest>()
val startRequests = mutableListOf<CacheDownloadRequest>()
synchronized(admissionQueue) {
// 快照当前已准入的书籍集合;同一批次内同一本书的后续请求直接准入,
// 绕过 maxActiveBookCount 限制。并发批次各自持有独立的快照,
// 因此准入上限是尽力而为的(best-effort),非严格保证。
val activeBookUrls = admittedBookUrls().toMutableSet()
requests.forEach { request ->
if (!admissionQueue.shouldQueue(request, activeBookUrls)) {
startRequests.add(request)
activeBookUrls.add(request.bookUrl)
return@forEach
}
admissionQueue.add(request)
queuedRequests.add(request)
}
}
if (queuedRequests.isNotEmpty()) {
CacheBook.addPendingAdmissions(queuedRequests)
ensureDownloadJob()
}
startRequests.forEach { request ->
submitDownloadRequest(request, fromAdmissionQueue = false)
}
}
val book = cacheBook.book
val chapterCount = appDb.bookChapterDao.getChapterCount(request.bookUrl)
private fun submitDownloadRequest(
request: CacheDownloadRequest,
fromAdmissionQueue: Boolean,
) {
val shouldStart = synchronized(admissionLock) {
val generation = admissionGenerations[request.bookUrl] ?: 0L
admissionBuffers.getOrPut(request.bookUrl) { ArrayDeque() }
.addLast(AdmissionRequest(request, fromAdmissionQueue, generation))
admittingBookUrls.add(request.bookUrl)
}
if (shouldStart) {
startAdmissionJob(request.bookUrl)
}
}
if (chapterCount == 0) {
cacheBook.setLoading()
mutex.withLock {
val name = book.name
if (book.tocUrl.isEmpty()) {
kotlin.runCatching {
WebBook.getBookInfoAwait(cacheBook.bookSource, book)
}.onFailure {
removeDownload(request.bookUrl)
CacheBook.markBookFailed(
request.bookUrl,
getString(R.string.error_get_book_info)
)
AppLog.put(
"$name》目录为空且加载详情页失败\n${it.localizedMessage}",
it,
true
)
return@execute
}
}
private fun startAdmissionJob(bookUrl: String) {
execute(executeContext = Dispatchers.IO) {
while (currentCoroutineContext().isActive) {
val admission = nextAdmissionRequest(bookUrl) ?: return@execute
processAdmissionRequest(admission)
}
}.onFinally {
finishAdmissionJob(bookUrl)
drainPendingDownloadRequests()
ensureDownloadJob()
}
}
WebBook.getChapterListAwait(cacheBook.bookSource, book).onFailure {
if (book.totalChapterNum > 0) {
book.totalChapterNum = 0
book.update()
}
private fun nextAdmissionRequest(bookUrl: String): AdmissionRequest? {
return synchronized(admissionLock) {
val buffer = admissionBuffers[bookUrl] ?: return@synchronized null
if (buffer.isEmpty()) return@synchronized null
buffer.removeFirst()
}
}
private suspend fun processAdmissionRequest(admission: AdmissionRequest) {
val request = admission.request
if (!admission.isCurrent()) return
val cacheBook = CacheBook.getOrCreate(request.bookUrl) ?: run {
markBookAdmissionFailed(request.bookUrl, getString(R.string.error_no_source))
return
}
if (!admission.isCurrent()) {
CacheBook.removeModelFromService(request.bookUrl, cacheBook)
return
}
val book = cacheBook.book
val chapterCount = appDb.bookChapterDao.getChapterCount(request.bookUrl)
if (chapterCount == 0) {
cacheBook.setLoading()
mutex.withLock {
val name = book.name
if (!admission.isCurrent()) {
CacheBook.removeModelFromService(request.bookUrl, cacheBook)
return
}
if (book.tocUrl.isEmpty()) {
kotlin.runCatching {
WebBook.getBookInfoAwait(cacheBook.bookSource, book)
}.onFailure {
removeDownload(request.bookUrl)
CacheBook.markBookFailed(
markBookAdmissionFailed(
request.bookUrl,
getString(R.string.error_get_chapter_list)
getString(R.string.error_get_book_info)
)
AppLog.put(
"$name》目录为空且加载目录失败\n${it.localizedMessage}",
"$name》目录为空且加载详情页失败\n${it.localizedMessage}",
it,
true
)
return@execute
}.getOrNull()?.let { toc ->
appDb.bookChapterDao.insert(*toc.toTypedArray())
return
}
}
book.update()
if (!admission.isCurrent()) {
CacheBook.removeModelFromService(request.bookUrl, cacheBook)
return
}
WebBook.getChapterListAwait(cacheBook.bookSource, book).onFailure {
if (book.totalChapterNum > 0) {
book.totalChapterNum = 0
book.update()
}
removeDownload(request.bookUrl)
markBookAdmissionFailed(
request.bookUrl,
getString(R.string.error_get_chapter_list)
)
AppLog.put(
"$name》目录为空且加载目录失败\n${it.localizedMessage}",
it,
true
)
return
}.getOrNull()?.let { toc ->
appDb.bookChapterDao.insert(*toc.toTypedArray())
}
book.update()
}
}
if (!admission.isCurrent()) {
CacheBook.removeModelFromService(request.bookUrl, cacheBook)
return
}
//添加章节到下载队列
cacheBook.addRequest(request)
if (admission.fromAdmissionQueue) {
CacheBook.removePendingAdmission(request)
}
notificationContent = CacheBook.downloadSummary
upCacheBookNotification()
}
private fun AdmissionRequest.isCurrent(): Boolean {
return synchronized(admissionLock) {
admissionGenerations[request.bookUrl].orZero() == generation
}
}
private fun admittedBookUrls(): Set<String> {
return CacheBook.cacheBookMap.keys.toHashSet().apply {
synchronized(admissionLock) {
addAll(admittingBookUrls)
}
}
}
private fun hasPendingDownloadRequests(): Boolean {
return synchronized(admissionQueue) {
!admissionQueue.isEmpty()
}
}
private fun hasAdmittingRequests(): Boolean {
return synchronized(admissionLock) {
admittingBookUrls.isNotEmpty()
}
}
private fun removeQueuedBook(bookUrl: String): Boolean {
return synchronized(admissionQueue) {
admissionQueue.removeBook(bookUrl)
}
}
private suspend fun removeBookCompletely(bookUrl: String): Boolean {
val removedQueued = removeQueuedBook(bookUrl)
val removedAdmission = cancelAdmission(bookUrl)
val removedActive = CacheBook.removeBookFromService(bookUrl)
waitAdmissionIdle(bookUrl)
return removedQueued || removedAdmission || removedActive
}
private fun markBookAdmissionFailed(bookUrl: String, message: String) {
removeQueuedBook(bookUrl)
CacheBook.markBookFailed(bookUrl, message)
}
private fun cancelAdmission(bookUrl: String): Boolean {
return synchronized(admissionLock) {
admissionGenerations[bookUrl] = admissionGenerations[bookUrl].orZero() + 1L
val removedBuffered = admissionBuffers.remove(bookUrl)?.isNotEmpty() == true
removedBuffered || bookUrl in admittingBookUrls
}
}
private suspend fun waitAdmissionIdle(bookUrl: String) {
val waiter = synchronized(admissionLock) {
if (bookUrl !in admittingBookUrls) {
null
} else {
CompletableDeferred<Unit>().also {
admissionIdleWaiters.getOrPut(bookUrl) { mutableListOf() }.add(it)
}
}
}
waiter?.await()
}
//添加章节到下载队列
cacheBook.addRequest(request)
private fun finishAdmissionJob(bookUrl: String) {
var restart = false
val waiters = synchronized(admissionLock) {
val buffer = admissionBuffers[bookUrl]
if (buffer != null && buffer.isNotEmpty()) {
restart = true
emptyList()
} else {
admissionBuffers.remove(bookUrl)
admittingBookUrls.remove(bookUrl)
admissionIdleWaiters.remove(bookUrl).orEmpty()
}
}
if (restart) {
startAdmissionJob(bookUrl)
return
}
waiters.forEach { it.complete(Unit) }
}
notificationContent = CacheBook.downloadSummary
upCacheBookNotification()
}.onFinally {
if (downloadJob == null) {
download()
private fun drainPendingDownloadRequests() {
while (true) {
val request = synchronized(admissionQueue) {
admissionQueue.pollReady(admittedBookUrls())
} ?: return
submitDownloadRequest(request, fromAdmissionQueue = true)
}
}
private fun ensureDownloadJob() {
synchronized(downloadJobLock) {
if (downloadJob?.isActive == true) return
downloadJob = lifecycleScope.launch(cachePool) {
runDownloadLoop()
}
}
}
@@ -201,19 +477,41 @@ class CacheBookService : BaseService() {
private fun removeDownload(bookUrl: String?) {
CacheBook.cacheBookMap[bookUrl]?.stop()
if (downloadJob == null && CacheBook.isRun) {
download()
if (CacheBook.isRun) {
ensureDownloadJob()
return
}
if (CacheBook.cacheBookMap.isEmpty()) {
stopSelf()
stopIfIdle()
}
private suspend fun runDownloadLoop() {
try {
while (currentCoroutineContext().isActive) {
drainPendingDownloadRequests()
if (CacheBook.isGloballyPaused) {
delay(200)
continue
}
if (!CacheBook.isRun) {
if (!hasPendingDownloadRequests() && !hasAdmittingRequests()) break
delay(200)
continue
}
CacheBook.startProcessJob(cachePool)
}
} finally {
val finishedJob = currentCoroutineContext()[Job]
synchronized(downloadJobLock) {
if (downloadJob == finishedJob) {
downloadJob = null
}
}
stopIfIdle()
}
}
private fun download() {
downloadJob?.cancel()
downloadJob = lifecycleScope.launch(cachePool) {
CacheBook.startProcessJob(cachePool)
private fun stopIfIdle() {
if (!CacheBook.isRun && !hasPendingDownloadRequests() && !hasAdmittingRequests()) {
stopSelf()
}
}
@@ -221,7 +519,12 @@ class CacheBookService : BaseService() {
private fun upCacheBookNotification() {
val total = CacheBook.totalCount
val progress = CacheBook.completedCount
val summary = CacheBook.downloadSummary
val pendingBookCount = synchronized(admissionQueue) { admissionQueue.size }
val summary = if (pendingBookCount > 0) {
"${CacheBook.downloadSummary} | 待入队:$pendingBookCount"
} else {
CacheBook.downloadSummary
}
notificationBuilder.apply {
setContentText(summary)
@@ -235,6 +538,41 @@ class CacheBookService : BaseService() {
notificationManager.notify(NotificationId.CacheBookService, notificationBuilder.build())
}
private fun logDownloadDiagnostics() {
val now = System.currentTimeMillis()
if (now - lastDiagnosticsLogTime < DIAGNOSTICS_LOG_INTERVAL_MILLIS) return
lastDiagnosticsLogTime = now
val pendingBookCount = synchronized(admissionQueue) { admissionQueue.size }
if (!CacheBook.isRun && pendingBookCount == 0 && !hasAdmittingRequests()) return
val diagnostics = CacheBook.diagnostics()
val runtime = Runtime.getRuntime()
val usedMemoryMb = (runtime.totalMemory() - runtime.freeMemory()) / MB
val totalMemoryMb = runtime.totalMemory() / MB
val maxMemoryMb = runtime.maxMemory() / MB
LogUtils.d("CacheBookDiagnostics") {
"activeBooks=${diagnostics.activeBookCount}, " +
"admittingBooks=${admittingBookCount()}, " +
"pendingBooks=$pendingBookCount, " +
"waitingChapters=${diagnostics.waitingChapterCount}, " +
"runningChapters=${diagnostics.runningChapterCount}, " +
"chapterTasks=${diagnostics.trackedChapterTaskCount}, " +
"loadingBooks=${diagnostics.loadingBookCount}, " +
"retryingBooks=${diagnostics.retryingBookCount}, " +
"configuredThreads=$threadCount, " +
"heap=${usedMemoryMb}MB/${totalMemoryMb}MB max=${maxMemoryMb}MB"
}
}
private fun admittingBookCount(): Int {
return synchronized(admissionLock) {
admittingBookUrls.size
}
}
private fun Long?.orZero(): Long = this ?: 0L
/**
* 更新通知
*/
@@ -96,7 +96,7 @@ private fun BookCacheManageScreen(
var pendingDeleteBook by remember { mutableStateOf<BookCacheBookItem?>(null) }
var pendingDeleteChapter by remember { mutableStateOf<Pair<BookCacheBookItem, BookCacheChapterItem>?>(null) }
val allBooks = state.shelfBooks + state.notShelfBooks
val hasRunningDownload = allBooks.any { it.isDownloading }
val hasRunningDownload = allBooks.any { it.hasActiveDownload }
val hasDownloadTarget = allBooks.any { it.cachedCount < it.totalCount }
AppScaffold(
@@ -351,22 +351,26 @@ private fun BookCacheBookCard(
verticalAlignment = Alignment.CenterVertically
) {
AppText(
text = "下载中 ${item.downloadingCount} · 等待 ${item.waitingCount} · 失败 ${item.errorCount}",
text = "下载中 ${item.downloadingCount} · 等待 ${item.waitingCount} · 暂停 ${item.pausedCount} · 失败 ${item.errorCount}",
modifier = Modifier.weight(1f),
style = LegadoTheme.typography.labelMediumEmphasized,
color = LegadoTheme.colorScheme.onSurfaceVariant
)
if (item.isDownloading || item.cachedCount < item.totalCount) {
if (item.hasDownloadTask || item.cachedCount < item.totalCount) {
SmallTonalIconButton(
onClick = {
if (item.isDownloading) {
if (item.hasActiveDownload) {
onIntent(BookCacheManageIntent.StopBookDownload(item.bookUrl))
} else {
onIntent(BookCacheManageIntent.StartBookDownload(item.bookUrl))
}
},
imageVector = if (item.isDownloading) Icons.Default.Stop else Icons.Default.PlayArrow,
contentDescription = if (item.isDownloading) "暂停本书下载" else "开始本书下载"
imageVector = if (item.hasActiveDownload) Icons.Default.Stop else Icons.Default.PlayArrow,
contentDescription = when {
item.hasActiveDownload -> "暂停本书下载"
item.isPaused -> "继续本书下载"
else -> "开始本书下载"
}
)
}
SmallTonalIconButton(
@@ -423,11 +427,11 @@ private fun BookCacheChapterRow(
imageVector = Icons.Default.Stop,
contentDescription = "暂停章节下载"
)
} else if (!item.isCached) {
} else if (item.isPaused || !item.isCached) {
SmallTonalIconButton(
onClick = onDownload,
imageVector = Icons.Default.Download,
contentDescription = "下载章节"
contentDescription = if (item.isPaused) "继续章节下载" else "下载章节"
)
}
SmallTonalIconButton(
@@ -442,6 +446,7 @@ private fun chapterStatusText(item: BookCacheChapterItem): String {
return when {
item.isDownloading -> "下载中"
item.isWaiting -> "等待下载"
item.isPaused -> "已暂停"
item.isError -> "下载失败"
item.isCached -> "已缓存"
else -> "未缓存"
@@ -38,6 +38,7 @@ data class BookCacheManageUiState(
val expandedBookUrls: Set<String> = emptySet(),
val chaptersByBookUrl: Map<String, List<BookCacheChapterItem>> = emptyMap(),
val downloadSummary: String = "",
val hasPausedDownloads: Boolean = false,
val version: Long = 0,
)
@@ -50,11 +51,15 @@ data class BookCacheBookItem(
val cachedFileCount: Int,
val waitingCount: Int,
val downloadingCount: Int,
val pausedCount: Int,
val errorCount: Int,
val isNotShelf: Boolean,
) {
val progress: Float get() = if (totalCount == 0) 0f else cachedCount.toFloat() / totalCount
val isDownloading: Boolean get() = waitingCount > 0 || downloadingCount > 0
val hasActiveDownload: Boolean get() = waitingCount > 0 || downloadingCount > 0
val isPaused: Boolean get() = pausedCount > 0
val hasDownloadTask: Boolean get() = hasActiveDownload || isPaused
val isDownloading: Boolean get() = hasActiveDownload
}
data class BookCacheChapterItem(
@@ -64,6 +69,7 @@ data class BookCacheChapterItem(
val isCached: Boolean,
val isWaiting: Boolean,
val isDownloading: Boolean,
val isPaused: Boolean,
val isError: Boolean,
)
@@ -161,6 +167,12 @@ class BookCacheManageViewModel(
pendingDownloadSummaryRefresh = true
}
}
viewModelScope.launch {
CacheBook.pendingAdmissionFlow.collect { pending ->
pending.keys.forEach { scheduleDownloadStatusRefresh(it) }
pendingDownloadSummaryRefresh = true
}
}
viewModelScope.launch {
CacheBook.queueChangedFlow.collect { bookUrl ->
scheduleDownloadStatusRefresh(bookUrl)
@@ -212,6 +224,7 @@ class BookCacheManageViewModel(
expandedBookUrls = result.expandedBookUrls,
chaptersByBookUrl = result.chaptersByBookUrl,
downloadSummary = buildDownloadSummary(result.items),
hasPausedDownloads = CacheBook.hasPausedDownloads,
version = state.version + 1,
)
}
@@ -237,6 +250,7 @@ class BookCacheManageViewModel(
_uiState.update {
it.copy(
downloadSummary = buildDownloadSummary(it.shelfBooks + it.notShelfBooks),
hasPausedDownloads = CacheBook.hasPausedDownloads,
version = it.version + 1,
)
}
@@ -295,6 +309,7 @@ class BookCacheManageViewModel(
expandedBookUrls = expandedBookUrls,
chaptersByBookUrl = chaptersByBookUrl,
downloadSummary = buildDownloadSummary(sortedBooks),
hasPausedDownloads = CacheBook.hasPausedDownloads,
version = state.version + 1,
)
}
@@ -302,13 +317,25 @@ class BookCacheManageViewModel(
private fun buildBookItem(book: Book): BookCacheBookItem? {
val cacheFiles = BookHelp.getChapterFiles(book)
val bookState = CacheBook.downloadStateFlow.value.books[book.bookUrl]
val model = CacheBook.cacheBookMap[book.bookUrl]
val (waitingCount, downloadingCount) = model?.queueCounts() ?: (0 to 0)
val rawWaitingCount = bookState?.waitingCount.orZero() +
CacheBook.pendingAdmissionFlow.value[book.bookUrl].orZero()
val rawDownloadingCount = bookState?.runningIndices?.size.orZero()
val isBookPaused = model?.isPaused() == true ||
(CacheBook.hasPausedDownloads && CacheBook.pendingAdmissionFlow.value.containsKey(book.bookUrl))
val pausedCount = if (isBookPaused) {
rawWaitingCount + rawDownloadingCount + bookState?.pausedIndices?.size.orZero()
} else {
bookState?.pausedIndices?.size.orZero()
}
val waitingCount = if (isBookPaused) 0 else rawWaitingCount
val downloadingCount = if (isBookPaused) 0 else rawDownloadingCount
val errorIndices = errorIndices(book.bookUrl)
val totalCount = bookChapterDao.getChapterCount(book.bookUrl)
val cachedFileCount = cacheFiles.count { it.endsWith(".nb") }
val cachedCount = min(cachedFileCount + bookChapterDao.getVolumeCount(book.bookUrl), totalCount)
if (totalCount == 0 && cacheFiles.isEmpty() && model == null && !book.isNotShelf) {
if (totalCount == 0 && cacheFiles.isEmpty() && waitingCount == 0 && downloadingCount == 0 && pausedCount == 0 && !book.isNotShelf) {
return null
}
return BookCacheBookItem(
@@ -320,13 +347,14 @@ class BookCacheManageViewModel(
cachedFileCount = cachedFileCount,
waitingCount = waitingCount,
downloadingCount = downloadingCount,
pausedCount = pausedCount,
errorCount = errorIndices.size,
isNotShelf = book.isNotShelf,
)
}
private fun shouldShowItem(item: BookCacheBookItem): Boolean {
return item.cachedFileCount > 0 || item.isDownloading || item.errorCount > 0
return item.cachedFileCount > 0 || item.hasDownloadTask || item.errorCount > 0
}
private fun buildChapterItems(bookUrl: String): List<BookCacheChapterItem> {
@@ -336,13 +364,15 @@ class BookCacheManageViewModel(
val model = CacheBook.cacheBookMap[bookUrl]
val errorIndices = errorIndices(bookUrl)
return chapters.map { chapter ->
val isPaused = model?.isPaused(chapter.index) == true
BookCacheChapterItem(
chapterUrl = chapter.url,
title = chapter.title,
index = chapter.index,
isCached = cacheFiles.contains(chapter.getFileName()) || chapter.isVolume,
isWaiting = model?.isWaiting(chapter.index) == true,
isDownloading = model?.isDownloading(chapter.index) == true,
isWaiting = !isPaused && model?.isWaiting(chapter.index) == true,
isDownloading = !isPaused && model?.isDownloading(chapter.index) == true,
isPaused = isPaused,
isError = errorIndices.contains(chapter.index),
)
}
@@ -352,6 +382,8 @@ class BookCacheManageViewModel(
return CacheBook.errorIndices(bookUrl)
}
private fun Int?.orZero(): Int = this ?: 0
private fun toggleBookExpanded(bookUrl: String) {
val shouldExpand = !_uiState.value.expandedBookUrls.contains(bookUrl)
_uiState.update { state ->
@@ -392,18 +424,27 @@ class BookCacheManageViewModel(
}
private fun stopAllDownloads() {
CacheBook.stop(context)
reloadAll(forceDatabase = true)
execute {
CacheBook.pause(context)
}.onFinally {
reloadAll(forceDatabase = true)
}
}
private fun stopBookDownload(bookUrl: String) {
CacheBook.removeBook(bookUrl)
scheduleBookReload(bookUrl, debounceMillis = 0)
execute {
CacheBook.pauseBook(context, bookUrl)
}.onFinally {
scheduleBookReload(bookUrl, debounceMillis = 0)
}
}
private fun startAllDownloads() {
val items = uiState.value.shelfBooks + uiState.value.notShelfBooks
execute {
if (CacheBook.resume(context)) {
return@execute null
}
var count = 0
items.forEach { item ->
downloadableChapterIndexBatches(item.bookUrl).forEach { chapterIndices ->
@@ -412,7 +453,8 @@ class BookCacheManageViewModel(
currentCoroutineContext().ensureActive()
}
count
}.onSuccess { count ->
}.onSuccess { countOrNull ->
val count = countOrNull ?: return@onSuccess
if (count > 0) {
_effects.tryEmit(BookCacheManageEffect.ShowMessage("已加入缓存队列: $count"))
} else {
@@ -427,12 +469,16 @@ class BookCacheManageViewModel(
private fun startBookDownload(bookUrl: String) {
execute {
if (CacheBook.resumeBook(context, bookUrl)) {
return@execute null
}
var count = 0
downloadableChapterIndexBatches(bookUrl).forEach { chapterIndices ->
count += cacheBookChaptersUseCase.execute(bookUrl, chapterIndices)
}
count
}.onSuccess { count ->
}.onSuccess { countOrNull ->
val count = countOrNull ?: return@onSuccess
if (count > 0) {
_effects.tryEmit(BookCacheManageEffect.ShowMessage("已加入缓存队列: $count"))
} else {
@@ -457,6 +503,7 @@ class BookCacheManageViewModel(
if (
chapter.isVolume ||
cacheFiles.contains(chapter.getFileName()) ||
model?.isPaused(chapter.index) == true ||
model?.isWaiting(chapter.index) == true ||
model?.isDownloading(chapter.index) == true
) {
@@ -474,8 +521,8 @@ class BookCacheManageViewModel(
}
private fun deleteBookCache(bookUrl: String) {
CacheBook.removeBook(bookUrl)
execute {
CacheBook.removeAwait(context, bookUrl)
clearBookCacheUseCase.execute(bookUrl)
}.onSuccess {
_effects.tryEmit(BookCacheManageEffect.ShowMessage("缓存已删除"))
@@ -488,8 +535,13 @@ class BookCacheManageViewModel(
private fun downloadChapter(bookUrl: String, chapterIndex: Int) {
execute {
if (CacheBook.resumeChapter(context, bookUrl, chapterIndex)) {
return@execute false
}
cacheBookChaptersUseCase.execute(bookUrl, listOf(chapterIndex))
}.onSuccess {
true
}.onSuccess { enqueued ->
if (!enqueued) return@onSuccess
_effects.tryEmit(BookCacheManageEffect.ShowMessage("章节已加入缓存队列"))
}.onError {
_effects.tryEmit(BookCacheManageEffect.ShowMessage("章节缓存失败\n${it.localizedMessage}"))
@@ -499,7 +551,9 @@ class BookCacheManageViewModel(
}
private fun stopChapterDownload(bookUrl: String, chapterIndex: Int) {
if (CacheBook.removeChapter(bookUrl, chapterIndex)) {
execute {
CacheBook.pauseChapter(bookUrl, chapterIndex)
}.onSuccess {
scheduleBookReload(bookUrl, debounceMillis = 0)
}
}
@@ -510,16 +564,18 @@ class BookCacheManageViewModel(
chapterTitle: String,
chapterIndex: Int,
) {
val book = bookDao.getBook(bookUrl) ?: return
val chapter = BookChapter(
url = chapterUrl,
title = chapterTitle,
bookUrl = bookUrl,
index = chapterIndex,
)
execute {
val book = bookDao.getBook(bookUrl) ?: return@execute false
val chapter = BookChapter(
url = chapterUrl,
title = chapterTitle,
bookUrl = bookUrl,
index = chapterIndex,
)
BookHelp.delContent(book, chapter)
}.onSuccess {
true
}.onSuccess { deleted ->
if (!deleted) return@onSuccess
_effects.tryEmit(BookCacheManageEffect.ShowMessage("章节缓存已删除"))
}.onError {
_effects.tryEmit(BookCacheManageEffect.ShowMessage("删除章节缓存失败\n${it.localizedMessage}"))
@@ -538,9 +594,10 @@ class BookCacheManageViewModel(
if (items.isEmpty()) return ""
val downloadingCount = items.sumOf { it.downloadingCount }
val waitingCount = items.sumOf { it.waitingCount }
val pausedCount = items.sumOf { it.pausedCount }
val errorCount = items.sumOf { it.errorCount }
val cachedCount = items.sumOf { it.cachedCount }
return "正在下载:$downloadingCount | 等待:$waitingCount | 失败:$errorCount | 成功:$cachedCount"
return "下载:$downloadingCount | 等待:$waitingCount | 暂停:$pausedCount | 失败:$errorCount | 已缓存:$cachedCount"
}
private data class LoadedCacheState(
@@ -304,7 +304,9 @@ class BookshelfManageScreenViewModel(
fun getCacheCount(bookUrl: String): Int? = cacheCounts[bookUrl]
fun isBookDownloading(bookUrl: String): Boolean {
return CacheBook.cacheBookMap[bookUrl]?.isStop() == false
if (CacheBook.pendingAdmissionFlow.value[bookUrl].orZero() > 0) return true
val bookState = CacheBook.downloadStateFlow.value.books[bookUrl] ?: return false
return bookState.waitingCount > 0 || bookState.runningIndices.isNotEmpty()
}
fun getDownloadFailureMessage(bookUrl: String): String? {
@@ -409,13 +411,23 @@ class BookshelfManageScreenViewModel(
pendingDownloadBookUrls.removeAll(downloadState.books.keys)
successfulBookUrls.forEach { downloadFailureMessages.remove(it) }
downloadFailureMessages.putAll(failureMsgs)
_uiState.update { it.copy(isDownloadRunning = downloadState.isRunning) }
_uiState.update {
it.copy(isDownloadRunning = downloadState.isRunning || CacheBook.pendingAdmissionFlow.value.isNotEmpty())
}
downloadState.books.keys.forEach { bookUrl ->
scheduleDownloadStatusRefresh(bookUrl)
}
scheduleDownloadStatusRefresh()
}
}
viewModelScope.launch {
CacheBook.pendingAdmissionFlow.collect { pending ->
pending.keys.forEach { bookUrl ->
scheduleDownloadStatusRefresh(bookUrl)
}
scheduleDownloadStatusRefresh()
}
}
viewModelScope.launch {
CacheBook.queueChangedFlow.collect { bookUrl ->
scheduleDownloadStatusRefresh(bookUrl)
@@ -454,7 +466,9 @@ class BookshelfManageScreenViewModel(
}
private fun syncDownloadRunning() {
_uiState.update { it.copy(isDownloadRunning = CacheBook.isRun) }
_uiState.update {
it.copy(isDownloadRunning = CacheBook.isRun || CacheBook.pendingAdmissionFlow.value.isNotEmpty())
}
}
private fun refreshGroupName(groupId: Long) {
@@ -541,6 +555,8 @@ class BookshelfManageScreenViewModel(
return min(cachedFileCount + bookChapterDao.getVolumeCount(book.bookUrl), totalCount)
}
private fun Int?.orZero(): Int = this ?: 0
private fun startDownloadForVisibleBooks(books: List<Book>, downloadAllChapters: Boolean) {
val bookUrls = books.mapTo(hashSetOf()) { it.bookUrl }
bookUrls.forEach { downloadFailureMessages.remove(it) }