Port fixes for connections (#111) to Axon Framework 5 (main)

Author: CodeDrivenMitch

framework-client-messaging/src/main/java/io/axoniq/platform/framework/application/ApplicationMetricReporter.kt

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2022-2025. AxonIQ B.V.
+ * Copyright (c) 2022-2026. AxonIQ B.V.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -52,7 +52,15 @@ class ApplicationMetricReporter(
     }
 
     private fun report() {
-        client.sendReport(Routes.Application.REPORT, reportCreator.createReport()).block()
+        if (!client.isConnected()) {
+            return
+        }
+        client.sendReport(Routes.Application.REPORT, reportCreator.createReport())
+                .doOnError { e ->
+                    logger.debug { "Failed to send application report: ${e.message}" }
+                }
+                .onErrorComplete()
+                .subscribe()
     }
 
     override fun onDisconnected() {

framework-client-messaging/src/main/java/io/axoniq/platform/framework/client/AxoniqConsoleRSocketClient.kt

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2022-2025. AxonIQ B.V.
+ * Copyright (c) 2022-2026. AxonIQ B.V.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -41,6 +41,8 @@ import java.time.temporal.ChronoUnit
 import java.util.concurrent.ScheduledExecutorService
 import java.util.concurrent.ScheduledFuture
 import java.util.concurrent.TimeUnit
+import java.util.concurrent.locks.ReentrantLock
+import kotlin.concurrent.withLock
 import kotlin.math.pow
 
 /**
@@ -76,9 +78,12 @@ class AxoniqConsoleRSocketClient(
     private var maintenanceTask: ScheduledFuture<*>? = null
     private val logger = LoggerFactory.getLogger(this::class.java)
 
+    private val connectionLock = ReentrantLock()
+    @Volatile
     private var rsocket: RSocket? = null
     private var lastConnectionTry = Instant.EPOCH
     private var connectionRetryCount = 0
+    @Volatile
     private var pausedReports = false
     private var supressConnectMessage = false
 
@@ -110,7 +115,7 @@ class AxoniqConsoleRSocketClient(
      * Sends a report to Axoniq Platform. If reports are paused, does nothing silently.
      */
     fun sendReport(route: String, payload: Any): Mono<Unit> {
-        if(pausedReports) {
+        if (pausedReports) {
             return Mono.empty()
         }
         return sendMessage(payload, route)
@@ -236,10 +241,20 @@ class AxoniqConsoleRSocketClient(
 
     fun isConnected() = rsocket != null
 
+    /**
+     * Disposes the current RSocket connection in a thread-safe manner.
+     * This method can be called from multiple threads (e.g., TCP disconnect callback,
+     * heartbeat checker), but will only perform the disposal once per connection.
+     */
     fun disposeCurrentConnection() {
-        rsocket?.dispose()
-        rsocket = null
-        clientSettingsService.clearSettings()
+        connectionLock.withLock {
+            val currentRSocket = rsocket
+            if (currentRSocket != null) {
+                rsocket = null
+                currentRSocket.dispose()
+                clientSettingsService.clearSettings()
+            }
+        }
     }
 
     fun disposeClient() {

framework-client-messaging/src/main/java/io/axoniq/platform/framework/client/ServerProcessorReporter.kt

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2022-2025. AxonIQ B.V.
+ * Copyright (c) 2022-2026. AxonIQ B.V.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -50,7 +50,15 @@ class ServerProcessorReporter(
     }
 
     private fun report() {
-        client.sendReport(Routes.EventProcessor.REPORT, processorReportCreator.createReport()).block()
+        if (!client.isConnected()) {
+            return
+        }
+        client.sendReport(Routes.EventProcessor.REPORT, processorReportCreator.createReport())
+                .doOnError { e ->
+                    logger.debug { "Failed to send processor report: ${e.message}" }
+                }
+                .onErrorComplete()
+                .subscribe()
     }
 
     override fun onDisconnected() {

framework-client-messaging/src/main/java/io/axoniq/platform/framework/messaging/HandlerMetricsRegistry.kt

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2022-2025. AxonIQ B.V.
+ * Copyright (c) 2022-2026. AxonIQ B.V.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -68,12 +68,21 @@ class HandlerMetricsRegistry(
     override fun onConnectedWithSettings(settings: ClientSettingsV2) {
         logger.debug { "Sending handler information every ${settings.handlerReportInterval}ms to Axoniq Platform" }
         this.reportTask = executor.scheduleAtFixedRate({
+            if (!axoniqConsoleRSocketClient.isConnected()) {
+                return@scheduleAtFixedRate
+            }
             try {
                 val stats = getStats()
-                logger.debug { "${"Sending metrics: {}"} $stats" }
+                logger.debug { "Sending metrics: $stats" }
+                axoniqConsoleRSocketClient.sendReport(Routes.MessageFlow.STATS, stats)
+                        .doOnError { e ->
+                            logger.debug { "Failed to send handler metrics: ${e.message}" }
+                        }
+                        .onErrorComplete()
+                        .subscribe()
                 axoniqConsoleRSocketClient.sendReport(Routes.MessageFlow.STATS, stats).block()
             } catch (e: Exception) {
-                logger.warn { "${"No metrics could be reported to Axoniq Platform: {}"} ${e.message}" }
+                logger.warn { "No metrics could be reported to AxonIQ Console: ${e.message}" }
             }
         }, 0, settings.handlerReportInterval, TimeUnit.MILLISECONDS)
     }