Add better connection and timeout handling

codepitbull · codepitbull · commit 485695bb2861 · 2025-11-03T12:00:43.000+01:00
diff --git a/modules/hivemq-edge-module-opcua/src/main/java/com/hivemq/edge/adapters/opcua/OpcUaClientConnection.java b/modules/hivemq-edge-module-opcua/src/main/java/com/hivemq/edge/adapters/opcua/OpcUaClientConnection.java
@@ -45,10 +45,13 @@
 
 import java.util.List;
 import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicReference;
 
 import static com.hivemq.edge.adapters.opcua.Constants.PROTOCOL_ID_OPCUA;
-import static org.eclipse.milo.opcua.stack.core.types.builtin.unsigned.Unsigned.uint;
 
 public class OpcUaClientConnection {
     private static final @NotNull Logger log = LoggerFactory.getLogger(OpcUaClientConnection.class);
@@ -72,8 +75,7 @@ public class OpcUaClientConnection {
             final @NotNull DataPointFactory dataPointFactory,
             final @NotNull EventService eventService,
             final @NotNull ProtocolAdapterMetricsService protocolAdapterMetricsService,
-            final @NotNull OpcUaSpecificAdapterConfig config,
-            final @NotNull AtomicReference<UInteger> lastSubscriptionId) {
+            final @NotNull OpcUaSpecificAdapterConfig config) {
         this.config = config;
         this.tagStreamingService = tagStreamingService;
         this.dataPointFactory = dataPointFactory;
@@ -122,12 +124,56 @@ synchronized boolean start(final ParsedConfig parsedConfig) {
                         ignore -> {},
                         new OpcUaClientConfigurator(adapterId, parsedConfig));
             client.addFaultListener(faultListener);
-            client.connect();
+
+            // Add timeout to connection attempt to prevent hanging forever
+            // Wrap synchronous connect() call with CompletableFuture timeout
+            try {
+                CompletableFuture.runAsync(() -> {
+                    try {
+                        client.connect();
+                    } catch (final UaException e) {
+                        throw new RuntimeException(e);
+                    }
+                }).get(30, TimeUnit.SECONDS);
+                log.debug("OPC UA client connected successfully for adapter '{}'", adapterId);
+            } catch (final TimeoutException e) {
+                log.error("Connection timeout after 30 seconds for OPC UA adapter '{}'", adapterId);
+                eventService
+                        .createAdapterEvent(adapterId, PROTOCOL_ID_OPCUA)
+                        .withMessage("Connection timeout after 30 seconds for adapter '" + adapterId + "'")
+                        .withSeverity(Event.SEVERITY.ERROR)
+                        .fire();
+                protocolAdapterState.setConnectionStatus(ProtocolAdapterState.ConnectionStatus.ERROR);
+                quietlyCloseClient(client, false, faultListener, null);
+                return false;
+            } catch (final InterruptedException e) {
+                Thread.currentThread().interrupt();
+                log.error("Connection interrupted for OPC UA adapter '{}'", adapterId, e);
+                eventService
+                        .createAdapterEvent(adapterId, PROTOCOL_ID_OPCUA)
+                        .withMessage("Connection interrupted for adapter '" + adapterId + "'")
+                        .withSeverity(Event.SEVERITY.ERROR)
+                        .fire();
+                protocolAdapterState.setConnectionStatus(ProtocolAdapterState.ConnectionStatus.ERROR);
+                quietlyCloseClient(client, false, faultListener, null);
+                return false;
+            } catch (final ExecutionException e) {
+                final Throwable cause = e.getCause();
+                log.error("Connection failed for OPC UA adapter '{}'", adapterId, cause);
+                eventService
+                        .createAdapterEvent(adapterId, PROTOCOL_ID_OPCUA)
+                        .withMessage("Connection failed for adapter '" + adapterId + "': " + cause.getMessage())
+                        .withSeverity(Event.SEVERITY.ERROR)
+                        .fire();
+                protocolAdapterState.setConnectionStatus(ProtocolAdapterState.ConnectionStatus.ERROR);
+                quietlyCloseClient(client, false, faultListener, null);
+                return false;
+            }
         } catch (final UaException e) {
-            log.error("Unable to connect and subscribe to the OPC UA server for adapter '{}'", adapterId, e);
+            log.error("Unable to create OPC UA client for adapter '{}'", adapterId, e);
             eventService
                     .createAdapterEvent(adapterId, PROTOCOL_ID_OPCUA)
-                    .withMessage("Unable to connect and subscribe to the OPC UA server for adapter '" + adapterId + "'")
+                    .withMessage("Unable to create OPC UA client for adapter '" + adapterId + "'")
                     .withSeverity(Event.SEVERITY.ERROR)
                     .fire();
             protocolAdapterState.setConnectionStatus(ProtocolAdapterState.ConnectionStatus.ERROR);
diff --git a/modules/hivemq-edge-module-opcua/src/main/java/com/hivemq/edge/adapters/opcua/OpcUaProtocolAdapter.java b/modules/hivemq-edge-module-opcua/src/main/java/com/hivemq/edge/adapters/opcua/OpcUaProtocolAdapter.java
@@ -46,7 +46,6 @@
 import org.eclipse.milo.opcua.stack.core.types.builtin.NodeId;
 import org.eclipse.milo.opcua.stack.core.types.builtin.StatusCode;
 import org.eclipse.milo.opcua.stack.core.types.builtin.Variant;
-import org.eclipse.milo.opcua.stack.core.types.builtin.unsigned.UInteger;
 import org.jetbrains.annotations.NotNull;
 import org.jetbrains.annotations.VisibleForTesting;
 import org.slf4j.Logger;
@@ -55,12 +54,17 @@
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Function;
 import java.util.stream.Collectors;
 
 public class OpcUaProtocolAdapter implements WritingProtocolAdapter {
     private static final @NotNull Logger log = LoggerFactory.getLogger(OpcUaProtocolAdapter.class);
+    private static final int RETRY_DELAY_SECONDS = 30;
 
     private final @NotNull ProtocolAdapterInformation adapterInformation;
     private final @NotNull ProtocolAdapterState protocolAdapterState;
@@ -72,7 +76,8 @@ public class OpcUaProtocolAdapter implements WritingProtocolAdapter {
     private final @NotNull DataPointFactory dataPointFactory;
     private final @NotNull ProtocolAdapterMetricsService protocolAdapterMetricsService;
     private final @NotNull OpcUaSpecificAdapterConfig config;
-    private final @NotNull AtomicReference<UInteger> lastSubscriptionId = new AtomicReference<>();
+    private final @NotNull ScheduledExecutorService retryScheduler = Executors.newSingleThreadScheduledExecutor();
+    private final @NotNull AtomicReference<ScheduledFuture<?>> retryFuture = new AtomicReference<>();
 
     public OpcUaProtocolAdapter(
             final @NotNull ProtocolAdapterInformation adapterInformation,
@@ -100,13 +105,13 @@ public synchronized void start(
         log.info("Starting OPC UA protocol adapter {}", adapterId);
         final ParsedConfig parsedConfig;
         final var result = ParsedConfig.fromConfig(config);
-        if (result instanceof final Failure<ParsedConfig, String> failure) {
-            log.error("Failed to parse configuration for OPC UA client: {}", failure.failure());
-            output.failStart(new IllegalStateException(failure.failure()),
+        if (result instanceof Failure<ParsedConfig, String>(final String failure)) {
+            log.error("Failed to parse configuration for OPC UA client: {}", failure);
+            output.failStart(new IllegalStateException(failure),
                     "Failed to parse configuration for OPC UA client");
             return;
-        } else if (result instanceof final Success<ParsedConfig, String> success) {
-            parsedConfig = success.result();
+        } else if (result instanceof Success<ParsedConfig, String>(final ParsedConfig result1)) {
+            parsedConfig = result1;
         } else {
             output.failStart(new IllegalStateException("Unexpected result type: " + result.getClass().getName()),
                     "Failed to parse configuration for OPC UA client");
@@ -121,22 +126,23 @@ public synchronized void start(
                 dataPointFactory,
                 input.moduleServices().eventService(),
                 protocolAdapterMetricsService,
-                config,
-                lastSubscriptionId))) {
+                config))) {
 
             protocolAdapterState.setConnectionStatus(ProtocolAdapterState.ConnectionStatus.DISCONNECTED);
-            CompletableFuture.supplyAsync(() -> conn.start(parsedConfig)).whenComplete((success, throwable) -> {
-                if (!success || throwable != null) {
-                    this.opcUaClientConnection.set(null);
-                    protocolAdapterState.setConnectionStatus(ProtocolAdapterState.ConnectionStatus.ERROR);
-                    log.error("Failed to start OPC UA client", throwable);
-                }
-            });
 
+            // Attempt initial connection asynchronously
+            attemptConnection(conn, parsedConfig, input);
+
+            // Adapter starts successfully even if connection isn't established yet
+            // Hardware may come online later and automatic retry will connect
             log.info("Successfully started OPC UA protocol adapter {}", adapterId);
             output.startedSuccessfully();
         } else {
-            log.warn("Tried starting already started OPC UA protocol adapter {}", adapterId);
+            log.error("Cannot start OPC UA protocol adapter '{}' - adapter is already started", adapterId);
+            output.failStart(
+                new IllegalStateException("Adapter already started"),
+                "Cannot start already started adapter. Please stop the adapter first."
+            );
         }
     }
 
@@ -145,6 +151,10 @@ public synchronized void stop(
             final @NotNull ProtocolAdapterStopInput input,
             final @NotNull ProtocolAdapterStopOutput output) {
         log.info("Stopping OPC UA protocol adapter {}", adapterId);
+
+        // Cancel any pending retries
+        cancelRetry();
+
         final OpcUaClientConnection conn = opcUaClientConnection.getAndSet(null);
         if (conn != null) {
             conn.stop();
@@ -157,6 +167,21 @@ public synchronized void stop(
     @Override
     public void destroy() {
         log.info("Destroying OPC UA protocol adapter {}", adapterId);
+
+        // Cancel any pending retries
+        cancelRetry();
+
+        // Shutdown retry scheduler
+        retryScheduler.shutdown();
+        try {
+            if (!retryScheduler.awaitTermination(5, TimeUnit.SECONDS)) {
+                retryScheduler.shutdownNow();
+            }
+        } catch (final InterruptedException e) {
+            Thread.currentThread().interrupt();
+            retryScheduler.shutdownNow();
+        }
+
         final OpcUaClientConnection conn = opcUaClientConnection.getAndSet(null);
         if (conn != null) {
             CompletableFuture.runAsync(() -> {
@@ -300,4 +325,89 @@ public void createTagSchema(
     public @NotNull ProtocolAdapterState getProtocolAdapterState() {
         return protocolAdapterState;
     }
+
+    /**
+     * Attempts to establish connection to OPC UA server.
+     * On failure, schedules automatic retry after RETRY_DELAY_SECONDS.
+     */
+    private void attemptConnection(
+            final @NotNull OpcUaClientConnection conn,
+            final @NotNull ParsedConfig parsedConfig,
+            final @NotNull ProtocolAdapterStartInput input) {
+
+        CompletableFuture.supplyAsync(() -> conn.start(parsedConfig)).whenComplete((success, throwable) -> {
+            if (success && throwable == null) {
+                // Connection succeeded - cancel any pending retries
+                cancelRetry();
+                log.info("OPC UA adapter '{}' connected successfully", adapterId);
+            } else {
+                // Connection failed - clean up and schedule retry
+                this.opcUaClientConnection.set(null);
+                protocolAdapterState.setConnectionStatus(ProtocolAdapterState.ConnectionStatus.ERROR);
+
+                if (throwable != null) {
+                    log.warn("OPC UA adapter '{}' connection failed, will retry in {} seconds",
+                            adapterId, RETRY_DELAY_SECONDS, throwable);
+                } else {
+                    log.warn("OPC UA adapter '{}' connection returned false, will retry in {} seconds",
+                            adapterId, RETRY_DELAY_SECONDS);
+                }
+
+                // Schedule retry attempt
+                scheduleRetry(parsedConfig, input);
+            }
+        });
+    }
+
+    /**
+     * Schedules a retry attempt after RETRY_DELAY_SECONDS.
+     */
+    private void scheduleRetry(
+            final @NotNull ParsedConfig parsedConfig,
+            final @NotNull ProtocolAdapterStartInput input) {
+
+        final ScheduledFuture<?> future = retryScheduler.schedule(() -> {
+            // Check if adapter was stopped before retry executes
+            if (opcUaClientConnection.get() == null) {
+                log.debug("OPC UA adapter '{}' retry cancelled - adapter was stopped", adapterId);
+                return;
+            }
+
+            log.info("Retrying connection for OPC UA adapter '{}'", adapterId);
+
+            // Create new connection object for retry
+            final OpcUaClientConnection newConn = new OpcUaClientConnection(adapterId,
+                    tagList,
+                    protocolAdapterState,
+                    input.moduleServices().protocolAdapterTagStreamingService(),
+                    dataPointFactory,
+                    input.moduleServices().eventService(),
+                    protocolAdapterMetricsService,
+                    config);
+
+            // Set as current connection and attempt
+            if (opcUaClientConnection.compareAndSet(null, newConn)) {
+                attemptConnection(newConn, parsedConfig, input);
+            } else {
+                log.debug("OPC UA adapter '{}' retry skipped - connection already exists", adapterId);
+            }
+        }, RETRY_DELAY_SECONDS, TimeUnit.SECONDS);
+
+        // Store future so it can be cancelled if needed
+        final ScheduledFuture<?> oldFuture = retryFuture.getAndSet(future);
+        if (oldFuture != null && !oldFuture.isDone()) {
+            oldFuture.cancel(false);
+        }
+    }
+
+    /**
+     * Cancels any pending retry attempts.
+     */
+    private void cancelRetry() {
+        final ScheduledFuture<?> future = retryFuture.getAndSet(null);
+        if (future != null && !future.isDone()) {
+            future.cancel(false);
+            log.debug("Cancelled pending retry for OPC UA adapter '{}'", adapterId);
+        }
+    }
 }
