diff options
Diffstat (limited to 'app/src/main/java/org/terst/feedreader/work')
| -rw-r--r-- | app/src/main/java/org/terst/feedreader/work/FeedFetchWorker.kt | 136 |
1 files changed, 110 insertions, 26 deletions
diff --git a/app/src/main/java/org/terst/feedreader/work/FeedFetchWorker.kt b/app/src/main/java/org/terst/feedreader/work/FeedFetchWorker.kt index 5ec6bab..9c2962f 100644 --- a/app/src/main/java/org/terst/feedreader/work/FeedFetchWorker.kt +++ b/app/src/main/java/org/terst/feedreader/work/FeedFetchWorker.kt @@ -7,13 +7,24 @@ import androidx.work.NetworkType import androidx.work.PeriodicWorkRequestBuilder import androidx.work.WorkManager import androidx.work.WorkerParameters +import androidx.work.workDataOf +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.async +import kotlinx.coroutines.awaitAll +import kotlinx.coroutines.coroutineScope +import kotlinx.coroutines.sync.Mutex +import kotlinx.coroutines.sync.Semaphore +import kotlinx.coroutines.sync.withPermit import org.terst.feedreader.data.AppDatabase import org.terst.feedreader.data.ArticleEntity import org.terst.feedreader.data.FeedEntity import org.terst.feedreader.data.FeedParser import org.terst.feedreader.data.Net +import org.terst.feedreader.data.Settings +import org.terst.feedreader.data.await import okhttp3.Request import java.security.MessageDigest +import java.util.concurrent.atomic.AtomicInteger import java.util.concurrent.TimeUnit /** @@ -27,33 +38,80 @@ class FeedFetchWorker(context: Context, params: WorkerParameters) : CoroutineWor private val client = Net.client override suspend fun doWork(): Result { - val db = AppDatabase.get(applicationContext) - val feeds = db.feedDao().getAll() + // WorkManager's "unique work name" dedup does NOT prevent this: schedulePeriodic + // and fetchNow use two different unique-work names (periodic and one-time requests + // can't share one), so an hourly periodic run and a pull-to-refresh/morning-alarm + // fetch firing at the same moment both start doWork() concurrently -- every feed + // gets fetched twice, and both instances race writes to the same feed rows' + // etag/lastModified/lastFetchedAt. This mutex is process-local, which is exactly + // right here: WorkManager runs this worker in the app's own process, never a + // separate one, so a single in-memory lock covers every real overlap case. + if (!fetchMutex.tryLock()) { + return Result.success() + } + try { + val db = AppDatabase.get(applicationContext) + val feeds = db.feedDao().getAll() + val completedCount = AtomicInteger(0) - for (feed in feeds) { - fetchOne(feed, db) + // Fetched with bounded concurrency, not one feed at a time: with a large + // subscription list (this app has seen 100+ feeds in the wild -- see + // AppDatabase's v9 migration comments), a strictly sequential loop means one + // slow/stalling server delays every feed queued behind it, and the whole run + // can take tens of seconds to minutes. Dispatchers.IO (not the CoroutineWorker + // default of Dispatchers.Default) because this is blocking network I/O, not + // CPU-bound work -- running it on Default would compete with the app's other + // CPU-bound coroutine work for that dispatcher's small, core-sized thread pool. + val newArticleCount = coroutineScope { + val semaphore = Semaphore(FETCH_CONCURRENCY) + feeds.map { feed -> + async(Dispatchers.IO) { + semaphore.withPermit { + val count = fetchOne(feed, db) + // KEY_CURRENT now means "how many feeds have finished," not + // "which one's currently being checked" -- there's no single + // "current" feed anymore under concurrent fetching, but a + // monotonically-advancing completed-count is just as good a + // progress signal for the UI. + setProgress(workDataOf(KEY_CURRENT to completedCount.incrementAndGet(), KEY_TOTAL to feeds.size)) + count + } + } + }.awaitAll().sum() + } + // Durable, not WorkInfo-state-based -- see Settings.recordFetchResult's doc + // comment for why periodic work's own state can't be used for this. + Settings(applicationContext).recordFetchResult(newArticleCount) + return Result.success(workDataOf(KEY_TOTAL to feeds.size, KEY_NEW_ARTICLES to newArticleCount)) + } finally { + fetchMutex.unlock() } - return Result.success() } - private suspend fun fetchOne(feed: FeedEntity, db: AppDatabase) { + /** Returns how many articles from this feed were genuinely new (not already stored). */ + private suspend fun fetchOne(feed: FeedEntity, db: AppDatabase): Int { val requestBuilder = Request.Builder().url(feed.id) feed.etag?.let { requestBuilder.header("If-None-Match", it) } feed.lastModified?.let { requestBuilder.header("If-Modified-Since", it) } - try { - client.newCall(requestBuilder.build()).execute().use { response -> + return try { + client.newCall(requestBuilder.build()).await().use { response -> val now = System.currentTimeMillis() if (response.code == 304) { - db.feedDao().update(feed.copy(lastFetchedAt = now, lastError = null)) - return + db.feedDao().updateFetchResult( + feed.id, feed.title, feed.siteUrl, now, feed.etag, feed.lastModified, lastError = null + ) + return@use 0 } if (!response.isSuccessful) { - db.feedDao().update(feed.copy(lastFetchedAt = now, lastError = "HTTP ${response.code}")) - return + db.feedDao().updateFetchResult( + feed.id, feed.title, feed.siteUrl, now, feed.etag, feed.lastModified, + lastError = "HTTP ${response.code}" + ) + return@use 0 } - val body = response.body ?: return + val body = response.body ?: return@use 0 val parsed = FeedParser.parse(body.byteStream()) val articles = parsed.items.map { item -> @@ -70,21 +128,28 @@ class FeedFetchWorker(context: Context, params: WorkerParameters) : CoroutineWor createdAt = now ) } - db.articleDao().insertNew(articles) - - db.feedDao().update( - feed.copy( - title = parsed.title.ifBlank { feed.title }, - siteUrl = parsed.siteUrl ?: feed.siteUrl, - lastFetchedAt = now, - etag = response.header("ETag") ?: feed.etag, - lastModified = response.header("Last-Modified") ?: feed.lastModified, - lastError = null - ) + // Room returns -1 for a row OnConflictStrategy.IGNORE actually skipped + // (already stored), and the real inserted rowid (>= 0) for a genuinely new + // one -- count only the latter so "N new articles" reflects reality. + val newCount = db.articleDao().insertNew(articles).count { it != -1L } + + db.feedDao().updateFetchResult( + id = feed.id, + title = parsed.title.ifBlank { feed.title }, + siteUrl = parsed.siteUrl ?: feed.siteUrl, + lastFetchedAt = now, + etag = response.header("ETag") ?: feed.etag, + lastModified = response.header("Last-Modified") ?: feed.lastModified, + lastError = null ) + newCount } } catch (e: Exception) { - db.feedDao().update(feed.copy(lastFetchedAt = System.currentTimeMillis(), lastError = e.message ?: "Fetch failed")) + db.feedDao().updateFetchResult( + feed.id, feed.title, feed.siteUrl, System.currentTimeMillis(), feed.etag, feed.lastModified, + lastError = e.message ?: "Fetch failed" + ) + 0 } } @@ -96,6 +161,22 @@ class FeedFetchWorker(context: Context, params: WorkerParameters) : CoroutineWor companion object { private const val UNIQUE_WORK_NAME = "feed_fetch_periodic" + // Common tag on every enqueue (periodic AND one-off) so ArticleListScreen can + // observe "is a fetch happening right now" via a single getWorkInfosByTagFlow + // query, regardless of which of the two triggered it. + const val FETCH_WORK_TAG = "feed_fetch" + const val KEY_CURRENT = "current" + const val KEY_TOTAL = "total" + const val KEY_NEW_ARTICLES = "newArticles" + + // Bounded, not unbounded (Dispatchers.IO can grow to 64+ threads on demand) -- + // enough to meaningfully overlap slow feeds without opening 100+ simultaneous + // connections to 100+ different hosts at once on a fetch run with a large + // subscription list. + private const val FETCH_CONCURRENCY = 8 + + private val fetchMutex = Mutex() + fun schedulePeriodic(context: Context) { val request = PeriodicWorkRequestBuilder<FeedFetchWorker>(60, TimeUnit.MINUTES) .setConstraints( @@ -103,6 +184,7 @@ class FeedFetchWorker(context: Context, params: WorkerParameters) : CoroutineWor .setRequiredNetworkType(NetworkType.CONNECTED) .build() ) + .addTag(FETCH_WORK_TAG) .build() WorkManager.getInstance(context).enqueueUniquePeriodicWork( UNIQUE_WORK_NAME, @@ -113,7 +195,9 @@ class FeedFetchWorker(context: Context, params: WorkerParameters) : CoroutineWor /** One-off immediate fetch, e.g. a manual pull-to-refresh. */ fun fetchNow(context: Context) { - val request = androidx.work.OneTimeWorkRequestBuilder<FeedFetchWorker>().build() + val request = androidx.work.OneTimeWorkRequestBuilder<FeedFetchWorker>() + .addTag(FETCH_WORK_TAG) + .build() WorkManager.getInstance(context).enqueueUniqueWork( "feed_fetch_now", androidx.work.ExistingWorkPolicy.KEEP, |