Extract reconnect and isHealthy methods

Author: codepitbull

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

@@ -204,7 +204,7 @@ synchronized boolean start(final ParsedConfig parsedConfig) {
         final var subscription = subscriptionOptional.get();
         log.trace("Creating Subscription for OPC UA client");
 
-        context.set(new ConnectionContext(subscription.getClient(), faultListener, activityListener));
+        context.set(new ConnectionContext(subscription.getClient(), faultListener, activityListener, subscriptionLifecycleHandler));
         protocolAdapterState.setConnectionStatus(ProtocolAdapterState.ConnectionStatus.CONNECTED);
         client.addSessionActivityListener(activityListener);
 
@@ -250,6 +250,40 @@ void destroy() {
         }
     }
 
+    /**
+     * Checks if the connection is healthy by verifying both:
+     * 1. Session is active (client connection exists and session is present)
+     * 2. Keep-alive messages are being received (subscription is healthy)
+     *
+     * @return true if connection is healthy, false otherwise
+     */
+    public boolean isHealthy() {
+        final ConnectionContext ctx = context.get();
+        if (ctx == null) {
+            log.debug("Connection health check failed: no connection context");
+            return false;
+        }
+
+        try {
+            // Check 1: Session is active
+            if (ctx.client().getSession().isEmpty()) {
+                log.debug("Connection health check failed: session inactive for adapter '{}'", adapterId);
+                return false;
+            }
+
+            // Check 2: Keep-alive is healthy
+            if (!ctx.subscriptionHandler().isKeepAliveHealthy()) {
+                log.debug("Connection health check failed: keep-alive timeout for adapter '{}'", adapterId);
+                return false;
+            }
+
+            return true;
+        } catch (final UaException e) {
+            log.debug("Connection health check failed with exception for adapter '{}'", adapterId, e);
+            return false;
+        }
+    }
+
     @NotNull Optional<OpcUaClient> client() {
         final ConnectionContext ctx = context.get();
         if(ctx != null) {
@@ -361,6 +395,9 @@ private void cancelHealthCheck() {
         }
     }
 
-    private record ConnectionContext(@NotNull OpcUaClient client, @NotNull ServiceFaultListener faultListener, @NotNull SessionActivityListener activityListener) {
+    private record ConnectionContext(@NotNull OpcUaClient client,
+                                     @NotNull ServiceFaultListener faultListener,
+                                     @NotNull SessionActivityListener activityListener,
+                                     @NotNull OpcUaSubscriptionLifecycleHandler subscriptionHandler) {
     }
 }

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

@@ -27,6 +27,7 @@
 import com.hivemq.adapter.sdk.api.model.ProtocolAdapterStopOutput;
 import com.hivemq.adapter.sdk.api.schema.TagSchemaCreationInput;
 import com.hivemq.adapter.sdk.api.schema.TagSchemaCreationOutput;
+import com.hivemq.adapter.sdk.api.services.ModuleServices;
 import com.hivemq.adapter.sdk.api.services.ProtocolAdapterMetricsService;
 import com.hivemq.adapter.sdk.api.state.ProtocolAdapterState;
 import com.hivemq.adapter.sdk.api.writing.WritingContext;
@@ -78,6 +79,10 @@ public class OpcUaProtocolAdapter implements WritingProtocolAdapter {
     private final @NotNull ScheduledExecutorService retryScheduler = Executors.newSingleThreadScheduledExecutor();
     private final @NotNull AtomicReference<ScheduledFuture<?>> retryFuture = new AtomicReference<>();
 
+    // Stored for reconnection - set during start()
+    private volatile ParsedConfig parsedConfig;
+    private volatile ModuleServices moduleServices;
+
     public OpcUaProtocolAdapter(
             final @NotNull ProtocolAdapterInformation adapterInformation,
             final @NotNull ProtocolAdapterInput<OpcUaSpecificAdapterConfig> input) {
@@ -111,6 +116,9 @@ public synchronized void start(
             return;
         } else if (result instanceof Success<ParsedConfig, String>(final ParsedConfig result1)) {
             parsedConfig = result1;
+            // Store for reconnection
+            this.parsedConfig = result1;
+            this.moduleServices = input.moduleServices();
         } else {
             output.failStart(new IllegalStateException("Unexpected result type: " + result.getClass().getName()),
                     "Failed to parse configuration for OPC UA client");
@@ -154,6 +162,10 @@ public synchronized void stop(
         // Cancel any pending retries
         cancelRetry();
 
+        // Clear stored configuration to prevent reconnection after stop
+        this.parsedConfig = null;
+        this.moduleServices = null;
+
         final OpcUaClientConnection conn = opcUaClientConnection.getAndSet(null);
         if (conn != null) {
             conn.stop();
@@ -163,6 +175,56 @@ public synchronized void stop(
         output.stoppedSuccessfully();
     }
 
+    /**
+     * Triggers reconnection by stopping the current connection and creating a new one.
+     * Used for runtime reconnection when health check detects issues.
+     * Requires that start() has been called previously to initialize parsedConfig and moduleServices.
+     */
+    private void reconnect() {
+        log.info("Reconnecting OPC UA adapter '{}'", adapterId);
+
+        // Verify we have the necessary configuration
+        if (parsedConfig == null || moduleServices == null) {
+            log.error("Cannot reconnect OPC UA adapter '{}' - adapter has not been started yet", adapterId);
+            return;
+        }
+
+        // Cancel any pending retries
+        cancelRetry();
+
+        // Stop and clean up current connection
+        final OpcUaClientConnection oldConn = opcUaClientConnection.getAndSet(null);
+        if (oldConn != null) {
+            oldConn.stop();
+            log.debug("Stopped old connection for OPC UA adapter '{}'", adapterId);
+        }
+
+        // Create new connection
+        final OpcUaClientConnection newConn = new OpcUaClientConnection(adapterId,
+                tagList,
+                protocolAdapterState,
+                moduleServices.protocolAdapterTagStreamingService(),
+                dataPointFactory,
+                moduleServices.eventService(),
+                protocolAdapterMetricsService,
+                config);
+
+        // Set as current connection and attempt connection with retry logic
+        protocolAdapterState.setConnectionStatus(ProtocolAdapterState.ConnectionStatus.DISCONNECTED);
+        if (opcUaClientConnection.compareAndSet(null, newConn)) {
+            // Create a minimal ProtocolAdapterStartInput for attemptConnection
+            final ProtocolAdapterStartInput input = new ProtocolAdapterStartInput() {
+                @Override
+                public ModuleServices moduleServices() {
+                    return moduleServices;
+                }
+            };
+            attemptConnection(newConn, parsedConfig, input);
+        } else {
+            log.warn("OPC UA adapter '{}' reconnect failed - another connection was created concurrently", adapterId);
+        }
+    }
+
     @Override
     public void destroy() {
         log.info("Destroying OPC UA protocol adapter {}", adapterId);
@@ -369,7 +431,7 @@ private void scheduleRetry(
         final int retryIntervalSeconds = config.getRetryInterval();
         final ScheduledFuture<?> future = retryScheduler.schedule(() -> {
             // Check if adapter was stopped before retry executes
-            if (opcUaClientConnection.get() == null) {
+            if (this.parsedConfig == null || this.moduleServices == null) {
                 log.debug("OPC UA adapter '{}' retry cancelled - adapter was stopped", adapterId);
                 return;
             }
@@ -380,15 +442,16 @@ private void scheduleRetry(
             final OpcUaClientConnection newConn = new OpcUaClientConnection(adapterId,
                     tagList,
                     protocolAdapterState,
-                    input.moduleServices().protocolAdapterTagStreamingService(),
+                    this.moduleServices.protocolAdapterTagStreamingService(),
                     dataPointFactory,
-                    input.moduleServices().eventService(),
+                    this.moduleServices.eventService(),
                     protocolAdapterMetricsService,
                     config);
 
             // Set as current connection and attempt
+            protocolAdapterState.setConnectionStatus(ProtocolAdapterState.ConnectionStatus.DISCONNECTED);
             if (opcUaClientConnection.compareAndSet(null, newConn)) {
-                attemptConnection(newConn, parsedConfig, input);
+                attemptConnection(newConn, this.parsedConfig, input);
             } else {
                 log.debug("OPC UA adapter '{}' retry skipped - connection already exists", adapterId);
             }