1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
|
package org.terst.feedreader.work
import android.content.Context
import androidx.work.CoroutineWorker
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 org.terst.feedreader.readability.ReaderExtractor
import okhttp3.Request
import java.security.MessageDigest
import java.util.concurrent.atomic.AtomicInteger
/**
* Fetches every subscribed feed, parses new items, and inserts them (existing
* articles are left untouched -- see ArticleDao.insertNew's IGNORE conflict
* strategy). Uses conditional GET (ETag / Last-Modified) so an unchanged feed
* costs a 304 instead of a full re-download and re-parse. Only ever triggered by
* fetchNow -- a fresh app launch or an explicit tap on the refresh button, never
* a background schedule.
*
* For a reader-mode feed, new articles are also run through ReaderExtractor right
* here, so extractedContent is already cached by the time the user opens one --
* see fetchOne.
*/
class FeedFetchWorker(context: Context, params: WorkerParameters) : CoroutineWorker(context, params) {
private val client = Net.client
override suspend fun doWork(): Result {
// fetchNow uses ExistingWorkPolicy.KEEP against a single unique work name, so
// WorkManager itself already dedups a rapid double-tap of the refresh button or an
// overlapping cold-launch fetch. This mutex covers the remaining case: a manual
// fetchNow firing while doWork() from a still-running instance hasn't finished yet
// (e.g. a very slow feed run overlapping the next launch). 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)
// 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()
}
}
/** 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) }
return try {
client.newCall(requestBuilder.build()).await().use { response ->
val now = System.currentTimeMillis()
if (response.code == 304) {
db.feedDao().updateFetchResult(
feed.id, feed.title, feed.siteUrl, now, feed.etag, feed.lastModified, lastError = null
)
return@use 0
}
if (!response.isSuccessful) {
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@use 0
val parsed = FeedParser.parse(body.byteStream())
val articles = parsed.items.map { item ->
val id = item.guid ?: sha256(item.link)
ArticleEntity(
id = id,
feedId = feed.id,
title = item.title,
link = item.link,
author = item.author,
publishedAt = item.publishedAt,
feedContentHtml = item.contentHtml,
imageUrl = item.imageUrl,
createdAt = now
)
}
// 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 insertResults = db.articleDao().insertNew(articles)
val newArticles = articles.zip(insertResults).mapNotNull { (article, result) -> article.takeIf { result != -1L } }
val newCount = newArticles.size
if (feed.readerModeEnabled) {
// Pre-extract reader-mode content right here during the fetch, instead of
// lazily the first time the article is opened (see ReaderScreen's
// LaunchedEffect). For a reader-mode feed like West Hawaii Today, that
// lazy path meant every fresh article's first open sat on a spinner for a
// hidden WebView load + Readability JS eval (several seconds, sometimes
// much longer on a slow/heavy page) before any content showed. By the time
// the user actually taps into it here, it's already cached.
for (newArticle in newArticles) {
val extracted = ReaderExtractor.extract(applicationContext, newArticle.link)
extracted?.content?.let { db.articleDao().setExtractedContent(newArticle.id, it) }
}
}
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().updateFetchResult(
feed.id, feed.title, feed.siteUrl, System.currentTimeMillis(), feed.etag, feed.lastModified,
lastError = e.message ?: "Fetch failed"
)
0
}
}
private fun sha256(input: String): String {
val digest = MessageDigest.getInstance("SHA-256").digest(input.toByteArray())
return digest.joinToString("") { "%02x".format(it) }
}
companion object {
// Tagged so ArticleListScreen can observe "is a fetch happening right now" via a
// single getWorkInfosByTagFlow query.
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()
/** One-off immediate fetch: a fresh app launch or an explicit refresh-button tap. */
fun fetchNow(context: Context) {
val request = androidx.work.OneTimeWorkRequestBuilder<FeedFetchWorker>()
.addTag(FETCH_WORK_TAG)
.build()
WorkManager.getInstance(context).enqueueUniqueWork(
"feed_fetch_now",
androidx.work.ExistingWorkPolicy.KEEP,
request
)
}
}
}
|
