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

Changed behaviour

Browse Source
This commit is contained in:
bskjon 2024-04-21 20:16:09 +02:00
parent fecd87c136
commit 7663f4d7e5
1 changed files with 27 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

43
apps/converter/src/main/kotlin/no/iktdev/mediaprocessing/converter/tasks/ConvertService.kt
View File

@ -15,6 +15,7 @@ import no.iktdev.mediaprocessing.shared.kafka.dto.events_result.ConvertWorkPerfo
import no.iktdev.mediaprocessing.shared.kafka.dto.events_result.ConvertWorkerRequest import no.iktdev.mediaprocessing.shared.kafka.dto.events_result.ConvertWorkerRequest
import no.iktdev.mediaprocessing.shared.kafka.dto.isSuccess import no.iktdev.mediaprocessing.shared.kafka.dto.isSuccess
import no.iktdev.mediaprocessing.shared.kafka.dto.Status import no.iktdev.mediaprocessing.shared.kafka.dto.Status
import no.iktdev.mediaprocessing.shared.kafka.dto.events_result.work.ProcesserExtractWorkPerformed
import org.springframework.beans.factory.annotation.Autowired import org.springframework.beans.factory.annotation.Autowired
import org.springframework.scheduling.annotation.EnableScheduling import org.springframework.scheduling.annotation.EnableScheduling
import org.springframework.scheduling.annotation.Scheduled import org.springframework.scheduling.annotation.Scheduled
@ -57,23 +58,34 @@ class ConvertService(@Autowired override var coordinator: ConverterCoordinator)
event: PersistentProcessDataMessage, event: PersistentProcessDataMessage,
events: List<PersistentProcessDataMessage> events: List<PersistentProcessDataMessage>
): MessageDataWrapper? { ): MessageDataWrapper? {
val convertEvent =
events.find { it.event == KafkaEvents.EventWorkConvertCreated && it.data is ConvertWorkerRequest } val waitsForEventId = if (event.event == KafkaEvents.EventWorkConvertCreated) {
if (convertEvent == null) { // Do convert check
// No convert here.. val convertRequest = event.data as ConvertWorkerRequest? ?: return null
return null convertRequest.requiresEventId
} else if (event.event == KafkaEvents.EventWorkExtractPerformed) {
if (event.data is ProcesserExtractWorkPerformed) event.data.derivedFromEventId else return null
} else null
val convertData = if (event.event == KafkaEvents.EventWorkConvertCreated) {
event.data as ConvertWorkerRequest? ?: return null
} else {
val convertEvent = events.find { it.referenceId == event.referenceId && it.event == KafkaEvents.EventWorkConvertCreated } ?: return null
convertEvent.data as ConvertWorkerRequest? ?: return null
} }
val convertRequest = convertEvent.data as ConvertWorkerRequest? ?: return null
val requiredEventId = convertRequest.requiresEventId
if (requiredEventId != null) {
if (waitsForEventId != null) {
// Requires the eventId to be defined as consumed // Requires the eventId to be defined as consumed
val requiredEventToBeCompleted = getRequiredExtractProcessForContinuation( val requiredEventToBeCompleted = getRequiredExtractProcessForContinuation(
referenceId = event.referenceId, referenceId = event.referenceId,
requiresEventId = requiredEventId requiresEventId = waitsForEventId
) )
if (requiredEventToBeCompleted == null) { if (requiredEventToBeCompleted == null) {
log.warn { "$requiredEventId extract event with eventId: $requiredEventId was not found" } /*log.info { "Sending ${event.eventId} @ ${event.referenceId} to deferred check" }
log.info { "Sending ${event.eventId} @ ${event.referenceId} to deferred check" }
val existing = scheduled_deferred_events[event.referenceId] val existing = scheduled_deferred_events[event.referenceId]
val newList = (existing ?: listOf()) + listOf( val newList = (existing ?: listOf()) + listOf(
DerivedProcessIterationHolder( DerivedProcessIterationHolder(
@ -81,7 +93,7 @@ class ConvertService(@Autowired override var coordinator: ConverterCoordinator)
event = convertEvent event = convertEvent
) )
) )
scheduled_deferred_events[event.referenceId] = newList scheduled_deferred_events[event.referenceId] = newList*/
return null return null
} }
@ -106,8 +118,7 @@ class ConvertService(@Autowired override var coordinator: ConverterCoordinator)
return null return null
} }
val payload = event.data as ConvertWorkerRequest val converter = Converter(referenceId = event.referenceId, eventId = event.eventId, data = convertData)
val converter = Converter(referenceId = event.referenceId, eventId = event.eventId, data = payload)
if (!converter.canRead()) { if (!converter.canRead()) {
// Make claim regardless but push to schedule // Make claim regardless but push to schedule
return ConvertWorkPerformed( return ConvertWorkPerformed(
@ -179,7 +190,7 @@ class ConvertService(@Autowired override var coordinator: ConverterCoordinator)
} }
val scheduled_deferred_events: MutableMap<String, List<DerivedProcessIterationHolder>> = mutableMapOf() /*val scheduled_deferred_events: MutableMap<String, List<DerivedProcessIterationHolder>> = mutableMapOf()
@Scheduled(fixedDelay = (300_000)) @Scheduled(fixedDelay = (300_000))
fun validatePresenceOfRequiredEvent() { fun validatePresenceOfRequiredEvent() {
val continueDeferral: MutableMap<String, List<DerivedProcessIterationHolder>> = mutableMapOf() val continueDeferral: MutableMap<String, List<DerivedProcessIterationHolder>> = mutableMapOf()
@ -217,6 +228,6 @@ class ConvertService(@Autowired override var coordinator: ConverterCoordinator)
scheduled_deferred_events.clear() scheduled_deferred_events.clear()
scheduled_deferred_events.putAll(continueDeferral) scheduled_deferred_events.putAll(continueDeferral)
} }*/
} }