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

Updated check

Browse Source
This commit is contained in:
bskjon 2024-07-01 19:27:43 +02:00
parent 8ccefc1843
commit 46b053eaf0
4 changed files with 26 additions and 27 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

28
apps/coordinator/src/main/kotlin/no/iktdev/mediaprocessing/coordinator/utils/TasksUtil.kt
View File

@ -1,16 +1,13 @@
package no.iktdev.mediaprocessing.coordinator.utils package no.iktdev.mediaprocessing.coordinator.utils
import no.iktdev.mediaprocessing.shared.common.persistance.PersistentMessage import no.iktdev.mediaprocessing.shared.common.persistance.*
import no.iktdev.mediaprocessing.shared.common.persistance.PersistentMessageHelper
import no.iktdev.mediaprocessing.shared.common.persistance.isOfEvent
import no.iktdev.mediaprocessing.shared.common.persistance.isSuccess
import no.iktdev.mediaprocessing.shared.common.task.Task import no.iktdev.mediaprocessing.shared.common.task.Task
import no.iktdev.mediaprocessing.shared.common.task.TaskType import no.iktdev.mediaprocessing.shared.common.task.TaskType
import no.iktdev.mediaprocessing.shared.kafka.core.KafkaEvents import no.iktdev.mediaprocessing.shared.kafka.core.KafkaEvents
fun isAwaitingPrecondition(tasks: List<TaskType>, events: List<PersistentMessage>): Boolean { fun isAwaitingPrecondition(tasks: List<TaskType>, events: List<PersistentMessage>): Boolean {
if (tasks.contains(TaskType.Encode)) { if (tasks.any { it == TaskType.Encode }) {
if (events.lastOrNull { it.isOfEvent( if (events.lastOrNull { it.isOfEvent(
KafkaEvents.EventMediaParameterEncodeCreated KafkaEvents.EventMediaParameterEncodeCreated
) } == null) { ) } == null) {
@ -18,23 +15,21 @@ fun isAwaitingPrecondition(tasks: List<TaskType>, events: List<PersistentMessage
} }
} }
if (tasks.contains(TaskType.Convert) && !tasks.contains(TaskType.Extract)) {
if (tasks.any { it == TaskType.Convert } && tasks.none {it == TaskType.Extract}) {
if (events.lastOrNull { it.isOfEvent( if (events.lastOrNull { it.isOfEvent(
KafkaEvents.EventWorkConvertCreated KafkaEvents.EventWorkConvertCreated
) } == null) { ) } == null) {
return true return true
} }
} } else if (tasks.any { it == TaskType.Convert }) {
val extractEvent = events.lastOrNull { it.isOfEvent(KafkaEvents.EventMediaParameterExtractCreated) }
if (tasks.contains(TaskType.Extract)) { if (extractEvent == null || extractEvent.isSuccess()) {
if (events.lastOrNull { it.isOfEvent(
KafkaEvents.EventMediaParameterExtractCreated
) } == null) {
return true return true
} }
} }
if (tasks.contains(TaskType.Convert)) { if (tasks.contains(TaskType.Extract)) {
if (events.lastOrNull { it.isOfEvent( if (events.lastOrNull { it.isOfEvent(
KafkaEvents.EventMediaParameterExtractCreated KafkaEvents.EventMediaParameterExtractCreated
) } == null) { ) } == null) {
@ -86,6 +81,11 @@ fun isAwaitingTask(task: TaskType, events: List<PersistentMessage>): Boolean {
trailingEvents.filter { it.isOfEvent(taskCreatedEvent) }.size != trailingEvents.filter { it.isOfEvent(taskCreatedEvent) }.size trailingEvents.filter { it.isOfEvent(taskCreatedEvent) }.size != trailingEvents.filter { it.isOfEvent(taskCreatedEvent) }.size
} }
TaskType.Convert -> { TaskType.Convert -> {
val extractEvents = events.findLast { it.isOfEvent(KafkaEvents.EventMediaParameterExtractCreated) }
if (extractEvents == null || extractEvents.isSkipped()) {
false
} else {
val taskCreatedEvent = KafkaEvents.EventWorkConvertCreated val taskCreatedEvent = KafkaEvents.EventWorkConvertCreated
val taskCompletedEvent = KafkaEvents.EventWorkConvertPerformed val taskCompletedEvent = KafkaEvents.EventWorkConvertPerformed
@ -99,7 +99,7 @@ fun isAwaitingTask(task: TaskType, events: List<PersistentMessage>): Boolean {
) )
} }
trailingEvents.filter { it.isOfEvent(taskCreatedEvent) }.size != trailingEvents.filter { it.isOfEvent(taskCreatedEvent) }.size trailingEvents.filter { it.isOfEvent(taskCreatedEvent) }.size != trailingEvents.filter { it.isOfEvent(taskCreatedEvent) }.size
}
} }
} }
} }

2
apps/coordinator/src/main/kotlin/no/iktdev/mediaprocessing/coordinator/tasks/input/watcher/FileWatcherQueue.kt → apps/coordinator/src/main/kotlin/no/iktdev/mediaprocessing/coordinator/watcher/FileWatcherQueue.kt
View File

@ -1,4 +1,4 @@
package no.iktdev.mediaprocessing.coordinator.tasks.input.watcher package no.iktdev.mediaprocessing.coordinator.watcher
import kotlinx.coroutines.channels.Channel import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.delay import kotlinx.coroutines.delay

2
apps/coordinator/src/main/kotlin/no/iktdev/mediaprocessing/coordinator/tasks/input/watcher/InputDirectoryWatcher.kt → apps/coordinator/src/main/kotlin/no/iktdev/mediaprocessing/coordinator/watcher/InputDirectoryWatcher.kt
View File

@ -1,4 +1,4 @@
package no.iktdev.mediaprocessing.coordinator.tasks.input.watcher package no.iktdev.mediaprocessing.coordinator.watcher
import dev.vishna.watchservice.KWatchEvent.Kind.Deleted import dev.vishna.watchservice.KWatchEvent.Kind.Deleted
import dev.vishna.watchservice.KWatchEvent.Kind.Initialized import dev.vishna.watchservice.KWatchEvent.Kind.Initialized

1
shared/kafka/src/main/kotlin/no/iktdev/mediaprocessing/shared/kafka/core/KafkaEvents.kt
View File

@ -28,7 +28,6 @@ enum class KafkaEvents(val event: String) {
EventWorkConvertPerformed ("event:work-convert:performed"), EventWorkConvertPerformed ("event:work-convert:performed"),
EventWorkDownloadCoverPerformed ("event:work-download-cover:performed"), EventWorkDownloadCoverPerformed ("event:work-download-cover:performed"),
EVENT_STORE_VIDEO_PERFORMED ("event:store-video:performed"), EVENT_STORE_VIDEO_PERFORMED ("event:store-video:performed"),
EVENT_STORE_SUBTITLE_PERFORMED ("event:store-subtitle:performed"), EVENT_STORE_SUBTITLE_PERFORMED ("event:store-subtitle:performed"),
EVENT_STORE_COVER_PERFORMED ("event:store-cover:performed"), EVENT_STORE_COVER_PERFORMED ("event:store-cover:performed"),