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

Move event

Browse Source
This commit is contained in:
bskjon 2025-02-22 15:45:33 +01:00
parent 2ea3a69bdd
commit 5c04c368d8
44 changed files with 409 additions and 271 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
apps/coordinator/src/main/kotlin/no/iktdev/mediaprocessing/coordinator/CoordinatorEventCoordinator.kt
View File

@ -64,10 +64,10 @@ class Coordinator(
}
fun permitWorkToProceedOn(referenceId: String, events: List<Event>, message: String) {
val defaultRequiredBy = listOf(Events.EventMediaParameterEncodeCreated, Events.EventMediaParameterExtractCreated)
val defaultRequiredBy = listOf(Events.ParameterEncodeCreated, Events.ParameterExtractCreated)
val eventToAttachTo = if (events.any { it.eventType in defaultRequiredBy }) {
events.findLast { it.eventType in defaultRequiredBy }
} else events.find { it.eventType == Events.EventMediaProcessStarted }
} else events.find { it.eventType == Events.ProcessStarted }
if (eventToAttachTo == null) {
log.error { "No event to attach permit to" }
return

4
apps/coordinator/src/main/kotlin/no/iktdev/mediaprocessing/coordinator/tasksV2/implementations/WorkTaskListener.kt
View File

@ -12,11 +12,11 @@ abstract class WorkTaskListener: CoordinatorEventListener() {
private val log = KotlinLogging.logger {}
fun canStart(incomingEvent: Event, events: List<Event>): Boolean {
val autoStart = events.find { it.eventType == Events.EventMediaProcessStarted }?.az<MediaProcessStartEvent>()?.data
val autoStart = events.find { it.eventType == Events.ProcessStarted }?.az<MediaProcessStartEvent>()?.data
if (autoStart == null) {
log.error { "Start event not found. Requiring permitt event" }
}
return if (incomingEvent.eventType == Events.EventMediaWorkProceedPermitted) {
return if (incomingEvent.eventType == Events.WorkProceedPermitted) {
return true
} else {
if (autoStart == null || autoStart.type == ProcessType.MANUAL) {

4
apps/coordinator/src/main/kotlin/no/iktdev/mediaprocessing/coordinator/tasksV2/listeners/BaseInfoFromFileTaskListener.kt
View File

@ -23,8 +23,8 @@ class BaseInfoFromFileTaskListener() : CoordinatorEventListener() {
val log = KotlinLogging.logger {}
override val produceEvent: Events = Events.EventMediaReadBaseInfoPerformed
override val listensForEvents: List<Events> = listOf(Events.EventMediaProcessStarted)
override val produceEvent: Events = Events.ReadBaseInfoPerformed
override val listensForEvents: List<Events> = listOf(Events.ProcessStarted)
override fun getProducerName(): String {
return this::class.java.simpleName

110
apps/coordinator/src/main/kotlin/no/iktdev/mediaprocessing/coordinator/tasksV2/listeners/CompletedTaskListener.kt
View File

@ -41,17 +41,18 @@ class CompletedTaskListener : CoordinatorEventListener() {
@Autowired
override var coordinator: Coordinator? = null
override val produceEvent: Events = Events.EventMediaProcessCompleted
override val produceEvent: Events = Events.ProcessCompleted
override val listensForEvents: List<Events> = listOf(
Events.EventWorkDownloadCoverPerformed,
Events.EventWorkConvertPerformed,
Events.EventWorkEncodePerformed,
Events.EventWorkExtractPerformed
Events.WorkDownloadCoverPerformed,
Events.WorkConvertPerformed,
Events.WorkEncodePerformed,
Events.WorkExtractPerformed,
Events.PersistContentPerformed
)
override fun isPrerequisitesFulfilled(incomingEvent: Event, events: List<Event>): Boolean {
val started = events.find { it.eventType == Events.EventMediaProcessStarted }?.az<MediaProcessStartEvent>()
val started = events.find { it.eventType == Events.ProcessStarted }?.az<MediaProcessStartEvent>()
if (started == null) {
log.info { "No Start event" }
return false
@ -78,9 +79,9 @@ class CompletedTaskListener : CoordinatorEventListener() {
}
fun getVideo(events: List<Event>): VideoDetails? {
val mediaInfo = events.find { it.eventType == Events.EventMediaReadOutNameAndType }
val mediaInfo = events.find { it.eventType == Events.ReadOutNameAndType }
?.az<MediaOutInformationConstructedEvent>()
val encoded = events.find { it.eventType == Events.EventWorkEncodePerformed }?.dataAs<EncodedData>()?.outputFile
val encoded = events.find { it.eventType == Events.WorkEncodePerformed }?.dataAs<EncodedData>()?.outputFile
if (encoded == null) {
log.warn { "No encode no video details!" }
return null
@ -101,8 +102,6 @@ class CompletedTaskListener : CoordinatorEventListener() {
return details
}
override fun shouldIProcessAndHandleEvent(incomingEvent: Event, events: List<Event>): Boolean {
val result = super.shouldIProcessAndHandleEvent(incomingEvent, events)
return result
@ -125,50 +124,64 @@ class CompletedTaskListener : CoordinatorEventListener() {
mediaInfo.fallbackCollection
} else mediaInfo.fallbackCollection
val mover = ContentCompletionMover(usableCollection, events)
val genreIdsForCatalog = ContentGenresStore.storeAndGetIds(mediaInfo.genres)
val newCoverPath = mover.moveCover()
ContentCatalogStore.storeCatalog(
title = mediaInfo.title,
collection = usableCollection,
type = mediaInfo.type,
cover = newCoverPath?.second?.let { dp -> File(dp).name },
genres = genreIdsForCatalog,
)?.also { cid ->
mediaInfo.summaries.forEach {
ContentMetadataStore.storeSummary(cid, it)
val persistedContent: PersistedContent? = events.find { it.eventType == Events.PersistContentPerformed }?.dataAs<PersistedContentEvent>()?.data
if (persistedContent == null) {
log.error { "PersistedContent is null! can't continue" }
return
}
val completedData = CompletedData(
eventIdsCollected = events.map { it.eventId() },
metadataStored = MetadataStored(
title = mediaInfo.title,
titles = mediaInfo.titles,
type = mediaInfo.type,
cover = persistedContent.cover?.storeDestinationFileName?.let { File(it).name },
collection = usableCollection,
summary = mediaInfo.summaries,
foundTitles = existingTitles,
genres = mediaInfo.genres,
genreIds = genreIdsForCatalog
)
)
completedData.metadataStored.let { meta ->
val catalogId = ContentCatalogStore.storeCatalog(
title = meta.title,
collection = meta.collection,
type = meta.type,
cover = meta.cover,
genres = meta.genreIds
)
catalogId?.let { id ->
meta.summary.forEach { summary ->
ContentMetadataStore.storeSummary(id, summary)
}
}
ContentTitleStore.store(mediaInfo.title, mediaInfo.titles)
ContentTitleStore.store(meta.title, meta.titles)
}
val newVideoPath = mover.moveVideo()
val videoInfo = getVideo(events)
if (videoInfo != null) {
assert(newVideoPath == null)
assert(persistedContent.video == null)
ContentCatalogStore.storeMedia(
title = mediaInfo.title,
collection = usableCollection,
type = mediaInfo.type,
title = completedData.metadataStored.title,
collection = completedData.metadataStored.collection,
type = completedData.metadataStored.type,
videoDetails = videoInfo
)
}
val newSubtitles = mover.moveSubtitles()
try {
newSubtitles?.let { subtitles ->
persistedContent.subtitles.let { subtitles ->
subtitles.map {
ContentSubtitleStore.storeSubtitles(
collection = usableCollection,
language = it.language,
destinationFile = File(it.destination)
collection = completedData.metadataStored.collection,
destinationFile = File(it.storeDestinationFileName)
)
}
}
@ -176,25 +189,18 @@ class CompletedTaskListener : CoordinatorEventListener() {
e.printStackTrace()
}
val summary = EventsSummaryMapping().map(events)
ProcessedFileStore.store(
mediaInfo.title,
events,
summary
EventsSummaryMapping().map(events)
)
if (!doNotProduceComplete) {
onProduceEvent(MediaProcessCompletedEvent(
metadata = event.makeDerivedEventInfo(EventStatus.Success, getProducerName()),
data = CompletedEventData(
eventIdsCollected = events.map { it.eventId() },
coverMoved = newCoverPath?.let { c -> CoverMoved(c.first, c.second) },
videoMoved = newVideoPath?.let { v -> VideoMoved(v.first, v.second) },
subtitlesMoved = newSubtitles?.map { s -> SubtitlesMoved(s.source, s.destination) } ?: emptyList()
)
data = completedData
))
} else {
log.warn { "Do not produce complete is enabled!" }
@ -217,23 +223,23 @@ class CompletedTaskListener : CoordinatorEventListener() {
private fun composeMediaInfo(events: List<Event>): ComposedMediaInfo? {
val baseInfo =
events.find { it.eventType == Events.EventMediaReadBaseInfoPerformed }?.az<BaseInfoEvent>()?.let {
events.find { it.eventType == Events.ReadBaseInfoPerformed }?.az<BaseInfoEvent>()?.let {
it.data
} ?: run {
log.info { "Cant find BaseInfoEvent on ${Events.EventMediaReadBaseInfoPerformed}" }
log.info { "Cant find BaseInfoEvent on ${Events.ReadBaseInfoPerformed}" }
return null
}
val metadataInfo =
events.find { it.eventType == Events.EventMediaMetadataSearchPerformed }?.az<MediaMetadataReceivedEvent>()?.data
events.find { it.eventType == Events.MetadataSearchPerformed }?.az<MediaMetadataReceivedEvent>()?.data
?: run {
log.info { "Cant find MediaMetadataReceivedEvent on ${Events.EventMediaMetadataSearchPerformed}" }
log.info { "Cant find MediaMetadataReceivedEvent on ${Events.MetadataSearchPerformed}" }
null
}
val mediaInfo: MediaInfo = events.find { it.eventType == Events.EventMediaReadOutNameAndType }
val mediaInfo: MediaInfo = events.find { it.eventType == Events.ReadOutNameAndType }
?.az<MediaOutInformationConstructedEvent>()?.let {
it.data?.toValueObject()
} ?: run {
log.info { "Cant find MediaOutInformationConstructedEvent on ${Events.EventMediaReadOutNameAndType}" }
log.info { "Cant find MediaOutInformationConstructedEvent on ${Events.ReadOutNameAndType}" }
return null
}

10
apps/coordinator/src/main/kotlin/no/iktdev/mediaprocessing/coordinator/tasksV2/listeners/ConvertWorkTaskListener.kt
View File

@ -26,9 +26,9 @@ class ConvertWorkTaskListener: WorkTaskListener() {
@Autowired
override var coordinator: Coordinator? = null
override val produceEvent: Events = Events.EventWorkConvertCreated
override val produceEvent: Events = Events.WorkConvertCreated
override val listensForEvents: List<Events> = listOf(
Events.EventWorkExtractPerformed
Events.WorkExtractPerformed
)
override fun canProduceMultipleEvents(): Boolean {
@ -63,18 +63,18 @@ class ConvertWorkTaskListener: WorkTaskListener() {
var storeAsFile: String? = null
val file = if (event.eventType == Events.EventWorkExtractPerformed) {
val file = if (event.eventType == Events.WorkExtractPerformed) {
val foundEvent = event.az<ExtractWorkPerformedEvent>()?.data
language = foundEvent?.language
storeAsFile = foundEvent?.storeFileName
foundEvent?.outputFile
} else if (event.eventType == Events.EventMediaProcessStarted) {
} else if (event.eventType == Events.ProcessStarted) {
val startEvent = event.az<MediaProcessStartEvent>()?.data
if (startEvent?.operations?.isOnly(OperationEvents.CONVERT) == true) {
startEvent.file
} else null
} else {
events.find { it.eventType == Events.EventWorkExtractPerformed }
events.find { it.eventType == Events.WorkExtractPerformed }
?.az<ExtractWorkPerformedEvent>()?.data?.outputFile
}

9
apps/coordinator/src/main/kotlin/no/iktdev/mediaprocessing/coordinator/tasksV2/listeners/CoverDownloadTaskListener.kt
View File

@ -5,19 +5,14 @@ import mu.KotlinLogging
import no.iktdev.eventi.core.ConsumableEvent
import no.iktdev.eventi.core.WGson
import no.iktdev.eventi.data.EventStatus
import no.iktdev.eventi.implementations.EventCoordinator
import no.iktdev.exfl.using
import no.iktdev.mediaprocessing.coordinator.Coordinator
import no.iktdev.mediaprocessing.coordinator.CoordinatorEventListener
import no.iktdev.mediaprocessing.shared.common.DownloadClient
import no.iktdev.mediaprocessing.shared.common.SharedConfig
import no.iktdev.mediaprocessing.shared.common.contract.Events
import no.iktdev.mediaprocessing.shared.common.contract.EventsListenerContract
import no.iktdev.mediaprocessing.shared.common.contract.EventsManagerContract
import no.iktdev.mediaprocessing.shared.common.contract.data.*
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.stereotype.Service
import java.io.File
@Service
class CoverDownloadTaskListener : CoordinatorEventListener() {
@ -29,8 +24,8 @@ class CoverDownloadTaskListener : CoordinatorEventListener() {
@Autowired
override var coordinator: Coordinator? = null
override val produceEvent: Events = Events.EventWorkDownloadCoverPerformed
override val listensForEvents: List<Events> = listOf(Events.EventMediaReadOutCover)
override val produceEvent: Events = Events.WorkDownloadCoverPerformed
override val listensForEvents: List<Events> = listOf(Events.ReadOutCover)
override fun onEventsReceived(incomingEvent: ConsumableEvent<Event>, events: List<Event>) {
val event = incomingEvent.consume()
if (event == null) {

12
apps/coordinator/src/main/kotlin/no/iktdev/mediaprocessing/coordinator/tasksV2/listeners/CoverFromMetadataTaskListener.kt
View File

@ -25,13 +25,13 @@ class CoverFromMetadataTaskListener: CoordinatorEventListener() {
@Autowired
override var coordinator: Coordinator? = null
override val produceEvent: Events = Events.EventMediaReadOutCover
override val produceEvent: Events = Events.ReadOutCover
override val listensForEvents: List<Events> = listOf(
Events.EventMediaMetadataSearchPerformed
Events.MetadataSearchPerformed
)
override fun isPrerequisitesFulfilled(incomingEvent: Event, events: List<Event>): Boolean {
return (events.any { it.eventType == Events.EventMediaReadOutNameAndType && it.isSuccessful() })
return (events.any { it.eventType == Events.ReadOutNameAndType && it.isSuccessful() })
}
override fun shouldIProcessAndHandleEvent(incomingEvent: Event, events: List<Event>): Boolean {
@ -52,17 +52,17 @@ class CoverFromMetadataTaskListener: CoordinatorEventListener() {
}
active = true
val baseInfo = events.find { it.eventType == Events.EventMediaReadBaseInfoPerformed }?.az<BaseInfoEvent>()?.data
val baseInfo = events.find { it.eventType == Events.ReadBaseInfoPerformed }?.az<BaseInfoEvent>()?.data
if (baseInfo == null) {
log.info { "No base info" }
active = false
return
}
val metadataEvent = if (event.eventType == Events.EventMediaMetadataSearchPerformed) event else events.findLast { it.eventType == Events.EventMediaMetadataSearchPerformed }
val metadataEvent = if (event.eventType == Events.MetadataSearchPerformed) event else events.findLast { it.eventType == Events.MetadataSearchPerformed }
val metadata = metadataEvent?.az<MediaMetadataReceivedEvent>()?.data
?: return
val mediaOutInfo = events.find { it.eventType == Events.EventMediaReadOutNameAndType }?.az<MediaOutInformationConstructedEvent>()?.data
val mediaOutInfo = events.find { it.eventType == Events.ReadOutNameAndType }?.az<MediaOutInformationConstructedEvent>()?.data
if (mediaOutInfo == null) {
log.info { "No Media out info" }
active = false

12
apps/coordinator/src/main/kotlin/no/iktdev/mediaprocessing/coordinator/tasksV2/listeners/EncodeWorkArgumentsTaskListener.kt
View File

@ -25,11 +25,11 @@ class EncodeWorkArgumentsTaskListener: CoordinatorEventListener() {
@Autowired
override var coordinator: Coordinator? = null
override val produceEvent: Events = Events.EventMediaParameterEncodeCreated
override val produceEvent: Events = Events.ParameterEncodeCreated
override val listensForEvents: List<Events> = listOf(
Events.EventMediaParseStreamPerformed,
Events.EventMediaReadOutNameAndType
Events.ParseStreamPerformed,
Events.ReadOutNameAndType
)
val preference = Preference.getPreference()
@ -46,7 +46,7 @@ class EncodeWorkArgumentsTaskListener: CoordinatorEventListener() {
return
}
active = true
val started = events.find { it.eventType == Events.EventMediaProcessStarted }?.az<MediaProcessStartEvent>()
val started = events.find { it.eventType == Events.ProcessStarted }?.az<MediaProcessStartEvent>()
if (started == null) {
active = false
return
@ -55,13 +55,13 @@ class EncodeWorkArgumentsTaskListener: CoordinatorEventListener() {
active = false
return
}
val streams = events.find { it.eventType == Events.EventMediaParseStreamPerformed }?.az<MediaFileStreamsParsedEvent>()?.data
val streams = events.find { it.eventType == Events.ParseStreamPerformed }?.az<MediaFileStreamsParsedEvent>()?.data
if (streams == null) {
active = false
return
}
val mediaInfo = events.find { it.eventType == Events.EventMediaReadOutNameAndType }?.az<MediaOutInformationConstructedEvent>()
val mediaInfo = events.find { it.eventType == Events.ReadOutNameAndType }?.az<MediaOutInformationConstructedEvent>()
if (mediaInfo?.data == null) {
active = false
return

12
apps/coordinator/src/main/kotlin/no/iktdev/mediaprocessing/coordinator/tasksV2/listeners/EncodeWorkTaskListener.kt
View File

@ -7,13 +7,11 @@ import no.iktdev.eventi.data.EventStatus
import no.iktdev.eventi.data.derivedFromEventId
import no.iktdev.eventi.data.eventId
import no.iktdev.eventi.data.referenceId
import no.iktdev.eventi.implementations.EventCoordinator
import no.iktdev.mediaprocessing.coordinator.Coordinator
import no.iktdev.mediaprocessing.coordinator.taskManager
import no.iktdev.mediaprocessing.coordinator.tasksV2.implementations.WorkTaskListener
import no.iktdev.mediaprocessing.shared.common.task.TaskType
import no.iktdev.mediaprocessing.shared.common.contract.Events
import no.iktdev.mediaprocessing.shared.common.contract.EventsManagerContract
import no.iktdev.mediaprocessing.shared.common.contract.data.*
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.stereotype.Service
@ -28,10 +26,10 @@ class EncodeWorkTaskListener : WorkTaskListener() {
@Autowired
override var coordinator: Coordinator? = null
override val produceEvent: Events = Events.EventWorkEncodeCreated
override val produceEvent: Events = Events.WorkEncodeCreated
override val listensForEvents: List<Events> = listOf(
Events.EventMediaParameterEncodeCreated,
Events.EventMediaWorkProceedPermitted
Events.ParameterEncodeCreated,
Events.WorkProceedPermitted
)
override fun canProduceMultipleEvents(): Boolean {
@ -49,10 +47,10 @@ class EncodeWorkTaskListener : WorkTaskListener() {
return
}
val encodeArguments = if (event.eventType == Events.EventMediaParameterEncodeCreated) {
val encodeArguments = if (event.eventType == Events.ParameterEncodeCreated) {
event.az<EncodeArgumentCreatedEvent>()?.data
} else {
events.find { it.eventType == Events.EventMediaParameterEncodeCreated }
events.find { it.eventType == Events.ParameterEncodeCreated }
?.az<EncodeArgumentCreatedEvent>()?.data
}
if (encodeArguments == null) {

12
apps/coordinator/src/main/kotlin/no/iktdev/mediaprocessing/coordinator/tasksV2/listeners/ExtractWorkArgumentsTaskListener.kt
View File

@ -23,10 +23,10 @@ class ExtractWorkArgumentsTaskListener: CoordinatorEventListener() {
@Autowired
override var coordinator: Coordinator? = null
override val produceEvent: Events = Events.EventMediaParameterExtractCreated
override val produceEvent: Events = Events.ParameterExtractCreated
override val listensForEvents: List<Events> = listOf(
Events.EventMediaParseStreamPerformed,
Events.EventMediaReadOutNameAndType
Events.ParseStreamPerformed,
Events.ReadOutNameAndType
)
override fun shouldIProcessAndHandleEvent(incomingEvent: Event, events: List<Event>): Boolean {
@ -42,18 +42,18 @@ class ExtractWorkArgumentsTaskListener: CoordinatorEventListener() {
return
}
active = true
val started = events.find { it.eventType == Events.EventMediaProcessStarted }?.az<MediaProcessStartEvent>() ?: return
val started = events.find { it.eventType == Events.ProcessStarted }?.az<MediaProcessStartEvent>() ?: return
if (started.data == null || started.data?.operations?.contains(OperationEvents.EXTRACT) == false) {
active = false
return
}
val streams = events.find { it.eventType == Events.EventMediaParseStreamPerformed }?.az<MediaFileStreamsParsedEvent>()?.data
val streams = events.find { it.eventType == Events.ParseStreamPerformed }?.az<MediaFileStreamsParsedEvent>()?.data
if (streams == null) {
active = false
return
}
val mediaInfo = events.find { it.eventType == Events.EventMediaReadOutNameAndType }?.az<MediaOutInformationConstructedEvent>()
val mediaInfo = events.find { it.eventType == Events.ReadOutNameAndType }?.az<MediaOutInformationConstructedEvent>()
if (mediaInfo?.data == null) {
active = false
return

12
apps/coordinator/src/main/kotlin/no/iktdev/mediaprocessing/coordinator/tasksV2/listeners/ExtractWorkTaskListener.kt
View File

@ -7,13 +7,11 @@ import no.iktdev.eventi.data.EventStatus
import no.iktdev.eventi.data.derivedFromEventId
import no.iktdev.eventi.data.eventId
import no.iktdev.eventi.data.referenceId
import no.iktdev.eventi.implementations.EventCoordinator
import no.iktdev.mediaprocessing.coordinator.Coordinator
import no.iktdev.mediaprocessing.coordinator.taskManager
import no.iktdev.mediaprocessing.coordinator.tasksV2.implementations.WorkTaskListener
import no.iktdev.mediaprocessing.shared.common.task.TaskType
import no.iktdev.mediaprocessing.shared.common.contract.Events
import no.iktdev.mediaprocessing.shared.common.contract.EventsManagerContract
import no.iktdev.mediaprocessing.shared.common.contract.data.*
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.stereotype.Service
@ -28,10 +26,10 @@ class ExtractWorkTaskListener: WorkTaskListener() {
@Autowired
override var coordinator: Coordinator? = null
override val produceEvent: Events = Events.EventWorkExtractCreated
override val produceEvent: Events = Events.WorkExtractCreated
override val listensForEvents: List<Events> = listOf(
Events.EventMediaParameterExtractCreated,
Events.EventMediaWorkProceedPermitted
Events.ParameterExtractCreated,
Events.WorkProceedPermitted
)
override fun canProduceMultipleEvents(): Boolean {
@ -58,10 +56,10 @@ class ExtractWorkTaskListener: WorkTaskListener() {
return
}
val arguments = if (event.eventType == Events.EventMediaParameterExtractCreated) {
val arguments = if (event.eventType == Events.ParameterExtractCreated) {
event.az<ExtractArgumentCreatedEvent>()?.data
} else {
events.find { it.eventType == Events.EventMediaParameterExtractCreated }
events.find { it.eventType == Events.ParameterExtractCreated }
?.az<ExtractArgumentCreatedEvent>()?.data
}
if (arguments == null) {

10
apps/coordinator/src/main/kotlin/no/iktdev/mediaprocessing/coordinator/tasksV2/listeners/MediaOutInformationTaskListener.kt
View File

@ -4,7 +4,6 @@ import com.google.gson.JsonObject
import no.iktdev.eventi.core.ConsumableEvent
import no.iktdev.eventi.core.WGson
import no.iktdev.eventi.data.EventStatus
import no.iktdev.exfl.using
import no.iktdev.mediaprocessing.coordinator.Coordinator
import no.iktdev.mediaprocessing.coordinator.CoordinatorEventListener
import no.iktdev.mediaprocessing.coordinator.log
@ -19,7 +18,6 @@ import no.iktdev.mediaprocessing.shared.common.contract.data.pyMetadata
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.stereotype.Service
import java.io.FileFilter
import javax.naming.Name
@Service
@ -32,9 +30,9 @@ class MediaOutInformationTaskListener: CoordinatorEventListener() {
@Autowired
override var coordinator: Coordinator? = null
override val produceEvent: Events = Events.EventMediaReadOutNameAndType
override val produceEvent: Events = Events.ReadOutNameAndType
override val listensForEvents: List<Events> = listOf(
Events.EventMediaMetadataSearchPerformed
Events.MetadataSearchPerformed
)
override fun shouldIHandleFailedEvents(incomingEvent: Event): Boolean {
@ -50,9 +48,9 @@ class MediaOutInformationTaskListener: CoordinatorEventListener() {
active = true
val metadataResult = event.az<MediaMetadataReceivedEvent>()
val mediaBaseInfo = events.findLast { it.eventType == Events.EventMediaReadBaseInfoPerformed }?.az<BaseInfoEvent>()?.data
val mediaBaseInfo = events.findLast { it.eventType == Events.ReadBaseInfoPerformed }?.az<BaseInfoEvent>()?.data
if (mediaBaseInfo == null) {
log.error { "Required event ${Events.EventMediaReadBaseInfoPerformed} is not present" }
log.error { "Required event ${Events.ReadBaseInfoPerformed} is not present" }
coordinator?.produceNewEvent(
MediaOutInformationConstructedEvent(
metadata = event.makeDerivedEventInfo(EventStatus.Failed, getProducerName())

12
apps/coordinator/src/main/kotlin/no/iktdev/mediaprocessing/coordinator/tasksV2/listeners/MetadataWaitOrDefaultTaskListener.kt
View File

@ -40,10 +40,10 @@ class MetadataWaitOrDefaultTaskListener() : CoordinatorEventListener() {
val log = KotlinLogging.logger {}
override val produceEvent: Events = Events.EventMediaMetadataSearchPerformed
override val produceEvent: Events = Events.MetadataSearchPerformed
override val listensForEvents: List<Events> = listOf(
Events.EventMediaReadBaseInfoPerformed,
Events.EventMediaMetadataSearchPerformed
Events.ReadBaseInfoPerformed,
Events.MetadataSearchPerformed
)
@ -58,8 +58,8 @@ class MetadataWaitOrDefaultTaskListener() : CoordinatorEventListener() {
if (metadataTimeoutMinutes <= 0) {
return
}
val hasReadBaseInfo = events.any { it.eventType == Events.EventMediaReadBaseInfoPerformed && it.isSuccessful() }
val hasMetadataSearched = events.any { it.eventType == Events.EventMediaMetadataSearchPerformed }
val hasReadBaseInfo = events.any { it.eventType == Events.ReadBaseInfoPerformed && it.isSuccessful() }
val hasMetadataSearched = events.any { it.eventType == Events.MetadataSearchPerformed }
val hasPollerForMetadataEvent = waitingProcessesForMeta.containsKey(incomingEvent.metadata().referenceId)
if (!hasReadBaseInfo) {
@ -76,7 +76,7 @@ class MetadataWaitOrDefaultTaskListener() : CoordinatorEventListener() {
return
}
val baseInfo = events.find { it.eventType == Events.EventMediaReadBaseInfoPerformed}?.az<BaseInfoEvent>()?.data
val baseInfo = events.find { it.eventType == Events.ReadBaseInfoPerformed}?.az<BaseInfoEvent>()?.data
if (baseInfo == null) {
log.error { "BaseInfoEvent is null for referenceId: ${consumedIncoming.metadata.referenceId} on eventId: ${consumedIncoming.metadata.eventId}" }
return

8
apps/coordinator/src/main/kotlin/no/iktdev/mediaprocessing/coordinator/tasksV2/listeners/ParseMediaFileStreamsTaskListener.kt
View File

@ -7,15 +7,11 @@ import no.iktdev.eventi.core.ConsumableEvent
import no.iktdev.eventi.core.WGson
import no.iktdev.eventi.data.EventStatus
import no.iktdev.eventi.data.dataAs
import no.iktdev.eventi.implementations.EventCoordinator
import no.iktdev.mediaprocessing.coordinator.Coordinator
import no.iktdev.mediaprocessing.coordinator.CoordinatorEventListener
import no.iktdev.mediaprocessing.shared.common.contract.Events
import no.iktdev.mediaprocessing.shared.common.contract.EventsListenerContract
import no.iktdev.mediaprocessing.shared.common.contract.EventsManagerContract
import no.iktdev.mediaprocessing.shared.common.contract.data.Event
import no.iktdev.mediaprocessing.shared.common.contract.data.MediaFileStreamsParsedEvent
import no.iktdev.mediaprocessing.shared.common.contract.data.MediaFileStreamsReadEvent
import no.iktdev.mediaprocessing.shared.common.contract.ffmpeg.AudioStream
import no.iktdev.mediaprocessing.shared.common.contract.ffmpeg.ParsedMediaStreams
import no.iktdev.mediaprocessing.shared.common.contract.ffmpeg.SubtitleStream
@ -35,9 +31,9 @@ class ParseMediaFileStreamsTaskListener() : CoordinatorEventListener() {
override var coordinator: Coordinator? = null
override val produceEvent: Events = Events.EventMediaParseStreamPerformed
override val produceEvent: Events = Events.ParseStreamPerformed
override val listensForEvents: List<Events> = listOf(
Events.EventMediaReadStreamPerformed
Events.ReadStreamPerformed
)
override fun shouldIProcessAndHandleEvent(incomingEvent: Event, events: List<Event>): Boolean {

174
apps/coordinator/src/main/kotlin/no/iktdev/mediaprocessing/coordinator/tasksV2/listeners/PersistContentTaskListener.kt Normal file
View File

@ -0,0 +1,174 @@
package no.iktdev.mediaprocessing.coordinator.tasksV2.listeners
import mu.KotlinLogging
import no.iktdev.eventi.core.ConsumableEvent
import no.iktdev.eventi.data.*
import no.iktdev.mediaprocessing.coordinator.Coordinator
import no.iktdev.mediaprocessing.coordinator.CoordinatorEventListener
import no.iktdev.mediaprocessing.coordinator.tasksV2.mapping.store.*
import no.iktdev.mediaprocessing.coordinator.tasksV2.validator.CompletionValidator
import no.iktdev.mediaprocessing.shared.common.parsing.NameHelper
import no.iktdev.mediaprocessing.shared.common.contract.Events
import no.iktdev.mediaprocessing.shared.common.contract.data.*
import no.iktdev.mediaprocessing.shared.common.contract.reader.*
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.stereotype.Service
@Service
class PersistContentTaskListener : CoordinatorEventListener() {
val log = KotlinLogging.logger {}
var doNotProduceComplete = System.getenv("DISABLE_COMPLETE").toBoolean() ?: false
override fun getProducerName(): String {
return this::class.java.simpleName
}
override fun onReady() {
super.onReady()
if (doNotProduceComplete) {
log.warn { "DoNotProduceComplete is set!\n\tNo complete event will be triggered!\n\tTo enable production of complete vents, remove this line in your environment: \"DISABLE_COMPLETE\"" }
}
}
override fun shouldIHandleFailedEvents(incomingEvent: Event): Boolean {
return true
}
@Autowired
override var coordinator: Coordinator? = null
override val produceEvent: Events = Events.PersistContentPerformed
override val listensForEvents: List<Events> = listOf(
Events.WorkDownloadCoverPerformed,
Events.WorkConvertPerformed,
Events.WorkEncodePerformed,
Events.WorkExtractPerformed
)
override fun isPrerequisitesFulfilled(incomingEvent: Event, events: List<Event>): Boolean {
val started = events.find { it.eventType == Events.ProcessStarted }?.az<MediaProcessStartEvent>()
if (started == null) {
log.info { "No Start event" }
return false
}
val viableEvents = events.filter { it.isSuccessful() }
if (!CompletionValidator.req1(started, events)) {
return false
}
if (!CompletionValidator.req2(started.data?.operations ?: emptyList(), viableEvents)) {
return false
}
if (!CompletionValidator.req3(started.data?.operations ?: emptyList(), events)) {
return false
}
if (!CompletionValidator.req4(events)) {
return false
}
return super.isPrerequisitesFulfilled(incomingEvent, events)
}
override fun shouldIProcessAndHandleEvent(incomingEvent: Event, events: List<Event>): Boolean {
val result = super.shouldIProcessAndHandleEvent(incomingEvent, events)
return result
}
override fun onEventsReceived(incomingEvent: ConsumableEvent<Event>, events: List<Event>) {
val event = incomingEvent.consume() ?: return
active = true
val mediaInfo: ComposedMediaInfo = composeMediaInfo(events) ?: run {
log.error { "Unable to compose media info for ${event.referenceId()}" }
return
}
val existingTitles = ContentTitleStore.findMasterTitles(mediaInfo.titles)
val usableCollection: String = if (existingTitles.isNotEmpty())
ContentCatalogStore.getCollectionByTitleAndType(mediaInfo.type, existingTitles) ?: run {
log.warn { "Did not receive collection based on titles provided in list ${existingTitles.joinToString(",")}, falling back to fallbackCollection: ${mediaInfo.fallbackCollection}" }
mediaInfo.fallbackCollection
} else mediaInfo.fallbackCollection
val mover = ContentCompletionMover(usableCollection, events)
val newCoverPath = mover.moveCover()
val newVideoPath = mover.moveVideo()
val newSubtitles = mover.moveSubtitles()
val contentEvent = PersistedContent(
cover = newCoverPath?.let { PersistedItem(it.first, it.second) },
video = newVideoPath?.let { PersistedItem(it.first, it.second) },
subtitles = newSubtitles?.map { PersistedItem(it.source, it.destination) } ?: emptyList()
)
onProduceEvent(PersistedContentEvent(
metadata = event.makeDerivedEventInfo(EventStatus.Success, getProducerName()),
data = contentEvent
))
active = false
}
internal data class ComposedMediaInfo(
val title: String,
val fallbackCollection: String,
val titles: List<String>,
val type: String,
val summaries: List<SummaryInfo>,
val genres: List<String>
)
private fun composeMediaInfo(events: List<Event>): ComposedMediaInfo? {
val baseInfo =
events.find { it.eventType == Events.ReadBaseInfoPerformed }?.az<BaseInfoEvent>()?.let {
it.data
} ?: run {
log.info { "Cant find BaseInfoEvent on ${Events.ReadBaseInfoPerformed}" }
return null
}
val metadataInfo =
events.find { it.eventType == Events.MetadataSearchPerformed }?.az<MediaMetadataReceivedEvent>()?.data
?: run {
log.info { "Cant find MediaMetadataReceivedEvent on ${Events.MetadataSearchPerformed}" }
null
}
val mediaInfo: MediaInfo = events.find { it.eventType == Events.ReadOutNameAndType }
?.az<MediaOutInformationConstructedEvent>()?.let {
it.data?.toValueObject()
} ?: run {
log.info { "Cant find MediaOutInformationConstructedEvent on ${Events.ReadOutNameAndType}" }
return null
}
val summaries = metadataInfo?.summary?.filter { it.summary != null }
?.map { SummaryInfo(language = it.language, summary = it.summary!!) } ?: emptyList()
val titles: MutableList<String> = mutableListOf(mediaInfo.title)
metadataInfo?.let {
titles.addAll(it.altTitle)
titles.add(it.title)
titles.add(NameHelper.normalize(it.title))
}
return ComposedMediaInfo(
title = NameHelper.normalize(metadataInfo?.title ?: mediaInfo.title),
fallbackCollection = baseInfo.title,
titles = titles,
type = metadataInfo?.type ?: mediaInfo.type,
summaries = summaries,
genres = metadataInfo?.genres ?: emptyList()
)
}
}

4
apps/coordinator/src/main/kotlin/no/iktdev/mediaprocessing/coordinator/tasksV2/listeners/ReadMediaFileStreamsTaskListener.kt
View File

@ -35,8 +35,8 @@ class ReadMediaFileStreamsTaskListener() : CoordinatorEventListener() {
val log = KotlinLogging.logger {}
val requiredOperations = listOf(OperationEvents.ENCODE, OperationEvents.EXTRACT)
override val produceEvent: Events = Events.EventMediaReadStreamPerformed
override val listensForEvents: List<Events> = listOf(Events.EventMediaProcessStarted)
override val produceEvent: Events = Events.ReadStreamPerformed
override val listensForEvents: List<Events> = listOf(Events.ProcessStarted)
override fun shouldIProcessAndHandleEvent(incomingEvent: Event, events: List<Event>): Boolean {
val status = super.shouldIProcessAndHandleEvent(incomingEvent, events)

14
apps/coordinator/src/main/kotlin/no/iktdev/mediaprocessing/coordinator/tasksV2/mapping/EventsSummaryMapping.kt
View File

@ -13,7 +13,7 @@ import no.iktdev.mediaprocessing.shared.common.getChecksum
class EventsSummaryMapping {
fun map(events: List<Event>): EventSummary {
val startOperations = events.find { it.eventType == Events.EventMediaProcessStarted }?.dataAs<StartEventData>() ?: throw RuntimeException("No start event found")
val startOperations = events.find { it.eventType == Events.ProcessStarted }?.dataAs<StartEventData>() ?: throw RuntimeException("No start event found")
val successOperations = listOfNotNull(
if (isEncodedSuccessful(events)) OperationEvents.ENCODE else null,
if (isExtractedSuccessful(events)) OperationEvents.EXTRACT else null,
@ -33,26 +33,26 @@ class EventsSummaryMapping {
}
fun isEncodedSuccessful(events: List<Event>) = events.filter { it.eventType == Events.EventWorkEncodePerformed }.any { it.isSuccessful() }
fun isExtractedSuccessful(events: List<Event>) = events.filter { it.eventType == Events.EventWorkExtractPerformed }.any { it.isSuccessful() }
fun isConvertedSuccessful(events: List<Event>) = events.filter { it.eventType == Events.EventWorkConvertPerformed }.any { it.isSuccessful() }
fun isEncodedSuccessful(events: List<Event>) = events.filter { it.eventType == Events.WorkEncodePerformed }.any { it.isSuccessful() }
fun isExtractedSuccessful(events: List<Event>) = events.filter { it.eventType == Events.WorkExtractPerformed }.any { it.isSuccessful() }
fun isConvertedSuccessful(events: List<Event>) = events.filter { it.eventType == Events.WorkConvertPerformed }.any { it.isSuccessful() }
fun getProducesFiles(events: List<Event>): OutputFiles {
val encoded = if (isEncodedSuccessful(events)) {
events.filter { it.eventType == Events.EventWorkEncodePerformed }
events.filter { it.eventType == Events.WorkEncodePerformed }
.filter { it.isSuccessful() }
.mapNotNull { it.dataAs<EncodedData>()?.outputFile }
} else emptyList()
val extracted = if (isExtractedSuccessful(events)) {
events.filter { it.eventType == Events.EventWorkExtractPerformed }
events.filter { it.eventType == Events.WorkExtractPerformed }
.filter { it.isSuccessful() }
.mapNotNull { it.dataAs<ExtractedData>() }
.map { it.outputFile }
} else emptyList()
val converted = if (isConvertedSuccessful(events)) {
events.filter { it.eventType == Events.EventWorkConvertPerformed }
events.filter { it.eventType == Events.WorkConvertPerformed }
.filter { it.isSuccessful() }
.mapNotNull { it.dataAs<ConvertedData>() }
.flatMap { it.outputFiles }

8
apps/coordinator/src/main/kotlin/no/iktdev/mediaprocessing/coordinator/tasksV2/mapping/store/ContentCompletionMover.kt
View File

@ -27,7 +27,7 @@ class ContentCompletionMover(val collection: String, val events: List<Event>) {
* @return Pair<OldPath, NewPath> or null if no file found
*/
fun moveVideo(): Pair<String, String>? {
val encodedFile = events.find { it.eventType == Events.EventWorkEncodePerformed }?.dataAs<EncodedData>()?.outputFile?.let {
val encodedFile = events.find { it.eventType == Events.WorkEncodePerformed }?.dataAs<EncodedData>()?.outputFile?.let {
File(it)
} ?: return null
if (!encodedFile.exists()) {
@ -44,7 +44,7 @@ class ContentCompletionMover(val collection: String, val events: List<Event>) {
}
fun moveCover(): Pair<String, String>? {
val coverFile = events.find { it.eventType == Events.EventWorkDownloadCoverPerformed }?.
val coverFile = events.find { it.eventType == Events.WorkDownloadCoverPerformed }?.
az<MediaCoverDownloadedEvent>()?.data?.absoluteFilePath?.let {
File(it)
} ?: return null
@ -74,9 +74,9 @@ class ContentCompletionMover(val collection: String, val events: List<Event>) {
fun getMovableSubtitles(): List<MovableSubtitle> {
val extracted =
events.filter { it.eventType == Events.EventWorkExtractPerformed }.mapNotNull { it.dataAs<ExtractedData>() }
events.filter { it.eventType == Events.WorkExtractPerformed }.mapNotNull { it.dataAs<ExtractedData>() }
val converted =
events.filter { it.eventType == Events.EventWorkConvertPerformed }.mapNotNull { it.dataAs<ConvertedData>() }
events.filter { it.eventType == Events.WorkConvertPerformed }.mapNotNull { it.dataAs<ConvertedData>() }
val items = mutableListOf<MovableSubtitle>()

4
apps/coordinator/src/main/kotlin/no/iktdev/mediaprocessing/coordinator/tasksV2/mapping/store/ContentSubtitleStore.kt
View File

@ -11,11 +11,11 @@ import java.io.File
object ContentSubtitleStore {
val log = KotlinLogging.logger {}
fun storeSubtitles(collection: String, language: String, destinationFile: File): Boolean {
fun storeSubtitles(collection: String, destinationFile: File): Boolean {
return executeWithStatus (getStoreDatabase().database, block = {
subtitle.insert {
it[this.associatedWithVideo] = destinationFile.nameWithoutExtension
it[this.language] = language
it[this.language] = destinationFile.parentFile.nameWithoutExtension
it[this.collection] = collection
it[this.format] = destinationFile.extension.uppercase()
it[this.subtitle] = destinationFile.name

30
apps/coordinator/src/main/kotlin/no/iktdev/mediaprocessing/coordinator/tasksV2/validator/CompletionValidator.kt
View File

@ -19,11 +19,11 @@ object CompletionValidator {
*/
fun req1(started: MediaProcessStartEvent, events: List<Event>): Boolean {
val encodeFulfilledOrSkipped = if (started.data?.operations?.contains(OperationEvents.ENCODE) == true) {
events.any { it.eventType == Events.EventMediaParameterEncodeCreated }
events.any { it.eventType == Events.ParameterEncodeCreated }
} else true
val extractFulfilledOrSkipped = if (started.data?.operations?.contains(OperationEvents.EXTRACT) == true) {
events.any { it.eventType == Events.EventMediaParameterExtractCreated }
events.any { it.eventType == Events.ParameterExtractCreated }
} else true
if (!encodeFulfilledOrSkipped || !extractFulfilledOrSkipped) {
@ -37,21 +37,21 @@ object CompletionValidator {
*/
fun req2(operations: List<OperationEvents>, events: List<Event>): Boolean {
if (OperationEvents.ENCODE in operations) {
val encodeParamter = events.find { it.eventType == Events.EventMediaParameterEncodeCreated }?.az<EncodeArgumentCreatedEvent>()
val encodeWork = events.find { it.eventType == Events.EventWorkEncodeCreated }
val encodeParamter = events.find { it.eventType == Events.ParameterEncodeCreated }?.az<EncodeArgumentCreatedEvent>()
val encodeWork = events.find { it.eventType == Events.WorkEncodeCreated }
if (encodeParamter?.isSuccessful() == true && (encodeWork == null))
return false
}
val extractParamter = events.find { it.eventType == Events.EventMediaParameterExtractCreated }?.az<ExtractArgumentCreatedEvent>()
val extractWork = events.filter { it.eventType == Events.EventWorkExtractCreated }
val extractParamter = events.find { it.eventType == Events.ParameterExtractCreated }?.az<ExtractArgumentCreatedEvent>()
val extractWork = events.filter { it.eventType == Events.WorkExtractCreated }
if (OperationEvents.EXTRACT in operations) {
if (extractParamter?.isSuccessful() == true && extractParamter.data?.size != extractWork.size)
return false
}
if (OperationEvents.CONVERT in operations) {
val convertWork = events.filter { it.eventType == Events.EventWorkConvertCreated }
val convertWork = events.filter { it.eventType == Events.WorkConvertCreated }
val supportedSubtitleFormats = SubtitleFormats.entries.map { it.name }
val eventsSupportsConvert = extractWork.filter { it.data is ExtractArgumentData }
@ -69,22 +69,22 @@ object CompletionValidator {
*/
fun req3(operations: List<OperationEvents>, events: List<Event>): Boolean {
if (OperationEvents.ENCODE in operations) {
val encodeWork = events.filter { it.eventType == Events.EventWorkEncodeCreated }
val encodePerformed = events.filter { it.eventType == Events.EventWorkEncodePerformed }
val encodeWork = events.filter { it.eventType == Events.WorkEncodeCreated }
val encodePerformed = events.filter { it.eventType == Events.WorkEncodePerformed }
if (encodePerformed.size < encodeWork.size)
return false
}
if (OperationEvents.EXTRACT in operations) {
val extractWork = events.filter { it.eventType == Events.EventWorkExtractCreated }
val extractPerformed = events.filter { it.eventType == Events.EventWorkExtractPerformed }
val extractWork = events.filter { it.eventType == Events.WorkExtractCreated }
val extractPerformed = events.filter { it.eventType == Events.WorkExtractPerformed }
if (extractPerformed.size < extractWork.size)
return false
}
if (OperationEvents.CONVERT in operations) {
val convertWork = events.filter { it.eventType == Events.EventWorkConvertCreated }
val convertPerformed = events.filter { it.eventType == Events.EventWorkConvertPerformed }
val convertWork = events.filter { it.eventType == Events.WorkConvertCreated }
val convertPerformed = events.filter { it.eventType == Events.WorkConvertPerformed }
if (convertPerformed.size < convertWork.size)
return false
}
@ -96,7 +96,7 @@ object CompletionValidator {
* Checks if metadata has cover, if so, 2 events are expected
*/
fun req4(events: List<Event>): Boolean {
val metadata = events.find { it.eventType == Events.EventMediaMetadataSearchPerformed }
val metadata = events.find { it.eventType == Events.MetadataSearchPerformed }
if (metadata?.isSkipped() == true) {
return true
}
@ -110,7 +110,7 @@ object CompletionValidator {
return true
}
if (events.any { it.eventType == Events.EventMediaReadOutCover } && events.any { it.eventType == Events.EventWorkDownloadCoverPerformed }) {
if (events.any { it.eventType == Events.ReadOutCover } && events.any { it.eventType == Events.WorkDownloadCoverPerformed }) {
return true
}
return false

5
apps/coordinator/src/test/kotlin/no/iktdev/mediaprocessing/coordinator/tasksV2/listeners/ParseMediaFileStreamsTaskListenerTest.kt
View File

@ -1,20 +1,17 @@
package no.iktdev.mediaprocessing.coordinator.tasksV2.listeners
import com.google.gson.Gson
import com.google.gson.JsonObject
import no.iktdev.eventi.core.WGson
import no.iktdev.eventi.data.dataAs
import no.iktdev.mediaprocessing.shared.common.contract.Events
import no.iktdev.mediaprocessing.shared.common.contract.fromJsonWithDeserializer
import org.assertj.core.api.Assertions.assertThat
import org.junit.jupiter.api.Assertions.*
import org.junit.jupiter.api.Test
class ParseMediaFileStreamsTaskListenerTest {
@Test
fun testParse() {
val event = data.fromJsonWithDeserializer(Events.EventMediaReadStreamPerformed).dataAs<JsonObject>()
val event = data.fromJsonWithDeserializer(Events.ReadStreamPerformed).dataAs<JsonObject>()
val parser = ParseMediaFileStreamsTaskListener()
val result = parser.parseStreams(event)

5
apps/coordinator/src/test/kotlin/no/iktdev/mediaprocessing/coordinator/tasksV2/mapping/EncodeWorkArgumentsMappingTest.kt
View File

@ -5,18 +5,15 @@ import no.iktdev.mediaprocessing.shared.common.contract.data.MediaFileStreamsPar
import no.iktdev.mediaprocessing.shared.common.contract.data.az
import no.iktdev.mediaprocessing.shared.common.contract.ffmpeg.AudioPreference
import no.iktdev.mediaprocessing.shared.common.contract.ffmpeg.EncodingPreference
import no.iktdev.mediaprocessing.shared.common.contract.ffmpeg.ParsedMediaStreams
import no.iktdev.mediaprocessing.shared.common.contract.ffmpeg.VideoPreference
import no.iktdev.mediaprocessing.shared.common.contract.fromJsonWithDeserializer
import org.junit.jupiter.api.Assertions.*
import org.junit.jupiter.api.Test
import java.io.File
class EncodeWorkArgumentsMappingTest {
@Test
fun parse() {
val event = data.fromJsonWithDeserializer(Events.EventMediaParseStreamPerformed)
val event = data.fromJsonWithDeserializer(Events.ParseStreamPerformed)
val parser = EncodeWorkArgumentsMapping(
"potato.mkv",
"potato.mp4",

28
shared/common/src/main/kotlin/no/iktdev/mediaprocessing/shared/common/contract/EventToClazzTable.kt
View File

@ -6,35 +6,9 @@ import no.iktdev.eventi.core.WGson
import no.iktdev.mediaprocessing.shared.common.contract.data.*
private val log = KotlinLogging.logger {}
object EventToClazzTable {
val table = mutableMapOf(
Events.EventMediaProcessStarted to MediaProcessStartEvent::class.java,
Events.EventMediaReadBaseInfoPerformed to BaseInfoEvent::class.java,
Events.EventMediaReadStreamPerformed to MediaFileStreamsReadEvent::class.java,
Events.EventMediaParseStreamPerformed to MediaFileStreamsParsedEvent::class.java,
Events.EventWorkConvertCreated to ConvertWorkCreatedEvent::class.java,
Events.EventWorkConvertPerformed to ConvertWorkPerformed::class.java,
Events.EventMediaParameterEncodeCreated to EncodeArgumentCreatedEvent::class.java,
Events.EventWorkEncodeCreated to EncodeWorkCreatedEvent::class.java,
Events.EventWorkEncodePerformed to EncodeWorkPerformedEvent::class.java,
Events.EventMediaParameterExtractCreated to ExtractArgumentCreatedEvent::class.java,
Events.EventWorkExtractCreated to ExtractWorkCreatedEvent::class.java,
Events.EventWorkExtractPerformed to ExtractWorkPerformedEvent::class.java,
Events.EventWorkDownloadCoverPerformed to MediaCoverDownloadedEvent::class.java,
Events.EventMediaReadOutCover to MediaCoverInfoReceivedEvent::class.java,
Events.EventMediaParseStreamPerformed to MediaFileStreamsParsedEvent::class.java,
Events.EventMediaReadStreamPerformed to MediaFileStreamsReadEvent::class.java,
Events.EventMediaMetadataSearchPerformed to MediaMetadataReceivedEvent::class.java,
Events.EventMediaReadOutNameAndType to MediaOutInformationConstructedEvent::class.java,
Events.EventMediaWorkProceedPermitted to PermitWorkCreationEvent::class.java,
Events.EventMediaProcessCompleted to MediaProcessCompletedEvent::class.java
)
}
fun String.fromJsonWithDeserializer(event: Events): Event {
val clazz = EventToClazzTable.table[event]
val clazz = event.dataClass
clazz?.let { eventClass ->
try {
val type = TypeToken.getParameterized(eventClass).type

69
shared/common/src/main/kotlin/no/iktdev/mediaprocessing/shared/common/contract/Events.kt
View File

@ -1,41 +1,37 @@
package no.iktdev.mediaprocessing.shared.common.contract
enum class Events(val event: String) {
EventMediaProcessStarted ("event:media-process:started"),
import no.iktdev.mediaprocessing.shared.common.contract.data.*
EventMediaReadStreamPerformed ("event:media-read-stream:performed"),
EventMediaParseStreamPerformed ("event:media-parse-stream:performed"),
EventMediaReadBaseInfoPerformed ("event:media-read-base-info:performed"),
EventMediaMetadataSearchPerformed ("event:media-metadata-search:performed"),
EventMediaReadOutNameAndType ("event:media-read-out-name-and-type:performed"),
EventMediaReadOutCover ("event:media-read-out-cover:performed"),
enum class Events(val event: String, val dataClass: Class<out Event>) {
ProcessStarted ("event:media-process:started", dataClass = MediaProcessStartEvent::class.java),
EventMediaParameterEncodeCreated ("event:media-encode-parameter:created"),
EventMediaParameterExtractCreated ("event:media-extract-parameter:created"),
ReadStreamPerformed ("event:media-read-stream:performed", MediaFileStreamsReadEvent::class.java),
ParseStreamPerformed ("event:media-parse-stream:performed", MediaFileStreamsParsedEvent::class.java),
ReadBaseInfoPerformed ("event:media-read-base-info:performed", BaseInfoEvent::class.java),
MetadataSearchPerformed ("event:media-metadata-search:performed", MediaMetadataReceivedEvent::class.java),
ReadOutNameAndType ("event:media-read-out-name-and-type:performed", MediaOutInformationConstructedEvent::class.java),
ReadOutCover ("event:media-read-out-cover:performed", MediaCoverInfoReceivedEvent::class.java),
EventMediaParameterDownloadCoverCreated ("event:media-download-cover-parameter:created"),
ParameterEncodeCreated ("event:media-encode-parameter:created", EncodeArgumentCreatedEvent::class.java),
ParameterExtractCreated ("event:media-extract-parameter:created", ExtractArgumentCreatedEvent::class.java),
EventMediaWorkProceedPermitted ("event:media-work-proceed:permitted"),
//EventMediaParameterDownloadCoverCreated ("event:media-download-cover-parameter:created"),
EventNotificationOfWorkItemRemoval("event:notification-work-item-removal"),
WorkProceedPermitted ("event:media-work-proceed:permitted", PermitWorkCreationEvent::class.java),
EventWorkEncodeCreated ("event:work-encode:created"),
EventWorkExtractCreated ("event:work-extract:created"),
EventWorkConvertCreated ("event:work-convert:created"),
//EventNotificationOfWorkItemRemoval("event:notification-work-item-removal"),
EventWorkEncodePerformed ("event:work-encode:performed"),
EventWorkExtractPerformed ("event:work-extract:performed"),
EventWorkConvertPerformed ("event:work-convert:performed"),
EventWorkDownloadCoverPerformed ("event:work-download-cover:performed"),
WorkEncodeCreated ("event:work-encode:created", EncodeWorkCreatedEvent::class.java),
WorkExtractCreated ("event:work-extract:created", ExtractWorkCreatedEvent::class.java),
WorkConvertCreated ("event:work-convert:created", ConvertWorkCreatedEvent::class.java),
EVENT_STORE_VIDEO_PERFORMED ("event:store-video:performed"),
EVENT_STORE_SUBTITLE_PERFORMED ("event:store-subtitle:performed"),
EVENT_STORE_COVER_PERFORMED ("event:store-cover:performed"),
EVENT_STORE_METADATA_PERFORMED ("event:store-metadata:performed"),
EventMediaProcessCompleted ("event:media-process:completed"),
EventCollectAndStore ("event::save"),
WorkEncodePerformed ("event:work-encode:performed", EncodeWorkPerformedEvent::class.java),
WorkExtractPerformed ("event:work-extract:performed", ExtractWorkPerformedEvent::class.java),
WorkConvertPerformed ("event:work-convert:performed", ConvertWorkPerformed::class.java),
WorkDownloadCoverPerformed ("event:work-download-cover:performed", MediaCoverDownloadedEvent::class.java),
PersistContentPerformed ("event:media-persist:completed", PersistedContentEvent::class.java),
ProcessCompleted ("event:media-process:completed", MediaProcessCompletedEvent::class.java),
;
companion object {
@ -46,20 +42,13 @@ enum class Events(val event: String) {
fun isOfWork(event: Events): Boolean {
return event in listOf(
EventWorkConvertCreated,
EventWorkExtractCreated,
EventWorkEncodeCreated,
WorkConvertCreated,
WorkExtractCreated,
WorkEncodeCreated,
EventWorkEncodePerformed,
EventWorkConvertPerformed,
EventWorkExtractPerformed
)
}
fun isOfFinalize(event: Events): Boolean {
return event in listOf(
EventMediaProcessCompleted,
EventCollectAndStore
WorkEncodePerformed,
WorkConvertPerformed,
WorkExtractPerformed
)
}
}

3
shared/common/src/main/kotlin/no/iktdev/mediaprocessing/shared/common/contract/data/BaseInfoEvent.kt
View File

@ -1,12 +1,11 @@
package no.iktdev.mediaprocessing.shared.common.contract.data
import no.iktdev.eventi.data.EventImpl
import no.iktdev.eventi.data.EventMetadata
import no.iktdev.mediaprocessing.shared.common.contract.Events
data class BaseInfoEvent(
override val metadata: EventMetadata,
override val eventType: Events = Events.EventMediaReadBaseInfoPerformed,
override val eventType: Events = Events.ReadBaseInfoPerformed,
override val data: BaseInfo? = null
) : Event()

2
shared/common/src/main/kotlin/no/iktdev/mediaprocessing/shared/common/contract/data/ConvertWorkCreatedEvent.kt
View File

@ -7,7 +7,7 @@ import no.iktdev.mediaprocessing.shared.common.contract.dto.tasks.TaskData
data class ConvertWorkCreatedEvent(
override val metadata: EventMetadata,
override val eventType: Events = Events.EventWorkConvertCreated,
override val eventType: Events = Events.WorkConvertCreated,
override val data: ConvertData? = null
) : Event() {
}

2
shared/common/src/main/kotlin/no/iktdev/mediaprocessing/shared/common/contract/data/ConvertWorkPerformed.kt
View File

@ -5,7 +5,7 @@ import no.iktdev.mediaprocessing.shared.common.contract.Events
class ConvertWorkPerformed(
override val metadata: EventMetadata,
override val eventType: Events = Events.EventWorkConvertPerformed,
override val eventType: Events = Events.WorkConvertPerformed,
override val data: ConvertedData? = null,
val message: String? = null
) : Event() {

2
shared/common/src/main/kotlin/no/iktdev/mediaprocessing/shared/common/contract/data/EncodeArgumentCreatedEvent.kt
View File

@ -6,7 +6,7 @@ import no.iktdev.mediaprocessing.shared.common.contract.dto.tasks.TaskData
data class EncodeArgumentCreatedEvent(
override val metadata: EventMetadata,
override val eventType: Events = Events.EventMediaParameterEncodeCreated,
override val eventType: Events = Events.ParameterEncodeCreated,
override val data: EncodeArgumentData? = null
) : Event() {

2
shared/common/src/main/kotlin/no/iktdev/mediaprocessing/shared/common/contract/data/EncodeWorkCreatedEvent.kt
View File

@ -5,6 +5,6 @@ import no.iktdev.mediaprocessing.shared.common.contract.Events
data class EncodeWorkCreatedEvent(
override val metadata: EventMetadata,
override val eventType: Events = Events.EventWorkEncodeCreated,
override val eventType: Events = Events.WorkEncodeCreated,
override val data: EncodeArgumentData? = null
) : Event()

2
shared/common/src/main/kotlin/no/iktdev/mediaprocessing/shared/common/contract/data/EncodeWorkPerformedEvent.kt
View File

@ -5,7 +5,7 @@ import no.iktdev.mediaprocessing.shared.common.contract.Events
data class EncodeWorkPerformedEvent(
override val metadata: EventMetadata,
override val eventType: Events = Events.EventWorkEncodePerformed,
override val eventType: Events = Events.WorkEncodePerformed,
override val data: EncodedData? = null,
val message: String? = null
) : Event() {

2
shared/common/src/main/kotlin/no/iktdev/mediaprocessing/shared/common/contract/data/ExtractArgumentCreatedEvent.kt
View File

@ -6,7 +6,7 @@ import no.iktdev.mediaprocessing.shared.common.contract.dto.tasks.TaskData
data class ExtractArgumentCreatedEvent(
override val metadata: EventMetadata,
override val eventType: Events = Events.EventMediaParameterExtractCreated,
override val eventType: Events = Events.ParameterExtractCreated,
override val data: List<ExtractArgumentData>? = null
): Event()

2
shared/common/src/main/kotlin/no/iktdev/mediaprocessing/shared/common/contract/data/ExtractWorkCreatedEvent.kt
View File

@ -5,7 +5,7 @@ import no.iktdev.mediaprocessing.shared.common.contract.Events
data class ExtractWorkCreatedEvent(
override val metadata: EventMetadata,
override val eventType: Events = Events.EventWorkExtractCreated,
override val eventType: Events = Events.WorkExtractCreated,
override val data: ExtractArgumentData? = null
) : Event() {
}

2
shared/common/src/main/kotlin/no/iktdev/mediaprocessing/shared/common/contract/data/ExtractWorkPerformedEvent.kt
View File

@ -5,7 +5,7 @@ import no.iktdev.mediaprocessing.shared.common.contract.Events
data class ExtractWorkPerformedEvent(
override val metadata: EventMetadata,
override val eventType: Events = Events.EventWorkExtractPerformed,
override val eventType: Events = Events.WorkExtractPerformed,
override val data: ExtractedData? = null,
val message: String? = null
) : Event() {

2
shared/common/src/main/kotlin/no/iktdev/mediaprocessing/shared/common/contract/data/MediaCoverDownloadedEvent.kt
View File

@ -5,7 +5,7 @@ import no.iktdev.mediaprocessing.shared.common.contract.Events
data class MediaCoverDownloadedEvent(
override val metadata: EventMetadata,
override val eventType: Events = Events.EventWorkDownloadCoverPerformed,
override val eventType: Events = Events.WorkDownloadCoverPerformed,
override val data: DownloadedCover? = null
) : Event() {
}

2
shared/common/src/main/kotlin/no/iktdev/mediaprocessing/shared/common/contract/data/MediaCoverInfoReceivedEvent.kt
View File

@ -5,7 +5,7 @@ import no.iktdev.mediaprocessing.shared.common.contract.Events
data class MediaCoverInfoReceivedEvent(
override val metadata: EventMetadata,
override val eventType: Events = Events.EventMediaReadOutCover,
override val eventType: Events = Events.ReadOutCover,
override val data: CoverDetails? = null
) : Event() {
}

2
shared/common/src/main/kotlin/no/iktdev/mediaprocessing/shared/common/contract/data/MediaFileStreamsParsedEvent.kt
View File

@ -7,6 +7,6 @@ import no.iktdev.mediaprocessing.shared.common.contract.ffmpeg.ParsedMediaStream
class MediaFileStreamsParsedEvent(
override val metadata: EventMetadata,
override val data: ParsedMediaStreams? = null,
override val eventType: Events = Events.EventMediaParseStreamPerformed
override val eventType: Events = Events.ParseStreamPerformed
) : Event()

3
shared/common/src/main/kotlin/no/iktdev/mediaprocessing/shared/common/contract/data/MediaFileStreamsReadEvent.kt
View File

@ -1,12 +1,11 @@
package no.iktdev.mediaprocessing.shared.common.contract.data
import com.google.gson.JsonObject
import no.iktdev.eventi.data.EventImpl
import no.iktdev.eventi.data.EventMetadata
import no.iktdev.mediaprocessing.shared.common.contract.Events
class MediaFileStreamsReadEvent(
override val metadata: EventMetadata,
override val data: JsonObject? = null,
override val eventType: Events = Events.EventMediaReadStreamPerformed
override val eventType: Events = Events.ReadStreamPerformed
) : Event()

2
shared/common/src/main/kotlin/no/iktdev/mediaprocessing/shared/common/contract/data/MediaMetadataReceivedEvent.kt
View File

@ -5,7 +5,7 @@ import no.iktdev.mediaprocessing.shared.common.contract.Events
data class MediaMetadataReceivedEvent(
override val metadata: EventMetadata,
override val eventType: Events = Events.EventMediaMetadataSearchPerformed,
override val eventType: Events = Events.MetadataSearchPerformed,
override val data: pyMetadata? = null,
): Event() {
}

2
shared/common/src/main/kotlin/no/iktdev/mediaprocessing/shared/common/contract/data/MediaOutInformationConstructedEvent.kt
View File

@ -7,7 +7,7 @@ import no.iktdev.mediaprocessing.shared.common.contract.Events
data class MediaOutInformationConstructedEvent(
override val metadata: EventMetadata,
override val eventType: Events = Events.EventMediaReadOutNameAndType,
override val eventType: Events = Events.ReadOutNameAndType,
override val data: MediaInfoReceived? = null
) : Event() {
}

34
shared/common/src/main/kotlin/no/iktdev/mediaprocessing/shared/common/contract/data/MediaProcessCompletedEvent.kt
View File

@ -2,31 +2,27 @@ package no.iktdev.mediaprocessing.shared.common.contract.data
import no.iktdev.eventi.data.EventMetadata
import no.iktdev.mediaprocessing.shared.common.contract.Events
import no.iktdev.mediaprocessing.shared.common.contract.reader.SummaryInfo
data class MediaProcessCompletedEvent(
override val metadata: EventMetadata,
override val data: CompletedEventData?,
override val eventType: Events = Events.EventMediaProcessCompleted
override val data: CompletedData?,
override val eventType: Events = Events.ProcessCompleted
): Event()
data class CompletedEventData(
data class CompletedData(
val eventIdsCollected: List<String>,
val videoMoved: VideoMoved?,
val coverMoved: CoverMoved?,
val subtitlesMoved: List<SubtitlesMoved>
val metadataStored: MetadataStored
)
data class SubtitlesMoved(
val source: String,
val destination: String
)
data class CoverMoved(
val source: String,
val destination: String
)
data class VideoMoved(
val source: String,
val destination: String
data class MetadataStored(
val title: String,
val titles: List<String>,
val type: String,
val cover: String? = null,
val collection: String,
val summary: List<SummaryInfo> = emptyList(),
val foundTitles: List<String>,
val genres: List<String>,
val genreIds: String? = null
)

2
shared/common/src/main/kotlin/no/iktdev/mediaprocessing/shared/common/contract/data/MediaProcessStartEvent.kt
View File

@ -8,7 +8,7 @@ import no.iktdev.mediaprocessing.shared.common.contract.dto.OperationEvents
data class MediaProcessStartEvent(
override val metadata: EventMetadata,
override val data: StartEventData?,
override val eventType: Events = Events.EventMediaProcessStarted
override val eventType: Events = Events.ProcessStarted
): Event()
data class StartEventData(

2
shared/common/src/main/kotlin/no/iktdev/mediaprocessing/shared/common/contract/data/PermitWorkCreationEvent.kt
View File

@ -5,7 +5,7 @@ import no.iktdev.mediaprocessing.shared.common.contract.Events
data class PermitWorkCreationEvent(
override val metadata: EventMetadata,
override val eventType: Events = Events.EventMediaWorkProceedPermitted,
override val eventType: Events = Events.WorkProceedPermitted,
override val data: String?
) : Event() {
}

23
shared/common/src/main/kotlin/no/iktdev/mediaprocessing/shared/common/contract/data/PersistedContentEvent.kt Normal file
View File

@ -0,0 +1,23 @@
package no.iktdev.mediaprocessing.shared.common.contract.data
import no.iktdev.eventi.data.EventMetadata
import no.iktdev.mediaprocessing.shared.common.contract.Events
data class PersistedContentEvent(
override val eventType: Events = Events.PersistContentPerformed,
override val metadata: EventMetadata,
override val data: PersistedContent?
): Event() {}
data class PersistedContent(
val cover: PersistedItem?,
val video: PersistedItem?,
val subtitles: List<PersistedItem>
)
data class PersistedItem(
val cacheSourceFileName: String,
val storeDestinationFileName: String
)

15
shared/common/src/main/kotlin/no/iktdev/mediaprocessing/shared/common/database/cal/EventsManager.kt
View File

@ -9,7 +9,6 @@ import no.iktdev.eventi.database.DataSource
import no.iktdev.eventi.database.isCausedByDuplicateError
import no.iktdev.eventi.database.isExposedSqlException
import no.iktdev.eventi.database.withDirtyRead
import no.iktdev.eventi.implementations.EventsManagerImpl
import no.iktdev.mediaprocessing.shared.common.database.tables.allEvents
import no.iktdev.mediaprocessing.shared.common.database.tables.events
import no.iktdev.mediaprocessing.shared.common.contract.Events
@ -81,10 +80,10 @@ class EventsManager(dataSource: DataSource) : EventsManagerContract(dataSource)
private val exemptedFromSingleEvent = listOf(
Events.EventWorkConvertCreated,
Events.EventWorkExtractCreated,
Events.EventWorkConvertPerformed,
Events.EventWorkExtractPerformed
Events.WorkConvertCreated,
Events.WorkExtractCreated,
Events.WorkConvertPerformed,
Events.WorkExtractPerformed
)
private fun isExempted(event: Events): Boolean {
@ -95,7 +94,7 @@ class EventsManager(dataSource: DataSource) : EventsManagerContract(dataSource)
return withDirtyRead(dataSource.database) {
val completedEvents = events
.slice(events.referenceId)
.select { events.event eq Events.EventMediaProcessCompleted.event }
.select { events.event eq Events.ProcessCompleted.event }
events
.slice(events.referenceId)
@ -109,7 +108,7 @@ class EventsManager(dataSource: DataSource) : EventsManagerContract(dataSource)
return withDirtyRead(dataSource.database) {
events.selectAll()
.groupBy { it[events.referenceId] }
.mapNotNull { it.value.mapNotNull { v -> v.toEvent() } }.filter { it.none { e -> e.eventType == Events.EventMediaProcessCompleted } }
.mapNotNull { it.value.mapNotNull { v -> v.toEvent() } }.filter { it.none { e -> e.eventType == Events.ProcessCompleted } }
} ?: emptyList()
}
@ -118,7 +117,7 @@ class EventsManager(dataSource: DataSource) : EventsManagerContract(dataSource)
events.select { events.referenceId eq referenceId }
.mapNotNull { it.toEvent() }
} ?: emptyList()
return if (events.any { it.eventType == Events.EventMediaProcessCompleted }) emptyList() else events
return if (events.any { it.eventType == Events.ProcessCompleted }) emptyList() else events
}
override fun getAllEvents(): List<List<Event>> {