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 2025-03-09 23:06:41 +01:00
parent 422971a914
commit ac4f79ff7a
1 changed files with 24 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -51,6 +51,7 @@ class InputDirectoryWatcher(@Autowired var coordinator: Coordinator): FileWatche
val watchDirectories = SharedConfig.incomingContent val watchDirectories = SharedConfig.incomingContent
val queue = FileWatcherQueue() val queue = FileWatcherQueue()
@Volatile
private var isStopping: Boolean = false private var isStopping: Boolean = false
@PreDestroy @PreDestroy
fun setStop() { fun setStop() {
@ -86,8 +87,10 @@ class InputDirectoryWatcher(@Autowired var coordinator: Coordinator): FileWatche
private suspend fun startWatchOnDirectory(file: File) { private suspend fun startWatchOnDirectory(file: File) {
if (activeWatchers.any {it -> it.file.absolutePath == file.absolutePath}) { if (activeWatchers.any {it -> it.file.absolutePath == file.absolutePath}) {
log.error { "Attempting to start a watcher on an already watched directory ${file.absolutePath}" } log.error { "Attempting to start a watcher on an already watched directory ${file.absolutePath}" }
return
} }
val watcher = file.asWatcher { watcher -> val watcher = file.asWatcher { watcher ->
try {
watcher.consumeEach { watcher.consumeEach {
if (it.file == SharedConfig.incomingContent) { if (it.file == SharedConfig.incomingContent) {
logger.info { "IO Watcher ${it.kind} on ${it.file.absolutePath}" } logger.info { "IO Watcher ${it.kind} on ${it.file.absolutePath}" }
@ -109,6 +112,11 @@ class InputDirectoryWatcher(@Autowired var coordinator: Coordinator): FileWatche
e.printStackTrace() e.printStackTrace()
} }
} }
} catch (e: Exception) {
log.error { "Consume failed ${e.message}" }
e.printStackTrace()
watcher.close()
}
}.also { watcher -> }.also { watcher ->
watcher.watcher.invokeOnClose { watcher.watcher.invokeOnClose {
it?.printStackTrace() it?.printStackTrace()