Flagship/abox-mgr
5
2
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(scene): preserve fixed gateway device identifiers, fix edit loading race condition and parallel abort strategy

Browse Source
This commit is contained in:
KremeCN
2026-08-25 18:00:25 +08:00
parent b48079564f
commit e7da708e37
9 changed files with 694 additions and 465 deletions
Download Patch File Download Diff File Expand all files Collapse all files
domain/src/main/kotlin/com/flagship/abox/manager/domain/DefaultSceneExecutor.kt
+42 -8
View File
@@ -4,6 +4,7 @@ import com.flagship.abox.manager.domain.infrared.SendVirtualRemoteAction
import com.flagship.abox.manager.domain.infrared.VirtualRemoteId
import java.util.concurrent.ConcurrentHashMap
import kotlinx.coroutines.CancellationException
import kotlinx.coroutines.Deferred
import kotlinx.coroutines.async
import kotlinx.coroutines.awaitAll
import kotlinx.coroutines.coroutineScope
@@ -36,15 +37,48 @@ class DefaultSceneExecutor(
return try {
val results: List<SceneActionExecutionResult> = if (scene.concurrencyMode == ConcurrencyMode.PARALLEL) {
coroutineScope {
scene.actions.map { action ->
async {
val actionStart = System.currentTimeMillis()
val execResult = executeAction(scene.profileId, action)
val duration = System.currentTimeMillis() - actionStart
execResult.copy(executionDurationMillis = duration)
if (scene.strategy == ExecutionStrategy.ABORT_ON_ERROR) {
coroutineScope {
lateinit var deferreds: List<Deferred<SceneActionExecutionResult>>
deferreds = scene.actions.mapIndexed { index, action ->
async {
val actionStart = System.currentTimeMillis()
val execResult = executeAction(scene.profileId, action)
val duration = System.currentTimeMillis() - actionStart
val enriched = execResult.copy(executionDurationMillis = duration)
if (enriched.outcome == SceneActionOutcome.FAILURE) {
deferreds.forEachIndexed { otherIndex, otherDeferred ->
if (otherIndex != index) {
otherDeferred.cancel()
}
}
}
enriched
}
}
}.awaitAll()
deferreds.mapIndexed { index, deferred ->
try {
deferred.await()
} catch (e: CancellationException) {
SceneActionExecutionResult(
action = scene.actions[index],
outcome = SceneActionOutcome.SKIPPED,
error = null,
)
}
}
}
} else {
coroutineScope {
scene.actions.map { action ->
async {
val actionStart = System.currentTimeMillis()
val execResult = executeAction(scene.profileId, action)
val duration = System.currentTimeMillis() - actionStart
execResult.copy(executionDurationMillis = duration)
}
}.awaitAll()
}
}
} else {
val sequentialResults = mutableListOf<SceneActionExecutionResult>()
domain/src/test/kotlin/com/flagship/abox/manager/domain/DefaultSceneExecutorTest.kt
+40
View File
@@ -308,5 +308,45 @@ class DefaultSceneExecutorTest {
)
executor.execute(scene)
}
@Test
fun parallelAbortOnErrorCancelsPendingActions() = runTest {
val socketGateway = TestSocketGateway(
response = DomainResult.Failure(DomainError.OperationFailed(detail = "Socket failed")),
)
val executor = DefaultSceneExecutor(
socketGateway = socketGateway,
infraredGateway = TestInfraredGateway(),
delayDurationProvider = {
kotlinx.coroutines.delay(1000)
},
)
val action1 = SceneAction.SocketPower(
id = SceneActionId("a1"),
profileId = profileId,
deviceName = DeviceName("socket-1"),
powerState = SocketPowerState.ON,
)
val action2 = SceneAction.Delay(
id = SceneActionId("a2"),
profileId = profileId,
durationSeconds = 10,
)
val scene = Scene(
id = SceneId("s-par-abort"),
profileId = profileId,
name = "Parallel Abort Scene",
concurrencyMode = ConcurrencyMode.PARALLEL,
strategy = ExecutionStrategy.ABORT_ON_ERROR,
actions = listOf(action1, action2),
)
val result = executor.execute(scene)
assertTrue(result is DomainResult.Success)
val executionResult = (result as DomainResult.Success).value
assertEquals(2, executionResult.results.size)
assertEquals(SceneActionOutcome.FAILURE, executionResult.results[0].outcome)
assertEquals(SceneActionOutcome.SKIPPED, executionResult.results[1].outcome)
}
}