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

v3 31 - Restart watcher

Browse Source
This commit is contained in:
bskjon 2024-07-19 23:36:31 +02:00
parent f5a3603a5a
commit 0aee64c7f3
1 changed files with 45 additions and 23 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

68
apps/coordinator/src/main/kotlin/no/iktdev/mediaprocessing/coordinator/watcher/InputDirectoryWatcher.kt
View File

@ -4,14 +4,18 @@ import dev.vishna.watchservice.KWatchEvent.Kind.Deleted
import dev.vishna.watchservice.KWatchEvent.Kind.Initialized
import dev.vishna.watchservice.asWatchChannel
import kotlinx.coroutines.channels.consumeEach
import kotlinx.coroutines.coroutineScope
import kotlinx.coroutines.launch
import mu.KotlinLogging
import no.iktdev.mediaprocessing.coordinator.*
import no.iktdev.mediaprocessing.shared.common.SharedConfig
import no.iktdev.mediaprocessing.shared.common.extended.isSupportedVideoFile
import no.iktdev.mediaprocessing.shared.contract.ProcessType
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.context.ApplicationContext
import org.springframework.stereotype.Service
import java.io.File
import javax.annotation.PreDestroy
interface FileWatcherEvents {
@ -35,39 +39,57 @@ interface FileWatcherEvents {
@Service
class InputDirectoryWatcher(@Autowired var coordinator: Coordinator): FileWatcherEvents {
private val logger = KotlinLogging.logger {}
val watcherChannel = SharedConfig.incomingContent.asWatchChannel()
val queue = FileWatcherQueue()
private var isStopping: Boolean = false
@PreDestroy
fun setStop() {
isStopping = true
}
suspend fun watchFiles() {
log.info { "Starting Watcher" }
watcherChannel.consumeEach {
if (it.file == SharedConfig.incomingContent) {
logger.info { "IO Watcher ${it.kind} on ${it.file.absolutePath}" }
} else {
logger.info { "IO Event: ${it.kind}: ${it.file.name}" }
}
try {
when (it.kind) {
Deleted -> removeFile(it.file)
Initialized -> { /* Do nothing */ }
else -> {
val added = addFile(it.file)
if (!added) {
logger.info { "Ignoring event kind: ${it.kind.name} for file ${it.file.name} as it is not a supported video file" }
}
}
}
} catch (e: Exception) {
e.printStackTrace()
}
}
log.warn { "Watcher stopped!!!" }
if (!isStopping) {
ioCoroutine.launch {
watchFiles()
}
}
}
init {
ioCoroutine.launch {
log.info { "Starting Watcher" }
watcherChannel.consumeEach {
if (it.file == SharedConfig.incomingContent) {
logger.info { "IO Watcher ${it.kind} on ${it.file.absolutePath}" }
} else {
logger.info { "IO Event: ${it.kind}: ${it.file.name}" }
}
try {
when (it.kind) {
Deleted -> removeFile(it.file)
Initialized -> { /* Do nothing */ }
else -> {
val added = addFile(it.file)
if (!added) {
logger.info { "Ignoring event kind: ${it.kind.name} for file ${it.file.name} as it is not a supported video file" }
}
}
}
} catch (e: Exception) {
e.printStackTrace()
}
}
log.info { "Reached end of watcherChannel" }
watchFiles()
}
}
private fun addFile(file: File): Boolean {
return if (file.isFile && file.isSupportedVideoFile()) {
log.info { "Adding ${file.name} to queue" }