Merge pull request #123 from leanix/feature/CID-3916/check-connection-before-sending-messages

CID-3916: check connection before sending messages

Author: mohamedlajmileanix

src/main/kotlin/net/leanix/githubagent/handler/BrokerStompSessionHandler.kt

@@ -9,7 +9,6 @@ import org.springframework.messaging.simp.stomp.StompHeaders
 import org.springframework.messaging.simp.stomp.StompSession
 import org.springframework.messaging.simp.stomp.StompSessionHandlerAdapter
 import org.springframework.stereotype.Component
-import java.util.concurrent.CountDownLatch
 
 @Component
 class BrokerStompSessionHandler(
@@ -24,12 +23,9 @@ class BrokerStompSessionHandler(
 
     private var isConnected = false
 
-    private val latch = CountDownLatch(1)
-
     override fun afterConnected(session: StompSession, connectedHeaders: StompHeaders) {
         logger.info("connected to the server: ${session.sessionId}")
         isConnected = true
-        latch.countDown()
         session.subscribe("/user/queue/message/artifact", artifactDownloadHandler)
         session.subscribe("/user/queue/message/repository", repositoryGetHandler)
     }
@@ -42,27 +38,17 @@ class BrokerStompSessionHandler(
         exception: Throwable,
     ) {
         logger.error("handleException", exception)
-        logger.error(exception.message)
     }
 
     override fun handleTransportError(session: StompSession, exception: Throwable) {
         logger.error("Connection error: ${exception.message}")
-        if (isConnected) {
-            logger.error("Session closed. This could be due to a network error or the server closing the connection.")
-            isConnected = false
-            logger.info("Reconnecting...")
-            webSocketService.initSession()
-        } else {
-            if (latch.count != 0L) latch.countDown()
-        }
+        isConnected = false
+        logger.error("Session closed. This could be due to a network error or the server closing the connection.")
+        logger.info("Reconnecting...")
+        webSocketService.initSession()
     }
 
     fun isConnected(): Boolean {
-        awaitConnection()
         return isConnected
     }
-
-    private fun awaitConnection() {
-        latch.await()
-    }
 }

src/main/kotlin/net/leanix/githubagent/services/WebSocketService.kt

@@ -1,14 +1,16 @@
 package net.leanix.githubagent.services
 
 import net.leanix.githubagent.config.WebSocketClientConfig
+import net.leanix.githubagent.handler.BrokerStompSessionHandler
 import net.leanix.githubagent.shared.TOPIC_PREFIX
 import org.slf4j.LoggerFactory
 import org.springframework.messaging.simp.stomp.StompSession
 import org.springframework.stereotype.Service
 
 @Service
 class WebSocketService(
-    private val webSocketClientConfig: WebSocketClientConfig
+    private val webSocketClientConfig: WebSocketClientConfig,
+    private val brokerStompSessionHandler: BrokerStompSessionHandler,
 ) {
 
     private val logger = LoggerFactory.getLogger(WebSocketService::class.java)
@@ -18,12 +20,17 @@ class WebSocketService(
         logger.info("Initializing websocket session")
         kotlin.runCatching {
             stompSession = webSocketClientConfig.initSession()
+
+            logger.info("Websocket session initialized with success")
         }.onFailure {
             logger.error("Failed to initialize WebSocket session")
         }
     }
 
     fun sendMessage(topic: String, data: Any) {
+        if (!brokerStompSessionHandler.isConnected()) {
+            return
+        }
         stompSession!!.send("$TOPIC_PREFIX$topic", data)
     }
 }

src/main/kotlin/net/leanix/githubagent/services/WebhookEventService.kt

@@ -39,7 +39,7 @@ class WebhookEventService(
             "PUSH" -> handlePushEvent(payload)
             "INSTALLATION" -> handleInstallationEvent(payload)
             else -> {
-                logger.info("Sending event of type: $eventType")
+                logger.debug("Sending event of type: $eventType")
                 webSocketService.sendMessage("/events/other/$eventType", payload)
             }
         }

src/test/kotlin/net/leanix/githubagent/services/WebSocketServiceTests.kt

@@ -3,6 +3,7 @@ package net.leanix.githubagent.services
 import io.mockk.every
 import io.mockk.mockk
 import net.leanix.githubagent.config.WebSocketClientConfig
+import net.leanix.githubagent.handler.BrokerStompSessionHandler
 import org.junit.jupiter.api.Assertions.assertNotNull
 import org.junit.jupiter.api.BeforeEach
 import org.junit.jupiter.api.Test
@@ -13,15 +14,17 @@ class WebSocketServiceTests {
     private lateinit var webSocketService: WebSocketService
     private val webSocketClientConfig: WebSocketClientConfig = mockk()
     private val stompSession: StompSession = mockk()
+    private val brokerStompSessionHandler: BrokerStompSessionHandler = mockk()
 
     @BeforeEach
     fun setUp() {
-        webSocketService = WebSocketService(webSocketClientConfig)
+        webSocketService = WebSocketService(webSocketClientConfig, brokerStompSessionHandler)
     }
 
     @Test
     fun `initSession should initialize stompSession successfully`() {
         every { webSocketClientConfig.initSession() } returns stompSession
+        every { brokerStompSessionHandler.isConnected() } returns true
 
         webSocketService.initSession()