diff --git a/V2rayNG/app/src/main/AndroidManifest.xml b/V2rayNG/app/src/main/AndroidManifest.xml index 0238a58610..4b397870c0 100644 --- a/V2rayNG/app/src/main/AndroidManifest.xml +++ b/V2rayNG/app/src/main/AndroidManifest.xml @@ -256,7 +256,7 @@ android:name=".service.CoreTestService" android:exported="false" android:foregroundServiceType="specialUse" - android:process=":RunSoLibV2RayDaemon"> + android:process=":Probe"> @@ -266,7 +266,7 @@ android:name=".service.SubscriptionUpdateService" android:exported="false" android:foregroundServiceType="specialUse" - android:process=":RunSoLibV2RayDaemon"> + android:process=":SubscriptionUpdate"> diff --git a/V2rayNG/app/src/main/java/com/v2ray/ang/core/CoreConfigContextBuilder.kt b/V2rayNG/app/src/main/java/com/v2ray/ang/core/CoreConfigContextBuilder.kt index 42ed66980a..7a38957bb8 100644 --- a/V2rayNG/app/src/main/java/com/v2ray/ang/core/CoreConfigContextBuilder.kt +++ b/V2rayNG/app/src/main/java/com/v2ray/ang/core/CoreConfigContextBuilder.kt @@ -4,6 +4,7 @@ import android.content.Context import com.v2ray.ang.AppConfig import com.v2ray.ang.dto.CoreConfigContext import com.v2ray.ang.dto.entities.ProfileItem +import com.v2ray.ang.dto.entities.SubscriptionItem import com.v2ray.ang.enums.BalancerStrategyType import com.v2ray.ang.enums.CoreResolvedType import com.v2ray.ang.enums.EConfigType @@ -23,6 +24,68 @@ import com.v2ray.ang.util.Utils */ object CoreConfigContextBuilder { + /** Lazily decoded profile snapshot shared by every config in one probe batch. */ + internal class ProbeProfileLookup(requestedGuids: List) { + private val profilesByGuid = linkedMapOf() + private val subscriptionsByGuid = mutableMapOf() + private var profilesByRemarks: Map? = null + private var allProfiles: List? = null + + init { + requestedGuids.forEach(::loadProfile) + } + + fun findByGuid(guid: String): ProfileItem? = + profilesByGuid[guid] ?: loadProfile(guid) + + fun findByRemarks(remarks: String?): ProfileItem? { + if (remarks.isNullOrEmpty()) return null + ensureAllProfilesLoaded() + return profilesByRemarks?.get(remarks) + } + + fun profiles(): List { + ensureAllProfilesLoaded() + return allProfiles.orEmpty() + } + + fun subscription(guid: String): SubscriptionItem? { + if (guid in subscriptionsByGuid) return subscriptionsByGuid[guid] + return MmkvManager.decodeSubscription(guid).also { subscriptionsByGuid[guid] = it } + } + + private fun loadProfile(guid: String): ProfileItem? { + if (guid.isBlank()) return null + return MmkvManager.decodeServerConfig(guid)?.also { profilesByGuid[guid] = it } + } + + private fun ensureAllProfilesLoaded() { + if (allProfiles != null) return + val ordered = mutableListOf() + val seenGuids = mutableSetOf() + MmkvManager.decodeAllServerList().forEach { guid -> + val profile = findByGuid(guid) ?: return@forEach + if (seenGuids.add(guid)) ordered += profile + } + profilesByGuid.forEach { (guid, profile) -> + if (seenGuids.add(guid)) ordered += profile + } + allProfiles = ordered + profilesByRemarks = buildMap { + ordered.forEach { profile -> putIfAbsent(profile.remarks, profile) } + } + } + } + + private fun findProfileByRemarks( + lookup: ProbeProfileLookup?, + remarks: String?, + ): ProfileItem? = if (lookup != null) { + lookup.findByRemarks(remarks) + } else { + SettingsManager.getServerViaRemarks(remarks) + } + /** * Load one profile and produce a fully analyzed context. * @@ -31,22 +94,42 @@ object CoreConfigContextBuilder { fun build(context: Context, guid: String): CoreConfigContext? { val config = MmkvManager.decodeServerConfig(guid) ?: return null + return buildResolved(context, guid, config, lookup = null, includeRouting = true) + } + + /** Build only the outbound dependency graph required by a RealDelay probe. */ + internal fun buildForProbe( + context: Context, + guid: String, + lookup: ProbeProfileLookup, + ): CoreConfigContext? { + val config = lookup.findByGuid(guid) ?: return null + return buildResolved(context, guid, config, lookup, includeRouting = false) + } + + private fun buildResolved( + context: Context, + guid: String, + config: ProfileItem, + lookup: ProbeProfileLookup?, + includeRouting: Boolean, + ): CoreConfigContext? { // CUSTOM: return immediately — CoreConfigManager handles this path on its own. if (config.configType == EConfigType.CUSTOM) { return CoreConfigContext(context = context, guid = guid, isCustom = true) } // Step 1: Resolve the main outbound (always tag = TAG_PROXY). - val primaryResolvedOutbound = resolveOutbound(AppConfig.TAG_PROXY, config) ?: run { + val primaryResolvedOutbound = resolveOutbound(AppConfig.TAG_PROXY, config, lookup) ?: run { LogUtil.e(AppConfig.TAG, "Failed to resolve main outbound for '${config.remarks}'") return null } // Step 2: Resolve all non-builtin routing outbound tags. - val routingResolvedOutbounds = resolveRoutingOutbounds() + val routingResolvedOutbounds = if (includeRouting) resolveRoutingOutbounds() else emptyList() val resolvedOutbounds = listOf(primaryResolvedOutbound) + routingResolvedOutbounds - val fallbackResolvedOutbounds = resolveFallbackOutbounds(resolvedOutbounds) - val routingDomainRules = collectRoutingDomainRulesForDns() + val fallbackResolvedOutbounds = resolveFallbackOutbounds(resolvedOutbounds, lookup) + val routingDomainRules = if (includeRouting) collectRoutingDomainRulesForDns() else emptyList() return CoreConfigContext( context = context, @@ -61,25 +144,29 @@ object CoreConfigContextBuilder { * * Custom profiles are ignored at this stage and produce no entry. */ - private fun resolveOutbound(tag: String, profile: ProfileItem): CoreConfigContext.ResolvedOutbound? { + private fun resolveOutbound( + tag: String, + profile: ProfileItem, + lookup: ProbeProfileLookup? = null, + ): CoreConfigContext.ResolvedOutbound? { if (profile.configType == EConfigType.CUSTOM) { return null } val (resolvedProfiles, resolvedType) = when (profile.configType) { EConfigType.POLICYGROUP -> Pair( - resolvePolicyGroupProfiles(profile), + resolvePolicyGroupProfiles(profile, lookup), CoreResolvedType.POLICYGROUP, ) EConfigType.PROXYCHAIN -> { - val chainProfiles = resolveProxyChainProfiles(profile) + val chainProfiles = resolveProxyChainProfiles(profile, lookup) val type = if (chainProfiles.size <= 1) CoreResolvedType.NORMAL else CoreResolvedType.PROXYCHAIN Pair(chainProfiles, type) } else -> { - val chainProfiles = resolveProxyChainProfilesFromGroup(profile) + val chainProfiles = resolveProxyChainProfilesFromGroup(profile, lookup) val type = if (chainProfiles.size <= 1) CoreResolvedType.NORMAL else CoreResolvedType.PROXYCHAIN Pair(chainProfiles, type) } @@ -141,12 +228,14 @@ object CoreConfigContextBuilder { return resolvedOutbounds } - private fun resolvePolicyGroupProfiles(config: ProfileItem): List { + private fun resolvePolicyGroupProfiles( + config: ProfileItem, + lookup: ProbeProfileLookup?, + ): List { try { - val serverList = MmkvManager.decodeAllServerList() - return serverList - .asSequence() - .mapNotNull { id -> MmkvManager.decodeServerConfig(id) } + val profiles = lookup?.profiles() ?: MmkvManager.decodeAllServerList() + .mapNotNull(MmkvManager::decodeServerConfig) + return profiles.asSequence() .filter { profile -> val subscriptionId = config.policyGroupSubscriptionId if (subscriptionId.isNullOrBlank()) { @@ -177,7 +266,10 @@ object CoreConfigContextBuilder { } } - private fun resolveProxyChainProfiles(config: ProfileItem): List { + private fun resolveProxyChainProfiles( + config: ProfileItem, + lookup: ProbeProfileLookup?, + ): List { if (config.proxyChainProfiles.isNullOrBlank()) { return listOf(config) } @@ -185,7 +277,7 @@ object CoreConfigContextBuilder { try { return config.proxyChainProfiles.orEmpty().split(",") .asSequence() - .mapNotNull { remark -> SettingsManager.getServerViaRemarks(remark) } + .mapNotNull { remark -> findProfileByRemarks(lookup, remark) } .filter { it.server.isNotNullEmpty() } .filter { Utils.isPureIpAddress(it.server!!) || Utils.isValidUrl(it.server!!) } .filter { !it.configType.isComplexType() } @@ -202,17 +294,26 @@ object CoreConfigContextBuilder { * * When no chain is available, return a single-node result. */ - private fun resolveProxyChainProfilesFromGroup(config: ProfileItem): List { + private fun resolveProxyChainProfilesFromGroup( + config: ProfileItem, + lookup: ProbeProfileLookup?, + ): List { if (config.subscriptionId.isEmpty()) { return listOf(config) } try { - val subItem = MmkvManager.decodeSubscription(config.subscriptionId) ?: return listOf(config) + val subItem = if (lookup != null) { + lookup.subscription(config.subscriptionId) + } else { + MmkvManager.decodeSubscription(config.subscriptionId) + } ?: return listOf(config) val resolved = mutableListOf() - SettingsManager.getServerViaRemarks(subItem.nextProfile)?.let { resolved.add(it) } + findProfileByRemarks(lookup, subItem.nextProfile) + ?.let { resolved.add(it) } resolved.add(config) - SettingsManager.getServerViaRemarks(subItem.prevProfile)?.let { resolved.add(it) } + findProfileByRemarks(lookup, subItem.prevProfile) + ?.let { resolved.add(it) } return resolved } catch (e: Exception) { LogUtil.e(AppConfig.TAG, "Failed to resolve proxy chain from group for '${config.remarks}'", e) @@ -255,7 +356,10 @@ object CoreConfigContextBuilder { * * Fallback targets must not overlap with already resolved tags or builtin tags. */ - private fun resolveFallbackOutbounds(resolvedOutbounds: List): List { + private fun resolveFallbackOutbounds( + resolvedOutbounds: List, + lookup: ProbeProfileLookup?, + ): List { return resolvedOutbounds .asSequence() .filter { it.resolvedType == CoreResolvedType.POLICYGROUP } @@ -264,9 +368,9 @@ object CoreConfigContextBuilder { .filter { it !in AppConfig.BUILTIN_OUTBOUND_TAGS && resolvedOutbounds.none { outbound -> outbound.tag == it } } .distinct() .mapNotNull { tag -> - SettingsManager.getServerViaRemarks(tag) + findProfileByRemarks(lookup, tag) ?.takeUnless { it.configType == EConfigType.CUSTOM || it.configType == EConfigType.POLICYGROUP } - ?.let { resolveOutbound(tag, it) } + ?.let { resolveOutbound(tag, it, lookup) } } .toList() } diff --git a/V2rayNG/app/src/main/java/com/v2ray/ang/core/CoreConfigManager.kt b/V2rayNG/app/src/main/java/com/v2ray/ang/core/CoreConfigManager.kt index e40ce604c6..9fe768e4f9 100644 --- a/V2rayNG/app/src/main/java/com/v2ray/ang/core/CoreConfigManager.kt +++ b/V2rayNG/app/src/main/java/com/v2ray/ang/core/CoreConfigManager.kt @@ -7,6 +7,7 @@ import com.google.gson.JsonObject import com.v2ray.ang.AppConfig import com.v2ray.ang.dto.ConfigResult import com.v2ray.ang.dto.CoreConfigContext +import com.v2ray.ang.dto.ProbePlan import com.v2ray.ang.dto.V2rayConfig import com.v2ray.ang.dto.entities.ProfileItem import com.v2ray.ang.dto.entities.RulesetItem @@ -58,7 +59,7 @@ object CoreConfigManager { * * The core flow is reused, then non-essential sections are removed. */ - fun getV2rayConfig4Speedtest(context: Context, guid: String): ConfigResult { + fun getV2rayConfig4RealDelay(context: Context, guid: String): ConfigResult { try { val configContext = CoreConfigContextBuilder.build(context, guid) ?: return ConfigResult( @@ -69,20 +70,52 @@ object CoreConfigManager { if (configContext.isCustom) { return buildV2rayCustomConfig(configContext) } - val v2rayConfig = buildUnifiedConfig(configContext) - postProcessForSpeedtest(v2rayConfig) - - return toConfigResult(configContext, v2rayConfig) + return toConfigResult(configContext, buildRealDelayConfig(configContext)) } catch (e: Exception) { - LogUtil.e(AppConfig.TAG, "Failed to get V2ray config for speedtest", e) + LogUtil.e(AppConfig.TAG, "Failed to get V2ray config for real delay", e) return ConfigResult( status = false, guid = guid, - errorMessage = "Failed to get V2ray config: ${e.message ?: e.javaClass.simpleName}" + errorMessage = "Failed to get V2ray config for real delay: ${e.message ?: e.javaClass.simpleName}" ) } } + /** Builds one isolated Xray configuration for a complete UI delay-test batch. */ + internal fun getProbePlan(context: Context, guids: List): ProbePlan { + val sources = mutableListOf() + val individualGuids = mutableListOf() + val failedGuids = mutableListOf() + val distinctGuids = guids.distinct() + val profileLookup = CoreConfigContextBuilder.ProbeProfileLookup(distinctGuids) + distinctGuids.forEach { guid -> + try { + val configContext = CoreConfigContextBuilder.buildForProbe(context, guid, profileLookup) + if (configContext == null) { + failedGuids += guid + } else if (configContext.isCustom) { + individualGuids += guid + } else { + sources += ProbeConfigBuilder.Source(guid, buildRealDelayConfig(configContext)) + } + } catch (error: Exception) { + LogUtil.e(AppConfig.TAG, "Failed to build probe config for $guid", error) + failedGuids += guid + } + } + val plan = ProbeConfigBuilder.build( + sources = sources, + destination = SettingsManager.getDelayTestUrl(), + ) + return plan.copy( + individualGuids = individualGuids + plan.individualGuids, + failedGuids = failedGuids, + ) + } + + private fun buildRealDelayConfig(configContext: CoreConfigContext): V2rayConfig = + buildUnifiedConfig(configContext).also(::postProcessForRealDelay) + /** * Build configuration for custom profiles. */ @@ -432,10 +465,21 @@ object CoreConfigManager { /** * Trim runtime sections that are not needed for latency testing. */ - private fun postProcessForSpeedtest(v2rayConfig: V2rayConfig) { + private fun postProcessForRealDelay(v2rayConfig: V2rayConfig) { v2rayConfig.log.loglevel = MmkvManager.decodeSettingsString(AppConfig.PREF_LOGLEVEL) ?: "warning" v2rayConfig.inbounds.clear() + val usesPrimaryBalancer = v2rayConfig.routing.balancers + ?.any { it.tag == AppConfig.TAG_BALANCER } + ?: false v2rayConfig.routing.rules.clear() + if (usesPrimaryBalancer) { + v2rayConfig.routing.rules.add( + V2rayConfig.RoutingBean.RulesBean( + network = "tcp,udp", + balancerTag = AppConfig.TAG_BALANCER, + ) + ) + } v2rayConfig.dns = null v2rayConfig.fakedns = null v2rayConfig.stats = null @@ -704,7 +748,7 @@ object CoreConfigManager { } /** - * Remove speed-test runtime sections when the feature is disabled. + * Remove speed-display runtime sections when the feature is disabled. */ private fun applySpeedDisabled(v2rayConfig: V2rayConfig) { if (MmkvManager.decodeSettingsBool(AppConfig.PREF_SPEED_ENABLED) != true) { diff --git a/V2rayNG/app/src/main/java/com/v2ray/ang/core/CoreNativeManager.kt b/V2rayNG/app/src/main/java/com/v2ray/ang/core/CoreNativeManager.kt index 663d0629d4..124baa7091 100644 --- a/V2rayNG/app/src/main/java/com/v2ray/ang/core/CoreNativeManager.kt +++ b/V2rayNG/app/src/main/java/com/v2ray/ang/core/CoreNativeManager.kt @@ -67,22 +67,6 @@ object CoreNativeManager { } } - /** - * Measure outbound connection delay. - * - * @param config The configuration JSON string - * @param testUrl The URL to test against - * @return Delay in milliseconds, or -1 if test failed - */ - fun measureOutboundDelay(config: String, testUrl: String): Long { - return try { - Libv2ray.measureOutboundDelay(config, testUrl) - } catch (e: Exception) { - LogUtil.e(AppConfig.TAG, "Failed to measure outbound delay", e) - -1L - } - } - /** * Create a new core controller instance. * @@ -97,4 +81,4 @@ object CoreNativeManager { throw e } } -} \ No newline at end of file +} diff --git a/V2rayNG/app/src/main/java/com/v2ray/ang/core/ProbeConfigBuilder.kt b/V2rayNG/app/src/main/java/com/v2ray/ang/core/ProbeConfigBuilder.kt new file mode 100644 index 0000000000..ca3100fed1 --- /dev/null +++ b/V2rayNG/app/src/main/java/com/v2ray/ang/core/ProbeConfigBuilder.kt @@ -0,0 +1,128 @@ +package com.v2ray.ang.core + +import com.v2ray.ang.AppConfig +import com.v2ray.ang.dto.ProbePlan +import com.v2ray.ang.dto.ProbeProfile +import com.v2ray.ang.dto.V2rayConfig +import com.v2ray.ang.util.JsonUtil + +/** Combines v2rayNG-generated real-delay configurations into one probe core. */ +internal object ProbeConfigBuilder { + data class Source(val guid: String, val config: V2rayConfig) + + fun build(sources: List, destination: String): ProbePlan { + val outbounds = mutableListOf() + val balancers = mutableListOf() + val profiles = mutableListOf() + val individualGuids = mutableListOf() + + sources.forEachIndexed { index, source -> + val prepared = try { + prepareSource(source, index) + } catch (_: Exception) { + null + } + if (prepared == null) { + individualGuids += source.guid + return@forEachIndexed + } + outbounds += prepared.outbounds + prepared.balancer?.let(balancers::add) + profiles += prepared.profile + } + + val routing = mutableMapOf( + "domainStrategy" to "AsIs", + "rules" to emptyList(), + ) + if (balancers.isNotEmpty()) routing["balancers"] = balancers + + val config = mapOf( + "log" to mapOf("loglevel" to "warning"), + "outbounds" to outbounds, + "routing" to routing, + "burstObservatory" to mapOf( + "subjectSelector" to emptyList(), + "pingConfig" to mapOf( + "destination" to destination, + "httpMethod" to DEFAULT_HTTP_METHOD, + "interval" to "1h", + "sampling" to 1, + "timeout" to DEFAULT_TIMEOUT, + ), + ), + ) + return ProbePlan( + content = JsonUtil.toJson(config), + profiles = profiles, + individualGuids = individualGuids, + ) + } + + /** Validate and rewrite one source completely before exposing any part of it to the batch. */ + private fun prepareSource(source: Source, index: Int): PreparedSource? { + val sourceOutbounds = source.config.outbounds + val namespace = "probe-$index-" + val tagMap = sourceOutbounds.associate { it.tag to "$namespace${it.tag}" } + val primaryBalancer = source.config.routing.balancers + ?.firstOrNull { it.tag == AppConfig.TAG_BALANCER } + val strategyType = primaryBalancer?.strategy?.type?.lowercase() + if (primaryBalancer != null && + (strategyType !in SUPPORTED_BALANCER_STRATEGIES || primaryBalancer.fallbackTag != null) + ) { + return null + } + + // Resolve every reference before mutating anything, so a malformed source + // cannot leave part of itself in the shared configuration. + val mappedDialerProxies = sourceOutbounds.map { outbound -> + outbound.streamSettings?.sockopt?.dialerProxy?.let { tagMap[it] ?: return null } + } + + val profile: ProbeProfile + val probeBalancer: V2rayConfig.RoutingBean.BalancerBean? + if (primaryBalancer == null) { + val proxyTag = tagMap[AppConfig.TAG_PROXY] ?: return null + profile = ProbeProfile(source.guid, listOf(proxyTag)) + probeBalancer = null + } else { + val selectors = primaryBalancer.selector + if (selectors.isEmpty() || selectors.any(String::isBlank)) return null + val outboundTags = tagMap + .filterKeys { tag -> selectors.any(tag::startsWith) } + .values + .toList() + if (outboundTags.isEmpty()) return null + probeBalancer = primaryBalancer.copy( + tag = "$namespace${primaryBalancer.tag}", + selector = selectors.map { "$namespace$it" }, + ) + profile = ProbeProfile( + guid = source.guid, + outboundTags = outboundTags, + balancerTag = probeBalancer.tag, + ) + } + + sourceOutbounds.forEachIndexed { outboundIndex, outbound -> + outbound.tag = tagMap.getValue(outbound.tag) + mappedDialerProxies[outboundIndex]?.let { mapped -> + outbound.streamSettings?.sockopt?.dialerProxy = mapped + } + } + return PreparedSource(sourceOutbounds, probeBalancer, profile) + } + + private data class PreparedSource( + val outbounds: List, + val balancer: V2rayConfig.RoutingBean.BalancerBean?, + val profile: ProbeProfile, + ) + + private val SUPPORTED_BALANCER_STRATEGIES = setOf( + "leastping", + "leastload", + ) + private const val DEFAULT_HTTP_METHOD = "HEAD" + private const val DEFAULT_TIMEOUT = "5s" +} diff --git a/V2rayNG/app/src/main/java/com/v2ray/ang/dto/ProbePlan.kt b/V2rayNG/app/src/main/java/com/v2ray/ang/dto/ProbePlan.kt new file mode 100644 index 0000000000..32b0a9f531 --- /dev/null +++ b/V2rayNG/app/src/main/java/com/v2ray/ang/dto/ProbePlan.kt @@ -0,0 +1,14 @@ +package com.v2ray.ang.dto + +data class ProbeProfile( + val guid: String, + val outboundTags: List, + val balancerTag: String? = null, +) + +data class ProbePlan( + val content: String, + val profiles: List, + val individualGuids: List = emptyList(), + val failedGuids: List = emptyList(), +) diff --git a/V2rayNG/app/src/main/java/com/v2ray/ang/dto/RealPingResult.kt b/V2rayNG/app/src/main/java/com/v2ray/ang/dto/RealPingResult.kt new file mode 100644 index 0000000000..ab2b8614df --- /dev/null +++ b/V2rayNG/app/src/main/java/com/v2ray/ang/dto/RealPingResult.kt @@ -0,0 +1,9 @@ +package com.v2ray.ang.dto + +import java.io.Serializable + +/** One persisted RealDelay result delivered from the probe process to the UI. */ +data class RealPingResult( + val guid: String, + val delayMillis: Long, +) : Serializable diff --git a/V2rayNG/app/src/main/java/com/v2ray/ang/service/CoreTestService.kt b/V2rayNG/app/src/main/java/com/v2ray/ang/service/CoreTestService.kt index f491711886..15ca659baa 100644 --- a/V2rayNG/app/src/main/java/com/v2ray/ang/service/CoreTestService.kt +++ b/V2rayNG/app/src/main/java/com/v2ray/ang/service/CoreTestService.kt @@ -4,12 +4,16 @@ import android.app.PendingIntent import android.app.Service import android.content.Context import android.content.Intent +import android.os.Handler import android.os.IBinder +import android.os.Looper +import android.os.Process import androidx.core.app.NotificationCompat 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.RealPingResult import com.v2ray.ang.dto.TestServiceMessage import com.v2ray.ang.enums.NotificationChannelType import com.v2ray.ang.extension.serializable @@ -19,16 +23,18 @@ import com.v2ray.ang.handler.MmkvManager import com.v2ray.ang.helper.MessageHelper import com.v2ray.ang.helper.NotificationHelper import com.v2ray.ang.util.LogUtil -import java.util.Collections class CoreTestService : Service() { + @Volatile + private var activeWorker: RealPingWorkerService? = null + @Volatile + private var replacementRequested = false + private var batchStarted = false override fun attachBaseContext(newBase: Context?) { super.attachBaseContext(newBase?.let(AppLocaleManager::localizedContext)) } - // manage active batch workers so each batch is independent and cancellable - private val activeWorkers = Collections.synchronizedList(mutableListOf()) private val cancelAction by lazy { val intent = Intent(this, CoreTestService::class.java).putExtra( "content", @@ -47,111 +53,113 @@ class CoreTestService : Service() { ).build() } - /** - * Initializes the V2Ray environment. - */ override fun onCreate() { super.onCreate() CoreNativeManager.initCoreEnv(this) } - /** - * Binds the service. - * @param intent The intent. - * @return The binder. - */ - override fun onBind(intent: Intent?): IBinder? { - return null - } + override fun onBind(intent: Intent?): IBinder? = null - /** - * Cleans up resources when the service is destroyed. - */ override fun onDestroy() { - LogUtil.i(AppConfig.TAG, "CoreTestService is being destroyed, cancelling ${activeWorkers.size} active workers") - // cancel any active workers - val snapshot = ArrayList(activeWorkers) - snapshot.forEach { it.cancel() } - activeWorkers.clear() + LogUtil.i(AppConfig.TAG, "CoreTestService is being destroyed") + activeWorker?.cancel() + activeWorker = null NotificationHelper.stopForeground(this) super.onDestroy() + // A new process for every batch prevents Xray's process-wide state from + // leaking into a later probe or overlapping the long-running VPN core. + Handler(Looper.getMainLooper()).post { Process.killProcess(Process.myPid()) } } - /** - * Handles the start command for the service. - * @param intent The intent. - * @param flags The flags. - * @param startId The start ID. - * @return The start mode. - */ override fun onStartCommand(intent: Intent?, flags: Int, startId: Int): Int { NotificationHelper.startForeground( this, NotificationChannelType.CORE_TEST, getString(R.string.app_name), getString(R.string.title_real_ping_all_server), - cancelAction + cancelAction, ) val message = intent?.serializable("content") if (message == null) { stopSelf(startId) return START_NOT_STICKY } - - when (message.key) { + return when (message.key) { AppConfig.MSG_MEASURE_CONFIG_START -> handleMeasureStart(message, startId) - AppConfig.MSG_MEASURE_CONFIG_CANCEL -> handleMeasureCancel() + AppConfig.MSG_MEASURE_CONFIG_CANCEL -> handleMeasureCancel(startId) else -> { - NotificationHelper.stopForeground(this); stopSelf(startId) + NotificationHelper.stopForeground(this) + stopSelf(startId) + START_NOT_STICKY } } - return START_NOT_STICKY } - private fun handleMeasureStart(message: TestServiceMessage, startId: Int) { - LogUtil.i(AppConfig.TAG, "CoreTestService starting worker subscription ${message.subscriptionId}") + private fun handleMeasureStart(message: TestServiceMessage, startId: Int): Int { + if (batchStarted) { + replacementRequested = true + LogUtil.i(AppConfig.TAG, "CoreTestService handing the next batch to a fresh process") + Handler(Looper.getMainLooper()).post { Process.killProcess(Process.myPid()) } + return START_REDELIVER_INTENT + } + batchStarted = true - val guidsList = when { + val guids = when { message.serverGuids.isNotEmpty() -> message.serverGuids message.subscriptionId.isNotEmpty() -> MmkvManager.decodeServerList(message.subscriptionId) else -> MmkvManager.decodeAllServerList() } - - if (guidsList.isNotEmpty()) { - lateinit var worker: RealPingWorkerService - worker = RealPingWorkerService( - context = this, - guids = guidsList, - onlyTcp = message.onlyTcp, - onEvent = { event -> handleWorkerEvent(event, message) { activeWorkers.remove(worker) } } - ) - activeWorkers.add(worker) - worker.start() - } else { + if (guids.isEmpty()) { + MessageHelper.sendMsg2UI(this, AppConfig.MSG_MEASURE_CONFIG_FINISH, "0") NotificationHelper.stopForeground(this) stopSelf(startId) + return START_NOT_STICKY } + + LogUtil.i(AppConfig.TAG, "CoreTestService starting a ${guids.size}-profile batch") + activeWorker = RealPingWorkerService( + context = this, + guids = guids, + onlyTcp = message.onlyTcp, + onEvent = { event -> handleWorkerEvent(event, message) }, + ).also { it.start() } + return START_NOT_STICKY + } + + private fun handleMeasureCancel(startId: Int): Int { + LogUtil.i(AppConfig.TAG, "CoreTestService cancelling the active batch") + replacementRequested = false + activeWorker?.cancel() + activeWorker = null + NotificationHelper.stopForeground(this) + stopSelf(startId) + return START_NOT_STICKY } - private fun handleWorkerEvent(event: RealPingEvent, message: TestServiceMessage, onWorkerDone: () -> Unit) { + private fun handleWorkerEvent(event: RealPingEvent, message: TestServiceMessage) { + if (replacementRequested) return when (event) { is RealPingEvent.Progress -> { NotificationHelper.updateNotification( channelType = NotificationChannelType.CORE_TEST, context = this, title = getString(R.string.app_name), - content = getString(R.string.connection_running_task_left, event.text) + content = getString(R.string.connection_running_task_left, event.text), ) MessageHelper.sendMsg2UI(this, AppConfig.MSG_MEASURE_CONFIG_NOTIFY, event.text) } is RealPingEvent.Result -> { MmkvManager.encodeServerTestDelayMillis(event.guid, event.delayMillis) - MessageHelper.sendMsg2UI(this, AppConfig.MSG_MEASURE_CONFIG_SUCCESS, event.guid) + MessageHelper.sendMsg2UI( + this, + AppConfig.MSG_MEASURE_CONFIG_SUCCESS, + RealPingResult(event.guid, event.delayMillis), + ) } is RealPingEvent.Finish -> { - if (message.subscriptionId.isNotEmpty()) { + if (event.status == "0" && message.subscriptionId.isNotEmpty()) { if (MmkvManager.decodeSettingsBool(AppConfig.PREF_AUTO_REMOVE_INVALID_AFTER_TEST, false)) { AngConfigManager.removeInvalidServer(message.subscriptionId) } @@ -160,24 +168,11 @@ class CoreTestService : Service() { AngConfigManager.sortByTestResultsForSub(message.subscriptionId) } } - MessageHelper.sendMsg2UI(this, AppConfig.MSG_MEASURE_CONFIG_FINISH, event.status) - onWorkerDone() - if (activeWorkers.isEmpty()) { - NotificationHelper.stopForeground(this) - stopSelf() - } + activeWorker = null + NotificationHelper.stopForeground(this) + stopSelf() } } } - - private fun handleMeasureCancel() { - MessageHelper.sendMsg2UI(this, AppConfig.MSG_MEASURE_CONFIG_FINISH, "0") - LogUtil.i(AppConfig.TAG, "CoreTestService received cancel message, cancelling ${activeWorkers.size} active workers") - val snapshot = ArrayList(activeWorkers) - snapshot.forEach { it.cancel() } - activeWorkers.clear() - NotificationHelper.stopForeground(this) - stopSelf() - } } diff --git a/V2rayNG/app/src/main/java/com/v2ray/ang/service/RealPingWorkerService.kt b/V2rayNG/app/src/main/java/com/v2ray/ang/service/RealPingWorkerService.kt index d37346f1ab..1c506dc0b6 100644 --- a/V2rayNG/app/src/main/java/com/v2ray/ang/service/RealPingWorkerService.kt +++ b/V2rayNG/app/src/main/java/com/v2ray/ang/service/RealPingWorkerService.kt @@ -1,8 +1,10 @@ package com.v2ray.ang.service import android.content.Context +import android.os.SystemClock +import com.v2ray.ang.AppConfig import com.v2ray.ang.core.CoreConfigManager -import com.v2ray.ang.core.CoreNativeManager +import com.v2ray.ang.dto.ProbePlan import com.v2ray.ang.dto.RealPingEvent import com.v2ray.ang.enums.EConfigType import com.v2ray.ang.extension.isComplexType @@ -10,127 +12,256 @@ import com.v2ray.ang.extension.isNotNullEmpty import com.v2ray.ang.handler.MmkvManager import com.v2ray.ang.handler.SettingsManager import com.v2ray.ang.handler.SpeedtestManager +import com.v2ray.ang.util.JsonUtil +import com.v2ray.ang.util.LogUtil import kotlinx.coroutines.CancellationException import kotlinx.coroutines.CoroutineName import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.SupervisorJob -import kotlinx.coroutines.asCoroutineDispatcher -import kotlinx.coroutines.isActive +import kotlinx.coroutines.currentCoroutineContext +import kotlinx.coroutines.ensureActive import kotlinx.coroutines.joinAll import kotlinx.coroutines.launch -import java.util.concurrent.Executors -import java.util.concurrent.atomic.AtomicInteger +import libv2ray.Libv2ray +import libv2ray.ProbeHandler -/** - * Worker that runs a batch of real-ping tests independently. - * Each batch owns its own CoroutineScope/dispatcher and can be cancelled separately. - */ +/** Runs one progressively reported delay-test batch through one native core. */ class RealPingWorkerService( private val context: Context, - private val guids: List, + guids: List, private val onlyTcp: Boolean = false, - private val onEvent: (RealPingEvent) -> Unit = {} + private val onEvent: (RealPingEvent) -> Unit = {}, ) { + private val guids = guids.distinct() private val job = SupervisorJob() - private val concurrency = SettingsManager.getRealPingConcurrency() - private val dispatcher = Executors.newFixedThreadPool(if (onlyTcp) concurrency * 2 else concurrency).asCoroutineDispatcher() - private val scope = CoroutineScope(job + dispatcher + CoroutineName("RealPingBatchWorker")) - - private val runningCount = AtomicInteger(0) - private val totalCount = AtomicInteger(0) + private val scope = CoroutineScope(job + Dispatchers.IO + CoroutineName("ProbeBatch")) + private val controller = Libv2ray.newProbeController() + @Volatile + private var finished = false + private val emittedDelays = mutableMapOf() + private var completedWorkUnits = 0 + private var totalWorkUnits = guids.size + private val remainingWorkUnits = guids.associateWith { 1 }.toMutableMap() + private var lastProgressAt = 0L fun start() { - val jobs = guids.map { guid -> - totalCount.incrementAndGet() - scope.launch { - runningCount.incrementAndGet() - try { - val result = if (onlyTcp) startTcping(guid) else startRealPing(guid) - if (scope.isActive) { - onEvent(RealPingEvent.Result(guid, result)) - } - } catch (_: Throwable) { - // ignore - } finally { - val count = totalCount.decrementAndGet() - val left = runningCount.decrementAndGet() - if (scope.isActive) { - onEvent(RealPingEvent.Progress("$left / $count")) - } + if (onlyTcp) { + startTcpBatch() + return + } + scope.launch { + try { + val plan = CoreConfigManager.getProbePlan(context, guids) + val probeCount = plan.profiles.sumOf { it.outboundTags.size } + setWorkUnits(plan) + val concurrency = SettingsManager.getRealPingConcurrency() + if (plan.profiles.isNotEmpty()) { + LogUtil.i( + AppConfig.TAG, + "Starting $probeCount real-delay probes for ${plan.profiles.size} profiles with limit $concurrency", + ) + } + runPlan(plan, concurrency) + failPending() + finish("0") + } catch (_: CancellationException) { + finish("-1") + } catch (error: Throwable) { + if (!finished) { + LogUtil.e(AppConfig.TAG, "Probe batch failed", error) + failPending() + finish("-1") } } } + } + private fun startTcpBatch() { + val dispatcher = Dispatchers.IO.limitedParallelism(SettingsManager.getRealPingConcurrency()) + val jobs = guids.map { guid -> + scope.launch(dispatcher) { + emitResult(guid, safelyProbe(guid, ::startTcping)) + completeWork(guid, profileCompleted = true) + } + } scope.launch { try { - joinAll(*jobs.toTypedArray()) - if (isActive) { - onEvent(RealPingEvent.Finish("0")) - } + jobs.joinAll() + finish("0") } catch (_: CancellationException) { - // If cancelled, don't send finish event to avoid confusion - } finally { - close() + finish("-1") } } } fun cancel() { + controller.cancel() job.cancel() + finish("-1") + } + + private suspend fun runPlan(plan: ProbePlan, concurrency: Int) { + plan.failedGuids.forEach { guid -> + emitResult(guid, -1L) + completeWork(guid, profileCompleted = true) + } + probeBatch(plan, concurrency) + probeIndividually(plan.individualGuids) } - private fun close() { + /** Each fallback needs its own Xray instance, so these cannot overlap safely. */ + private suspend fun probeIndividually(individualGuids: List) { + individualGuids.forEach { guid -> + currentCoroutineContext().ensureActive() + emitResult(guid, safelyProbe(guid, ::startRealPing)) + completeWork(guid, profileCompleted = true) + } + } + + private suspend fun probeBatch(plan: ProbePlan, concurrency: Int) { + if (plan.profiles.isEmpty()) return try { - dispatcher.close() - } catch (_: Throwable) { - // ignore + controller.probe( + plan.content, + JsonUtil.toJson(plan.profiles), + concurrency, + object : ProbeHandler { + override fun onProbeResult( + groupID: String?, + delay: Long, + completed: Boolean, + ) { + val guid = groupID ?: return + emitResult(guid, delay) + completeWork(guid, profileCompleted = completed) + } + }, + ) + } catch (error: CancellationException) { + throw error + } catch (error: Exception) { + currentCoroutineContext().ensureActive() + val retryGuids = plan.profiles.map { it.guid }.filter(::isPending) + LogUtil.w( + AppConfig.TAG, + "Shared probe core rejected ${retryGuids.size} profiles; isolating the failing profile", + error, + ) + retryProbeGuids(retryGuids, concurrency) } } - private fun startRealPing(guid: String): Long { - val retFailure = -1L - - val config = MmkvManager.decodeServerConfig(guid) ?: return retFailure - if (!config.configType.isComplexType() - && config.configType != EConfigType.HYSTERIA2 - && config.configType != EConfigType.WIREGUARD - && config.alpn?.startsWith("h3") != true - && config.server.isNotNullEmpty() - && config.serverPort?.toIntOrNull() != null - ) { - val url = config.server.orEmpty() - val port = config.serverPort.orEmpty().toInt() - val tcpTime = SpeedtestManager.socketConnectTime(url, port, 1000) - if (tcpTime <= -1L) { - return retFailure + /** Binary isolation keeps one malformed Xray config from degrading every valid profile. */ + private suspend fun retryProbeGuids(retryGuids: List, concurrency: Int) { + currentCoroutineContext().ensureActive() + val activeGuids = retryGuids.filter(::isPending) + if (activeGuids.isEmpty()) return + if (activeGuids.size == 1) { + probeIndividually(activeGuids) + return + } + val halves = activeGuids.chunked((activeGuids.size + 1) / 2) + halves.forEach { half -> + currentCoroutineContext().ensureActive() + val retryPlan = try { + CoreConfigManager.getProbePlan(context, half) + } catch (error: Exception) { + LogUtil.w(AppConfig.TAG, "Failed to rebuild ${half.size} isolated probe profiles", error) + retryProbeGuids(half, concurrency) + return@forEach } + runPlan(retryPlan, concurrency) + } + } + + @Synchronized + private fun failPending() { + remainingWorkUnits.filterValues { it > 0 }.keys.toList().forEach { guid -> + emitResult(guid, emittedDelays[guid] ?: -1L) + completeWork(guid, profileCompleted = true) } + } + + @Synchronized + private fun isPending(guid: String): Boolean = remainingWorkUnits[guid]?.let { it > 0 } == true - val configResult = CoreConfigManager.getV2rayConfig4Speedtest(context, guid) - if (!configResult.status) { - return retFailure + @Synchronized + private fun emitResult(guid: String, delay: Long) { + if (finished) return + if (emittedDelays[guid] != delay) { + emittedDelays[guid] = delay + onEvent(RealPingEvent.Result(guid, delay)) } - return CoreNativeManager.measureOutboundDelay(configResult.content, SettingsManager.getDelayTestUrl()) + } + + @Synchronized + private fun setWorkUnits(plan: ProbePlan) { + remainingWorkUnits.clear() + guids.forEach { remainingWorkUnits[it] = 1 } + plan.profiles.forEach { profile -> + remainingWorkUnits[profile.guid] = profile.outboundTags.size.coerceAtLeast(1) + } + totalWorkUnits = remainingWorkUnits.values.sum().coerceAtLeast(1) + completedWorkUnits = 0 + lastProgressAt = 0L + emitProgress(force = true) + } + + @Synchronized + private fun completeWork(guid: String, profileCompleted: Boolean) { + val remaining = remainingWorkUnits[guid] ?: return + if (remaining <= 0) return + val completed = if (profileCompleted) remaining else 1 + remainingWorkUnits[guid] = remaining - completed + completedWorkUnits = (completedWorkUnits + completed).coerceAtMost(totalWorkUnits) + emitProgress(force = completedWorkUnits == totalWorkUnits) + } + + private fun emitProgress(force: Boolean) { + val now = SystemClock.elapsedRealtime() + if (!force && now - lastProgressAt < PROGRESS_UPDATE_INTERVAL_MS) return + lastProgressAt = now + onEvent(RealPingEvent.Progress("$completedWorkUnits / $totalWorkUnits")) + } + + @Synchronized + private fun finish(status: String) { + if (finished) return + finished = true + onEvent(RealPingEvent.Finish(status)) } private fun startTcping(guid: String): Long { - val retFailure = -1L - - val config = MmkvManager.decodeServerConfig(guid) ?: return retFailure - if (!config.configType.isComplexType() - && config.configType != EConfigType.HYSTERIA2 - && config.configType != EConfigType.WIREGUARD - && config.alpn?.startsWith("h3") != true - && config.server.isNotNullEmpty() - && config.serverPort?.toIntOrNull() != null + val config = MmkvManager.decodeServerConfig(guid) ?: return -1L + if (!config.configType.isComplexType() && + config.configType != EConfigType.HYSTERIA2 && + config.configType != EConfigType.WIREGUARD && + config.alpn?.startsWith("h3") != true && + config.server.isNotNullEmpty() && + config.serverPort?.toIntOrNull() != null ) { - val url = config.server.orEmpty() - val port = config.serverPort.orEmpty().toInt() - val tcpTime = SpeedtestManager.socketConnectTime(url, port, 1000) - - return tcpTime + return SpeedtestManager.socketConnectTime(config.server.orEmpty(), config.serverPort.orEmpty().toInt(), 1000) } + return -1L + } + + private fun startRealPing(guid: String): Long { + val configResult = CoreConfigManager.getV2rayConfig4RealDelay(context, guid) + if (!configResult.status) return -1L + return controller.measureDelay(configResult.content, SettingsManager.getDelayTestUrl()) + } + + private fun safelyProbe(guid: String, probe: (String) -> Long): Long = try { + probe(guid) + } catch (error: CancellationException) { + throw error + } catch (error: Exception) { + LogUtil.e(AppConfig.TAG, "Probe failed for $guid", error) + -1L + } - return retFailure + private companion object { + const val PROGRESS_UPDATE_INTERVAL_MS = 100L } } diff --git a/V2rayNG/app/src/main/java/com/v2ray/ang/service/SubscriptionUpdateService.kt b/V2rayNG/app/src/main/java/com/v2ray/ang/service/SubscriptionUpdateService.kt index 88604bf2a4..f8b6a47627 100644 --- a/V2rayNG/app/src/main/java/com/v2ray/ang/service/SubscriptionUpdateService.kt +++ b/V2rayNG/app/src/main/java/com/v2ray/ang/service/SubscriptionUpdateService.kt @@ -42,6 +42,8 @@ class SubscriptionUpdateService : Service() { private val activeWorkers = Collections.synchronizedList(mutableListOf()) private val updateSemaphore = Semaphore(2) + // Downloads may overlap, but native probe batches in this process may not. + private val probeSemaphore = Semaphore(1) override fun onCreate() { super.onCreate() @@ -131,9 +133,11 @@ class SubscriptionUpdateService : Service() { } if (MmkvManager.decodeSettingsBool(AppConfig.PREF_AUTO_TEST_AFTER_UPDATE_SUBSCRIPTION, false)) { - testSubscriptionServers(sub) + val testCompleted = probeSemaphore.withPermit { + testSubscriptionServers(sub) + } - if (MmkvManager.decodeSettingsBool(AppConfig.PREF_AUTO_REMOVE_INVALID_AFTER_TEST, false)) { + if (testCompleted && MmkvManager.decodeSettingsBool(AppConfig.PREF_AUTO_REMOVE_INVALID_AFTER_TEST, false)) { LogUtil.i(AppConfig.TAG, "SubscriptionUpdateService: removing invalid servers for ${subItem.remarks}") showNotification( context = this, @@ -142,7 +146,7 @@ class SubscriptionUpdateService : Service() { ) AngConfigManager.removeInvalidServer(subId) } - if (MmkvManager.decodeSettingsBool(AppConfig.PREF_AUTO_SORT_AFTER_TEST, false)) { + if (testCompleted && MmkvManager.decodeSettingsBool(AppConfig.PREF_AUTO_SORT_AFTER_TEST, false)) { LogUtil.i(AppConfig.TAG, "SubscriptionUpdateService: sorting servers for ${subItem.remarks}") showNotification( context = this, @@ -156,7 +160,7 @@ class SubscriptionUpdateService : Service() { LogUtil.i(AppConfig.TAG, "SubscriptionUpdateService: Finished ${subItem.remarks}") } - private suspend fun testSubscriptionServers(sub: SubscriptionCache) { + private suspend fun testSubscriptionServers(sub: SubscriptionCache): Boolean { val subId = sub.guid LogUtil.i(AppConfig.TAG, "SubscriptionUpdateService: starting test phase for ${sub.subscription.remarks}") showNotification( @@ -166,27 +170,28 @@ class SubscriptionUpdateService : Service() { ) val guids = MmkvManager.decodeServerList(subId) - if (guids.isNotEmpty()) { - val deferred = CompletableDeferred() - lateinit var worker: RealPingWorkerService - worker = RealPingWorkerService( - context = this, - guids = guids, - onEvent = { event -> - handleWorkerEvent(event, sub.subscription.remarks) { - activeWorkers.remove(worker) - deferred.complete(Unit) - } + if (guids.isEmpty()) return true + + val deferred = CompletableDeferred() + lateinit var worker: RealPingWorkerService + worker = RealPingWorkerService( + context = this, + guids = guids, + onEvent = { event -> + handleWorkerEvent(event, sub.subscription.remarks) { completed -> + activeWorkers.remove(worker) + deferred.complete(completed) } - ) - activeWorkers.add(worker) - worker.start() - deferred.await() - LogUtil.i(AppConfig.TAG, "SubscriptionUpdateService: test phase finished for ${sub.subscription.remarks}") - } + }, + ) + activeWorkers.add(worker) + worker.start() + val completed = deferred.await() + LogUtil.i(AppConfig.TAG, "SubscriptionUpdateService: test phase finished for ${sub.subscription.remarks}") + return completed } - private fun handleWorkerEvent(event: RealPingEvent, remarks: String, onWorkerDone: () -> Unit) { + private fun handleWorkerEvent(event: RealPingEvent, remarks: String, onWorkerDone: (Boolean) -> Unit) { when (event) { is RealPingEvent.Progress -> { val notificationText = getString( @@ -207,7 +212,7 @@ class SubscriptionUpdateService : Service() { } is RealPingEvent.Finish -> { - onWorkerDone() + onWorkerDone(event.status == "0") } } } diff --git a/V2rayNG/app/src/main/java/com/v2ray/ang/ui/main/MainRepository.kt b/V2rayNG/app/src/main/java/com/v2ray/ang/ui/main/MainRepository.kt index e272f747b8..12ab44fa2c 100644 --- a/V2rayNG/app/src/main/java/com/v2ray/ang/ui/main/MainRepository.kt +++ b/V2rayNG/app/src/main/java/com/v2ray/ang/ui/main/MainRepository.kt @@ -9,6 +9,7 @@ import com.v2ray.ang.AngApplication import com.v2ray.ang.AppConfig import com.v2ray.ang.R import com.v2ray.ang.dto.ConnectionTestResult +import com.v2ray.ang.dto.RealPingResult import com.v2ray.ang.dto.SubscriptionUpdateResult import com.v2ray.ang.dto.TestServiceMessage import com.v2ray.ang.dto.entities.ProfileItem @@ -41,12 +42,17 @@ class MainRepository( private val _mainServiceEvent = MutableSharedFlow( replay = 0, - extraBufferCapacity = 64, + // Absorb large result bursts before the ViewModel coalesces UI updates. + extraBufferCapacity = SERVICE_EVENT_BUFFER_CAPACITY, onBufferOverflow = BufferOverflow.DROP_OLDEST ) override val mainServiceEvent: SharedFlow = _mainServiceEvent.asSharedFlow() + private companion object { + const val SERVICE_EVENT_BUFFER_CAPACITY = 2048 + } + private val serviceReceiver = object : BroadcastReceiver() { override fun onReceive(context: Context?, intent: Intent?) { val safeIntent = intent ?: return @@ -61,7 +67,9 @@ class MainRepository( .serializable("content") ?.let { MainServiceEvent.MeasureDelayResult(it) } - AppConfig.MSG_MEASURE_CONFIG_SUCCESS -> MainServiceEvent.MeasureConfigSuccess + AppConfig.MSG_MEASURE_CONFIG_SUCCESS -> safeIntent + .serializable("content") + ?.let(MainServiceEvent::MeasureConfigSuccess) AppConfig.MSG_MEASURE_CONFIG_NOTIFY -> MainServiceEvent.MeasureConfigNotify( safeIntent.getStringExtra("content").orEmpty() ) diff --git a/V2rayNG/app/src/main/java/com/v2ray/ang/ui/main/MainServiceEvent.kt b/V2rayNG/app/src/main/java/com/v2ray/ang/ui/main/MainServiceEvent.kt index 9da9eff1a8..506066c9bb 100644 --- a/V2rayNG/app/src/main/java/com/v2ray/ang/ui/main/MainServiceEvent.kt +++ b/V2rayNG/app/src/main/java/com/v2ray/ang/ui/main/MainServiceEvent.kt @@ -1,6 +1,7 @@ package com.v2ray.ang.ui.main import com.v2ray.ang.dto.ConnectionTestResult +import com.v2ray.ang.dto.RealPingResult sealed class MainServiceEvent { data object StateRunning : MainServiceEvent() @@ -9,7 +10,7 @@ sealed class MainServiceEvent { data object StateStartFailure : MainServiceEvent() data object StateStopSuccess : MainServiceEvent() data class MeasureDelayResult(val result: ConnectionTestResult) : MainServiceEvent() - data object MeasureConfigSuccess : MainServiceEvent() + data class MeasureConfigSuccess(val result: RealPingResult) : MainServiceEvent() data class MeasureConfigNotify(val progress: String) : MainServiceEvent() data class MeasureConfigFinish(val finishedCount: String?) : MainServiceEvent() } diff --git a/V2rayNG/app/src/main/java/com/v2ray/ang/ui/main/MainViewModel.kt b/V2rayNG/app/src/main/java/com/v2ray/ang/ui/main/MainViewModel.kt index 83f0814d0c..cd6d757c06 100644 --- a/V2rayNG/app/src/main/java/com/v2ray/ang/ui/main/MainViewModel.kt +++ b/V2rayNG/app/src/main/java/com/v2ray/ang/ui/main/MainViewModel.kt @@ -9,6 +9,7 @@ import com.v2ray.ang.R import com.v2ray.ang.dto.ConnectionTestResult import com.v2ray.ang.dto.GroupMapItem import com.v2ray.ang.dto.LocateTarget +import com.v2ray.ang.dto.RealPingResult import com.v2ray.ang.dto.TestServiceMessage import com.v2ray.ang.dto.entities.ProfileItem import com.v2ray.ang.dto.entities.ServersCache @@ -73,6 +74,8 @@ class MainViewModel( private var preloadJob: Job? = null private var selectedGroupLoadJob: Job? = null private var reloadJob: Job? = null + private var testResultFlushJob: Job? = null + private val pendingTestResults = linkedMapOf() @Volatile private var testingGroupId: String? = null @@ -112,22 +115,53 @@ class MainViewModel( _uiState.update { it.copy(status = MainStatus.ConnectionTest(event.result)) } } - MainServiceEvent.MeasureConfigSuccess -> { - viewModelScope.launch(ioDispatcher) { - val gid = testingGroupId ?: uiState.value.selectedGroupId - cacheMutex.withLock { groupDataCache.remove(gid) } - updateGroupUi(gid, loadGroup(gid, forceRefresh = true)) - } - } + is MainServiceEvent.MeasureConfigSuccess -> queueTestResult(event.result) is MainServiceEvent.MeasureConfigNotify -> { _uiState.update { it.copy(status = MainStatus.TestProgress(event.progress)) } } is MainServiceEvent.MeasureConfigFinish -> { - onTestsFinished() + val scheduledFlush = testResultFlushJob + testResultFlushJob = viewModelScope.launch { + scheduledFlush?.join() + flushPendingTestResults() + onTestsFinished() + } + } + } + } + + private fun queueTestResult(result: RealPingResult) { + pendingTestResults[result.guid] = result.delayMillis + if (testResultFlushJob?.isActive == true) return + testResultFlushJob = viewModelScope.launch { + while (pendingTestResults.isNotEmpty()) { + delay(TEST_RESULT_FLUSH_INTERVAL_MS) + flushPendingTestResults() + } + } + } + + private suspend fun flushPendingTestResults() { + if (pendingTestResults.isEmpty()) return + val updates = pendingTestResults.toMap() + pendingTestResults.clear() + val groupId = testingGroupId ?: uiState.value.selectedGroupId + val applyUpdates: (List) -> List = { servers -> + servers.map { server -> + val delayMillis = updates[server.guid] + if (delayMillis == null || delayMillis == server.testDelayMillis) { + server + } else { + server.copy(testDelayMillis = delayMillis) + } } } + cacheMutex.withLock { + groupDataCache[groupId]?.let { groupDataCache[groupId] = applyUpdates(it) } + } + mutableServersForGroup(groupId).update(applyUpdates) } internal fun formatStatus(status: MainStatus): String = when (status) { @@ -676,6 +710,9 @@ class MainViewModel( // ---------- Testing ---------- fun cancelAllPing() { dataSource.cancelAllPing() + testResultFlushJob?.cancel() + testResultFlushJob = null + pendingTestResults.clear() testingGroupId = null _uiState.update { it.copy( @@ -686,14 +723,22 @@ class MainViewModel( } fun testAllRealPing(onlyTcp: Boolean = false) { - dataSource.cancelAllPing() val groupId = uiState.value.selectedGroupId val servers = currentServers() + testResultFlushJob?.cancel() + testResultFlushJob = null + pendingTestResults.clear() dataSource.clearAllTestDelayResults(servers.map { it.guid }) if (servers.isEmpty()) { _uiState.update { it.copy(isTesting = false) } return } + mutableServersForGroup(groupId).update { current -> + current.map { server -> + if (server.testDelayMillis == 0L) server + else server.copy(testDelayMillis = 0L) + } + } testingGroupId = groupId _uiState.update { it.copy( @@ -774,6 +819,7 @@ class MainViewModel( selectedGroupLoadJob?.cancel() reloadJob?.cancel() filterJob?.cancel() + testResultFlushJob?.cancel() cancelAllPing() dataSource.close() super.onCleared() @@ -789,4 +835,8 @@ class MainViewModel( throw IllegalArgumentException("Unknown ViewModel class") } } + + private companion object { + const val TEST_RESULT_FLUSH_INTERVAL_MS = 500L + } }