iktdev/MediaProcessing
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Adjustments

Browse Source
This commit is contained in:
bskjon 2024-04-13 20:41:46 +02:00
parent 50b171df65
commit 66f35401bc
1 changed files with 7 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
apps/processer/src/main/kotlin/no/iktdev/mediaprocessing/processer/Coordinator.kt
View File

@ -97,11 +97,13 @@ class Coordinator(): CoordinatorBase<PersistentProcessDataMessage, PersistentEve
val existing = eventManager.getAllProcessEvents().map { it.eventId } val existing = eventManager.getAllProcessEvents().map { it.eventId }
val messages = eventManager.getEventsUncompleted() val messages = eventManager.getEventsUncompleted()
val validMessages = messages.filter { list -> list.any { it.event in processKafkaEvents } }
.filter { list -> val usableMessages = messages.filter { lists -> lists.any { it.event in processKafkaEvents } }
list.any { it.event == KafkaEvents.EventMediaProcessStarted } ||
(list.any { it.event == KafkaEvents.EventMediaWorkProceedPermitted } && list.any { it.event == KafkaEvents.EVENT_REQUEST_PROCESS_STARTED })
}.flatten().filter { it.event in processKafkaEvents } val validMessages = usableMessages.filter { lists -> lists.any { it.event == KafkaEvents.EventMediaProcessStarted } ||
(lists.any { it.event == KafkaEvents.EVENT_REQUEST_PROCESS_STARTED } && lists.any { it.event == KafkaEvents.EventMediaWorkProceedPermitted } )
}.flatten().filter { it.event in processKafkaEvents }
validMessages.filter { it.eventId !in existing }.forEach { validMessages.filter { it.eventId !in existing }.forEach {
eventManager.setProcessEvent(it.event, Message( eventManager.setProcessEvent(it.event, Message(