Wip

apps/processer/src/main/kotlin/no/iktdev/mediaprocessing/processer/services/EncodeService.kt

@@ -1,5 +1,6 @@
 package no.iktdev.mediaprocessing.processer.services
 
+import com.google.gson.Gson
 import kotlinx.coroutines.delay
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.runBlocking
@@ -209,6 +210,12 @@ class EncodeService(
             outputFiles = listOf(runner.outputFile),
             progress = progress?.toProcessProgress()
         )
+        try {
+            log.info { "Reporting encode progress ${Gson().toJson(processerEventInfo)}" }
+
+        } catch (e: Exception) {
+            e.printStackTrace()
+        }
         try {
             reporter.sendEncodeProgress(processerEventInfo)
         } catch (e: Exception) {

apps/ui/src/main/kotlin/no/iktdev/mediaprocessing/ui/socket/a2a/ProcesserListenerService.kt

@@ -7,6 +7,7 @@ import no.iktdev.mediaprocessing.ui.UIEnv
 import no.iktdev.mediaprocessing.ui.WebSocketMonitoringService
 import no.iktdev.mediaprocessing.ui.log
 import org.springframework.beans.factory.annotation.Autowired
+import org.springframework.messaging.converter.MappingJackson2MessageConverter
 import org.springframework.messaging.converter.StringMessageConverter
 import org.springframework.messaging.simp.SimpMessagingTemplate
 import org.springframework.messaging.simp.stomp.*
@@ -41,7 +42,7 @@ class ProcesserListenerService(
 
     private final fun connectAndListen() {
         log.info { "EncoderWsUrl: ${UIEnv.socketEncoder}" }
-        client.messageConverter = StringMessageConverter()
+        client.messageConverter = MappingJackson2MessageConverter()
         client.connect(UIEnv.socketEncoder, object : StompSessionHandlerAdapter() {
             override fun afterConnected(session: StompSession, connectedHeaders: StompHeaders) {
                 super.afterConnected(session, connectedHeaders)
@@ -73,11 +74,13 @@ class ProcesserListenerService(
     }
 
     private fun scheduleReconnect() {
+        log.info { "Scheduling reconnect!" }
         (reconnectTask ?: kotlinx.coroutines.Runnable {
             logger.info { "Attempting to reconnect" }
             connectAndListen()
         }).let {
             scheduler.scheduleAtFixedRate(it, 5, 5, TimeUnit.SECONDS)
+            reconnectTask = it
         }
     }
 
@@ -99,9 +102,22 @@ class ProcesserListenerService(
         }
 
         override fun handleFrame(headers: StompHeaders, payload: Any?) {
-            val response = gson.fromJson(payload.toString(), ProcesserEventInfo::class.java)
             if (webSocketMonitoringService.anyListening()) {
-                message?.convertAndSend("/topic/processer/encode/progress", response)
+                message?.convertAndSend("/topic/processer/encode/progress", payload)
+            }
+            val response = when (payload) {
+                is String -> {
+                    log.info { "Received string of: $payload" }
+                    gson.fromJson(payload.toString(), ProcesserEventInfo::class.java)
+                }
+                is ProcesserEventInfo -> {
+                    log.info { "Received object of ${Gson().toJson(payload)}" }
+                    payload
+                }
+                else -> {
+                    log.error { "Unsupported payload $payload" }
+                    return
+                }
             }
         }
     }