Disconnect monitoring

Author: codepitbull

modules/hivemq-edge-module-opcua/src/main/java/com/hivemq/edge/adapters/opcua/OpcUaClientConnection.java

@@ -47,6 +47,9 @@
 import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicReference;
@@ -55,6 +58,7 @@
 
 public class OpcUaClientConnection {
     private static final @NotNull Logger log = LoggerFactory.getLogger(OpcUaClientConnection.class);
+    private static final int HEALTH_CHECK_INTERVAL_SECONDS = 30;
 
     private final @NotNull OpcUaSpecificAdapterConfig config;
     private final @NotNull List<OpcuaTag> tags;
@@ -66,6 +70,8 @@ public class OpcUaClientConnection {
     private final @NotNull ProtocolAdapterMetricsService protocolAdapterMetricsService;
 
     private final @NotNull AtomicReference<ConnectionContext> context = new AtomicReference<>();
+    private final @NotNull ScheduledExecutorService healthCheckScheduler = Executors.newSingleThreadScheduledExecutor();
+    private final @NotNull AtomicReference<ScheduledFuture<?>> healthCheckFuture = new AtomicReference<>();
 
     OpcUaClientConnection(
             final @NotNull String adapterId,
@@ -202,12 +208,20 @@ synchronized boolean start(final ParsedConfig parsedConfig) {
         context.set(new ConnectionContext(subscription.getClient(), faultListener, activityListener));
         protocolAdapterState.setConnectionStatus(ProtocolAdapterState.ConnectionStatus.CONNECTED);
         client.addSessionActivityListener(activityListener);
+
+        // Schedule periodic health check to detect stale connections
+        scheduleHealthCheck();
+
         log.info("Client created and connected successfully");
         return true;
     }
 
     synchronized void stop() {
         log.info("Stopping OPC UA client");
+
+        // Cancel health check
+        cancelHealthCheck();
+
         final ConnectionContext ctx = context.get();
         if(ctx != null) {
             quietlyCloseClient(ctx.client(),true,  ctx.faultListener(), ctx.activityListener());
@@ -217,6 +231,19 @@ synchronized void stop() {
 
     void destroy() {
         log.info("Destroying OPC UA client");
+
+        // Cancel health check and shutdown scheduler
+        cancelHealthCheck();
+        healthCheckScheduler.shutdown();
+        try {
+            if (!healthCheckScheduler.awaitTermination(5, TimeUnit.SECONDS)) {
+                healthCheckScheduler.shutdownNow();
+            }
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            healthCheckScheduler.shutdownNow();
+        }
+
         final ConnectionContext ctx = context.get();
         if(ctx != null) {
             quietlyCloseClient(ctx.client(), false, ctx.faultListener(), ctx.activityListener());
@@ -283,6 +310,57 @@ private static void quietlyCloseClient(
         }
     }
 
+    /**
+     * Schedules periodic health checks to detect stale connections.
+     * Runs every HEALTH_CHECK_INTERVAL_SECONDS to verify session is active.
+     */
+    private void scheduleHealthCheck() {
+        final ScheduledFuture<?> future = healthCheckScheduler.scheduleAtFixedRate(() -> {
+            try {
+                final ConnectionContext ctx = context.get();
+                if (ctx == null) {
+                    log.trace("Health check skipped - client not connected for adapter '{}'", adapterId);
+                    return;
+                }
+
+                final OpcUaClient client = ctx.client();
+                if (client.getSession().isEmpty()) {
+                    log.warn("Health check failed for adapter '{}' - session is not active", adapterId);
+                    eventService
+                            .createAdapterEvent(adapterId, PROTOCOL_ID_OPCUA)
+                            .withMessage("Connection health check failed for adapter '" + adapterId + "' - session inactive")
+                            .withSeverity(Event.SEVERITY.WARN)
+                            .fire();
+                    protocolAdapterState.setConnectionStatus(ProtocolAdapterState.ConnectionStatus.ERROR);
+                } else {
+                    log.trace("Health check passed for adapter '{}' - session is active", adapterId);
+                }
+            } catch (Exception e) {
+                log.warn("Health check exception for adapter '{}'", adapterId, e);
+            }
+        }, HEALTH_CHECK_INTERVAL_SECONDS, HEALTH_CHECK_INTERVAL_SECONDS, TimeUnit.SECONDS);
+
+        // Store future and cancel any existing health check
+        final ScheduledFuture<?> oldFuture = healthCheckFuture.getAndSet(future);
+        if (oldFuture != null && !oldFuture.isDone()) {
+            oldFuture.cancel(false);
+        }
+
+        log.debug("Scheduled connection health check every {} seconds for adapter '{}'",
+                HEALTH_CHECK_INTERVAL_SECONDS, adapterId);
+    }
+
+    /**
+     * Cancels any pending health check.
+     */
+    private void cancelHealthCheck() {
+        final ScheduledFuture<?> future = healthCheckFuture.getAndSet(null);
+        if (future != null && !future.isDone()) {
+            future.cancel(false);
+            log.debug("Cancelled health check for adapter '{}'", adapterId);
+        }
+    }
+
     private record ConnectionContext(@NotNull OpcUaClient client, @NotNull ServiceFaultListener faultListener, @NotNull SessionActivityListener activityListener) {
     }
 }

modules/hivemq-edge-module-opcua/src/main/java/com/hivemq/edge/adapters/opcua/client/OpcUaClientConfigurator.java

@@ -18,6 +18,7 @@
 import com.hivemq.edge.adapters.opcua.Constants;
 import org.eclipse.milo.opcua.sdk.client.OpcUaClientConfigBuilder;
 import org.eclipse.milo.opcua.stack.core.types.builtin.LocalizedText;
+import org.eclipse.milo.opcua.stack.core.types.builtin.unsigned.UInteger;
 import org.jetbrains.annotations.NotNull;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -54,8 +55,14 @@ public void accept(final @NotNull OpcUaClientConfigBuilder configBuilder) {
                 .setApplicationName(LocalizedText.english(Constants.OPCUA_APPLICATION_NAME))
                 .setApplicationUri(applicationUri)
                 .setProductUri(Constants.OPCUA_PRODUCT_URI)
-                .setSessionName(() -> Constants.OPCUA_SESSION_NAME_PREFIX + adapterId);
+                .setSessionName(() -> Constants.OPCUA_SESSION_NAME_PREFIX + adapterId)
+                // Configure timeouts to prevent silent disconnects
+                .setSessionTimeout(UInteger.valueOf(120_000))  // 2 minutes - session renewal interval
+                .setRequestTimeout(UInteger.valueOf(30_000))   // 30 seconds - request timeout
+                .setKeepAliveInterval(UInteger.valueOf(10_000)) // 10 seconds - keep-alive ping interval
+                .setKeepAliveFailuresAllowed(UInteger.valueOf(3)); // 3 failures = 30s before disconnect
 
+        log.info("Configured OPC UA timeouts: session=120s, request=30s, keepAlive=10s, failuresAllowed=3");
         log.info("TLS is enabled: {}", parsedConfig.tlsEnabled());
         if (parsedConfig.tlsEnabled()) {
             if (log.isDebugEnabled()) {