refactor: move subscription update logic to SubscriptionUpdateService

- Refactor subscription synchronization and testing from static object to a dedicated foreground service.
- Implement SubscriptionUpdateService with accurate lifecycle management using AtomicInteger (runningTasks).
- Enforce serial execution of subscription updates via Semaphore(1) to ensure data integrity and bandwidth control.
- Fix NetworkOnMainThreadException by switching service execution to Dispatchers.IO.
- Decouple update logic from Main UI components for better modularity.
- Prevent rapid rescheduling loops by updating 'lastUpdated' timestamp at the start of the task.
- Simplify messaging by supporting multiple subscription IDs in a single SubscriptionUpdateMessage.
This commit is contained in:
2dust committed 2026-07-23 14:53:13 +08:00
1 parent f7f170e8d3
commit a231151e70
7 files changed
+307 -153

No files matched your search

+10
View File
@@ -253,6 +253,16 @@
android:value="test" />
</service>
<service
android:name=".service.SubscriptionUpdateService"
android:exported="false"
android:foregroundServiceType="specialUse"
android:process=":RunSoLibV2RayDaemon">
<property
android:name="android.app.PROPERTY_SPECIAL_USE_FGS_SUBTYPE"
android:value="test" />
</service>
<receiver
android:name=".receiver.WidgetProvider"
android:exported="true"
@@ -184,6 +184,9 @@ object AppConfig {
const val MSG_MEASURE_CONFIG_NOTIFY = 73
const val MSG_MEASURE_CONFIG_FINISH = 74
const val MSG_SUB_UPDATE_START = 8
const val MSG_SUB_UPDATE_CANCEL = 81
/** Notification channel IDs and names. */
const val RAY_NG_CHANNEL_ID = "RAY_NG_M_CH_ID"
const val RAY_NG_CHANNEL_NAME = "v2rayNG Background Service"
@@ -0,0 +1,8 @@
package com.v2ray.ang.dto
import java.io.Serializable
data class SubscriptionUpdateMessage(
val key: Int,
val subIds: List<String> = listOf()
) : Serializable
@@ -5,26 +5,16 @@ import android.content.Context
import androidx.work.Constraints
import androidx.work.CoroutineWorker
import androidx.work.ExistingPeriodicWorkPolicy
import androidx.work.ExistingWorkPolicy
import androidx.work.NetworkType
import androidx.work.OneTimeWorkRequestBuilder
import androidx.work.PeriodicWorkRequestBuilder
import androidx.work.WorkerParameters
import androidx.work.multiprocess.RemoteWorkManager
import androidx.work.workDataOf
import com.v2ray.ang.AngApplication
import com.v2ray.ang.AppConfig
import com.v2ray.ang.R
import com.v2ray.ang.core.CoreNativeManager
import com.v2ray.ang.dto.RealPingEvent
import com.v2ray.ang.dto.entities.SubscriptionCache
import com.v2ray.ang.enums.NotificationChannelType
import com.v2ray.ang.service.RealPingWorkerService
import com.v2ray.ang.dto.SubscriptionUpdateMessage
import com.v2ray.ang.util.LogUtil
import com.v2ray.ang.util.NotificationHelper
import kotlinx.coroutines.CompletableDeferred
import kotlinx.coroutines.sync.Semaphore
import kotlinx.coroutines.sync.withPermit
import com.v2ray.ang.util.MessageUtil
import java.util.concurrent.TimeUnit
object SubscriptionUpdater {
@@ -52,7 +42,9 @@ object SubscriptionUpdater {
ExistingPeriodicWorkPolicy.KEEP
}
MmkvManager.decodeSubscriptions().forEach { sub ->
MmkvManager.decodeSubscriptions()
.filter { it.subscription.autoUpdate && it.subscription.url.isNotEmpty() }
.forEach { sub ->
scheduleOne(
context = context,
subId = sub.guid,
@@ -87,47 +79,20 @@ object SubscriptionUpdater {
}
/**
* Update all subscriptions immediately in the background.
* Call from: SubscriptionsViewModel.updateSubscriptions().
* Update the last updated timestamp and reschedule the task.
* This is used to reset the periodic timer and prevent rapid rescheduling loops.
*/
fun updateAllByManual(context: Context = AngApplication.application) {
LogUtil.i(AppConfig.TAG, "SubscriptionUpdater: manual updateAll triggered")
val rw = RemoteWorkManager.getInstance(context)
MmkvManager.decodeSubscriptions().forEach { sub ->
if (!sub.subscription.enabled || sub.subscription.url.isEmpty()) {
return@forEach
}
val subId = sub.guid
val request = OneTimeWorkRequestBuilder<UpdateTask>()
.setConstraints(
Constraints.Builder()
.setRequiredNetworkType(NetworkType.CONNECTED)
.build()
)
.setInputData(
workDataOf(
KEY_SUB_ID to subId,
)
)
.addTag(AppConfig.SUBSCRIPTION_UPDATE_TASK_NAME)
.build()
rw.enqueueUniqueWork(
"${taskName(subId)}_manual",
ExistingWorkPolicy.KEEP,
request
)
LogUtil.d(AppConfig.TAG, "SubscriptionUpdater: enqueued manual task for ${sub.subscription.remarks} ($subId)")
}
fun updateLastUpdatedAndReschedule(context: Context = AngApplication.application, subId: String) {
val subItem = MmkvManager.decodeSubscription(subId) ?: return
subItem.lastUpdated = System.currentTimeMillis()
MmkvManager.encodeSubscription(subId, subItem)
syncOne(context, subId)
}
// -------------------------------------------------------------------------
// Internal scheduling logic
// -------------------------------------------------------------------------
private val updateSemaphore = Semaphore(2)
private fun taskName(subId: String) = "${AppConfig.SUBSCRIPTION_UPDATE_TASK_NAME}_$subId"
private fun scheduleOne(
@@ -143,6 +108,11 @@ object SubscriptionUpdater {
return
}
if (subItem.url.isEmpty()) {
LogUtil.i(AppConfig.TAG, "SubscriptionUpdater: url isEmpty for ${subItem.remarks}, skip")
return
}
val intervalMinutes = maxOf(
AppConfig.SUBSCRIPTION_MIN_INTERVAL_MINUTES,
subItem.updateInterval
@@ -152,12 +122,17 @@ object SubscriptionUpdater {
val lastUpdated = subItem.lastUpdated
val intervalMillis = intervalMinutes * 60 * 1000L
val now = System.currentTimeMillis()
val initialDelayMillis = if (lastUpdated <= 0L) {
var initialDelayMillis = if (lastUpdated <= 0L) {
0L
} else {
maxOf(0L, lastUpdated + intervalMillis - now)
}
// Add a small floor to initial delay to prevent rapid rescheduling loops.
if (existingWorkPolicy == ExistingPeriodicWorkPolicy.REPLACE && initialDelayMillis < 5000L) {
initialDelayMillis = 5000L
}
val request = PeriodicWorkRequestBuilder<UpdateTask>(intervalMinutes, TimeUnit.MINUTES)
.setConstraints(
Constraints.Builder()
@@ -192,121 +167,23 @@ object SubscriptionUpdater {
CoroutineWorker(context, params) {
@SuppressLint("MissingPermission")
override suspend fun doWork(): Result = updateSemaphore.withPermit {
override suspend fun doWork(): Result {
val subId = inputData.getString(KEY_SUB_ID)
LogUtil.i(AppConfig.TAG, "SubscriptionUpdater update starting: $subId")
LogUtil.i(AppConfig.TAG, "SubscriptionUpdater update starting via Service: $subId")
if (subId.isNullOrEmpty()) {
LogUtil.w(AppConfig.TAG, "SubscriptionUpdater: missing subId in worker input")
return Result.success()
}
val subItem = MmkvManager.decodeSubscription(subId)
if (subItem == null) {
LogUtil.w(AppConfig.TAG, "SubscriptionUpdater: no subscription found for $subId")
return Result.success()
}
updateLastUpdatedAndReschedule(applicationContext, subId)
if (subItem.url.isEmpty()) {
LogUtil.i(AppConfig.TAG, "SubscriptionUpdater: url isEmpty for ${subItem.remarks}, skip")
return Result.success()
}
val sub = SubscriptionCache(subId, subItem)
// Notify about update start
showNotification(
MessageUtil.sendMsg2SubscriptionService(
applicationContext,
R.string.title_pref_auto_update_subscription,
"Updating ${sub.subscription.remarks}"
SubscriptionUpdateMessage(AppConfig.MSG_SUB_UPDATE_START, listOf(subId))
)
LogUtil.i(AppConfig.TAG, "SubscriptionUpdater update: ---${subItem.remarks}")
AngConfigManager.updateConfigViaSub(sub)
if (MmkvManager.decodeSettingsBool(AppConfig.PREF_AUTO_TEST_AFTER_UPDATE_SUBSCRIPTION, false)) {
testSubscriptionServers(applicationContext, sub)
if (MmkvManager.decodeSettingsBool(AppConfig.PREF_AUTO_REMOVE_INVALID_AFTER_TEST, false)) {
LogUtil.i(AppConfig.TAG, "SubscriptionUpdater: removing invalid servers for ${subItem.remarks}")
showNotification(
applicationContext,
R.string.title_del_invalid_config,
sub.subscription.remarks
)
AngConfigManager.removeInvalidServer(subId)
}
if (MmkvManager.decodeSettingsBool(AppConfig.PREF_AUTO_SORT_AFTER_TEST, false)) {
LogUtil.i(AppConfig.TAG, "SubscriptionUpdater: sorting servers for ${subItem.remarks}")
showNotification(
applicationContext,
R.string.title_sort_by_test_results,
sub.subscription.remarks
)
AngConfigManager.sortByTestResultsForSub(subId)
}
}
// Clear notification
NotificationHelper.cancel(NotificationChannelType.SUBSCRIPTION_UPDATE, applicationContext)
// Reset periodic task timer to align with this successful update
syncOne(applicationContext, subId)
LogUtil.i(AppConfig.TAG, "SubscriptionUpdater update finished and rescheduled: ${subItem.remarks}")
return Result.success()
}
}
private suspend fun testSubscriptionServers(context: Context, sub: SubscriptionCache) {
val subId = sub.guid
LogUtil.i(AppConfig.TAG, "SubscriptionUpdater: starting test phase for ${sub.subscription.remarks}")
showNotification(
context,
R.string.title_real_ping_all_server,
sub.subscription.remarks
)
CoreNativeManager.initCoreEnv(context)
val guids = MmkvManager.decodeServerList(subId)
if (guids.isNotEmpty()) {
val deferred = CompletableDeferred<Unit>()
val worker = RealPingWorkerService(
context = context,
guids = guids,
onEvent = { event ->
when (event) {
is RealPingEvent.Progress -> {
showNotification(
context,
R.string.title_real_ping_all_server,
"${event.text} in ${sub.subscription.remarks}"
)
LogUtil.i(AppConfig.TAG, "SubscriptionUpdater: ${event.text} in ${sub.subscription.remarks}")
}
is RealPingEvent.Result -> {
MmkvManager.encodeServerTestDelayMillis(event.guid, event.delayMillis)
}
is RealPingEvent.Finish -> {
deferred.complete(Unit)
}
}
}
)
worker.start()
deferred.await()
LogUtil.i(AppConfig.TAG, "SubscriptionUpdater: test phase finished for ${sub.subscription.remarks}")
}
}
private fun showNotification(context: Context, titleResId: Int, content: String) {
NotificationHelper.notify(
NotificationChannelType.SUBSCRIPTION_UPDATE,
context,
context.getString(titleResId),
content
)
}
}
@@ -0,0 +1,213 @@
package com.v2ray.ang.service
import android.app.Service
import android.content.Context
import android.content.Intent
import android.os.IBinder
import com.v2ray.ang.AppConfig
import com.v2ray.ang.R
import com.v2ray.ang.core.CoreNativeManager
import com.v2ray.ang.dto.RealPingEvent
import com.v2ray.ang.dto.SubscriptionUpdateMessage
import com.v2ray.ang.dto.entities.SubscriptionCache
import com.v2ray.ang.enums.NotificationChannelType
import com.v2ray.ang.extension.serializable
import com.v2ray.ang.handler.AngConfigManager
import com.v2ray.ang.handler.MmkvManager
import com.v2ray.ang.util.LogUtil
import com.v2ray.ang.util.NotificationHelper
import kotlinx.coroutines.CompletableDeferred
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.Job
import kotlinx.coroutines.launch
import kotlinx.coroutines.sync.Semaphore
import kotlinx.coroutines.sync.withPermit
import java.util.Collections
import java.util.concurrent.atomic.AtomicInteger
class SubscriptionUpdateService : Service() {
private val serviceJob = Job()
private val serviceScope = CoroutineScope(Dispatchers.IO + serviceJob)
private val runningTasks = AtomicInteger(0)
// manage active batch workers so each batch is independent and cancellable
private val activeWorkers = Collections.synchronizedList(mutableListOf<RealPingWorkerService>())
private val updateSemaphore = Semaphore(2)
override fun onCreate() {
super.onCreate()
CoreNativeManager.initCoreEnv(this)
}
override fun onBind(intent: Intent?): IBinder? = null
override fun onDestroy() {
LogUtil.i(AppConfig.TAG, "SubscriptionUpdateService is being destroyed")
val snapshot = ArrayList(activeWorkers)
snapshot.forEach { it.cancel() }
activeWorkers.clear()
serviceJob.cancel()
NotificationHelper.stopForeground(this)
NotificationHelper.cancel(NotificationChannelType.SUBSCRIPTION_UPDATE, this)
super.onDestroy()
}
override fun onStartCommand(intent: Intent?, flags: Int, startId: Int): Int {
val message = intent?.serializable<SubscriptionUpdateMessage>("content")
if (message == null) {
stopSelf(startId)
return START_NOT_STICKY
}
when (message.key) {
AppConfig.MSG_SUB_UPDATE_START -> handleUpdateStart(message)
AppConfig.MSG_SUB_UPDATE_CANCEL -> {
NotificationHelper.stopForeground(this)
stopSelf(startId)
}
else -> {
NotificationHelper.stopForeground(this)
stopSelf(startId)
}
}
return START_NOT_STICKY
}
private fun handleUpdateStart(message: SubscriptionUpdateMessage) {
LogUtil.i(AppConfig.TAG, "SubscriptionUpdateService starting update task for ${message.subIds.size} subscriptions")
NotificationHelper.startForeground(
this,
NotificationChannelType.SUBSCRIPTION_UPDATE,
getString(R.string.title_pref_auto_update_subscription),
getString(R.string.app_name)
)
runningTasks.incrementAndGet()
serviceScope.launch {
updateSemaphore.withPermit {
try {
message.subIds.forEach { subId ->
updateSingle(subId)
}
} catch (e: Exception) {
LogUtil.e(AppConfig.TAG, "SubscriptionUpdateService update failed", e)
} finally {
if (runningTasks.decrementAndGet() == 0 && activeWorkers.isEmpty()) {
NotificationHelper.stopForeground(this@SubscriptionUpdateService)
stopSelf()
}
}
}
}
}
private suspend fun updateSingle(subId: String) {
val subItem = MmkvManager.decodeSubscription(subId) ?: return
if(!subItem.enabled || subItem.url.isEmpty()){
return
}
val sub = SubscriptionCache(subId, subItem)
LogUtil.i(AppConfig.TAG, "SubscriptionUpdateService: Updating ${subItem.remarks}")
showNotification(
context = this,
titleResId = R.string.title_pref_auto_update_subscription,
content = "Updating ${subItem.remarks}"
)
AngConfigManager.updateConfigViaSub(sub)
if (MmkvManager.decodeSettingsBool(AppConfig.PREF_AUTO_TEST_AFTER_UPDATE_SUBSCRIPTION, false)) {
testSubscriptionServers(sub)
if (MmkvManager.decodeSettingsBool(AppConfig.PREF_AUTO_REMOVE_INVALID_AFTER_TEST, false)) {
LogUtil.i(AppConfig.TAG, "SubscriptionUpdateService: removing invalid servers for ${subItem.remarks}")
showNotification(
context = this,
titleResId = R.string.title_del_invalid_config,
content = subItem.remarks
)
AngConfigManager.removeInvalidServer(subId)
}
if (MmkvManager.decodeSettingsBool(AppConfig.PREF_AUTO_SORT_AFTER_TEST, false)) {
LogUtil.i(AppConfig.TAG, "SubscriptionUpdateService: sorting servers for ${subItem.remarks}")
showNotification(
context = this,
titleResId = R.string.title_sort_by_test_results,
content = subItem.remarks
)
AngConfigManager.sortByTestResultsForSub(subId)
}
}
LogUtil.i(AppConfig.TAG, "SubscriptionUpdateService: Finished ${subItem.remarks}")
}
private suspend fun testSubscriptionServers(sub: SubscriptionCache) {
val subId = sub.guid
LogUtil.i(AppConfig.TAG, "SubscriptionUpdateService: starting test phase for ${sub.subscription.remarks}")
showNotification(
context = this,
titleResId = R.string.title_real_ping_all_server,
content = sub.subscription.remarks
)
val guids = MmkvManager.decodeServerList(subId)
if (guids.isNotEmpty()) {
val deferred = CompletableDeferred<Unit>()
lateinit var worker: RealPingWorkerService
worker = RealPingWorkerService(
context = this,
guids = guids,
onEvent = { event ->
handleWorkerEvent(event, sub.subscription.remarks) {
activeWorkers.remove(worker)
deferred.complete(Unit)
}
}
)
activeWorkers.add(worker)
worker.start()
deferred.await()
LogUtil.i(AppConfig.TAG, "SubscriptionUpdateService: test phase finished for ${sub.subscription.remarks}")
}
}
private fun handleWorkerEvent(event: RealPingEvent, remarks: String, onWorkerDone: () -> Unit) {
when (event) {
is RealPingEvent.Progress -> {
val text = "${event.text} in $remarks"
showNotification(
context = this,
titleResId = R.string.title_real_ping_all_server,
content = text
)
LogUtil.i(AppConfig.TAG, "SubscriptionUpdateService: $text")
}
is RealPingEvent.Result -> {
MmkvManager.encodeServerTestDelayMillis(event.guid, event.delayMillis)
}
is RealPingEvent.Finish -> {
onWorkerDone()
}
}
}
private fun showNotification(context: Context, titleResId: Int, content: String) {
NotificationHelper.notify(
NotificationChannelType.SUBSCRIPTION_UPDATE,
context,
context.getString(titleResId),
content
)
}
}
@@ -1,14 +1,16 @@
package com.v2ray.ang.ui.subscription
import android.app.Application
import com.v2ray.ang.AppConfig
import com.v2ray.ang.R
import com.v2ray.ang.dto.SubscriptionUpdateMessage
import com.v2ray.ang.dto.entities.SubscriptionCache
import com.v2ray.ang.dto.entities.SubscriptionItem
import com.v2ray.ang.handler.MmkvManager
import com.v2ray.ang.handler.SettingsChangeManager
import com.v2ray.ang.handler.SettingsManager
import com.v2ray.ang.handler.SubscriptionUpdater
import com.v2ray.ang.ui.base.BaseViewModel
import com.v2ray.ang.util.MessageUtil
import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.StateFlow
import kotlinx.coroutines.flow.asStateFlow
@@ -59,7 +61,13 @@ class SubscriptionsViewModel(application: Application) : BaseViewModel(applicati
fun updateSubscriptions() {
SettingsChangeManager.makeSetupGroupTab()
SubscriptionUpdater.updateAllByManual(app)
val subIds = MmkvManager.decodeSubscriptions()
.filter { it.subscription.enabled && it.subscription.url.isNotEmpty() }
.map { it.guid }
if (subIds.isNotEmpty()) {
MessageUtil.sendMsg2SubscriptionService(app, SubscriptionUpdateMessage(AppConfig.MSG_SUB_UPDATE_START, subIds))
}
toast(R.string.subscription_updater_job_tips)
}
@@ -6,8 +6,10 @@ import android.content.Intent
import android.os.Build
import androidx.core.content.ContextCompat
import com.v2ray.ang.AppConfig
import com.v2ray.ang.dto.SubscriptionUpdateMessage
import com.v2ray.ang.dto.TestServiceMessage
import com.v2ray.ang.service.CoreTestService
import com.v2ray.ang.service.SubscriptionUpdateService
import java.io.Serializable
object MessageUtil {
@@ -69,6 +71,39 @@ object MessageUtil {
}
}
/**
* Sends a message to the subscription service.
*
* @param ctx The context.
* @param message The subscription service message containing key and subId.
*/
fun sendMsg2SubscriptionService(ctx: Context, message: SubscriptionUpdateMessage) {
try {
val intent = Intent()
intent.component = ComponentName(ctx, SubscriptionUpdateService::class.java)
intent.putExtra("content", message)
when (message.key) {
AppConfig.MSG_SUB_UPDATE_START -> {
if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.O) {
ContextCompat.startForegroundService(ctx, intent)
} else {
ctx.startService(intent)
}
}
AppConfig.MSG_SUB_UPDATE_CANCEL -> {
ctx.stopService(intent)
}
else -> {
ctx.startService(intent)
}
}
} catch (e: Exception) {
LogUtil.e(AppConfig.TAG, "Failed to send message to subscription service", e)
}
}
/**
* Sends a message with the specified action.
*