Fixed a situation where rappidly restarting the adapter could lead to two open clients producing multiple tag reads (#1126)

Author: codepitbull

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

@@ -90,7 +90,7 @@ class OpcUaClientConnection {
         this.lastKnownSubscriptionId = lastSubscriptionId;
     }
 
-    @NotNull boolean start(final ParsedConfig parsedConfig) {
+    @NotNull synchronized boolean start(final ParsedConfig parsedConfig) {
         final var subscriptionIdOptional = Optional.ofNullable(lastKnownSubscriptionId.get());
         log.debug("Subscribing to OPC UA client with subscriptionId: {}", subscriptionIdOptional.orElse(null));
         final OpcUaClient client;
@@ -141,7 +141,7 @@ class OpcUaClientConnection {
         return true;
     }
 
-    void stop() {
+    synchronized void stop() {
         log.info("Stopping OPC UA client");
         final ConnectionContext ctx = context.get();
         if(ctx != null) {

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

@@ -95,7 +95,7 @@ public OpcUaProtocolAdapter(
     }
 
     @Override
-    public void start(
+    public synchronized void start(
             final @NotNull ProtocolAdapterStartInput input,
             final @NotNull ProtocolAdapterStartOutput output) {
         log.info("Starting OPC UA protocol adapter {}", adapterId);
@@ -112,9 +112,15 @@ public void start(
             return;
         }
 
+        if (opcUaClientConnection != null) {
+            log.warn("Tried starting already started OPC UA protocol adapter {}", adapterId);
+            output.startedSuccessfully();
+            return;
+        }
+
         protocolAdapterState.setConnectionStatus(ProtocolAdapterState.ConnectionStatus.DISCONNECTED);
 
-        final var newOpcUaClientConnection = new OpcUaClientConnection(
+        opcUaClientConnection = new OpcUaClientConnection(
                 adapterId,
                 tagList,
                 protocolAdapterState,
@@ -126,37 +132,32 @@ public void start(
                 lastSubscriptionId);
 
         CompletableFuture
-                .supplyAsync(() -> newOpcUaClientConnection.start(parsedConfig))
+                .supplyAsync(() -> opcUaClientConnection.start(parsedConfig))
                 .whenComplete((success, throwable) -> {
-                    if(throwable != null) {
+                    if(!success || throwable != null) {
+                        this.opcUaClientConnection = null;
                         protocolAdapterState.setConnectionStatus(ProtocolAdapterState.ConnectionStatus.ERROR);
                         log.error("Failed to start OPC UA client", throwable);
                         return;
                     }
-                    if(success) {
-                        this.opcUaClientConnection = newOpcUaClientConnection;
-                    } else {
-                        protocolAdapterState.setConnectionStatus(ProtocolAdapterState.ConnectionStatus.ERROR);
-                        log.error("Failed to start OPC UA client", throwable);
-                    }
                 });
 
         log.info("Successfully started OPC UA protocol adapter {}", adapterId);
         output.startedSuccessfully();
     }
 
     @Override
-    public void stop(final @NotNull ProtocolAdapterStopInput input, final @NotNull ProtocolAdapterStopOutput output) {
+    public synchronized void stop(final @NotNull ProtocolAdapterStopInput input, final @NotNull ProtocolAdapterStopOutput output) {
         log.info("Stopping OPC UA protocol adapter {}", adapterId);
         final var tempOpcUaClientConnection = opcUaClientConnection;
         if(tempOpcUaClientConnection != null) {
+            opcUaClientConnection = null;
             tempOpcUaClientConnection.stop();
             output.stoppedSuccessfully();
         } else {
             log.info("Tried stopping stopped OPC UA protocol adapter {}", adapterId);
             output.stoppedSuccessfully();
         }
-
     }
 
     @Override

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

@@ -107,6 +107,7 @@ public void onDataReceived(
             try {
                 protocolAdapterMetricsService.increment(Constants.METRIC_SUBSCRIPTION_DATA_RECEIVED_COUNT);
                 final String payload = extractPayload(client, values.get(i));
+                log.error("RECEIVED " + payload.replace("\n","") + " " + this.hashCode());
                 tagStreamingService.feed(tn, List.of(dataPointFactory.createJsonDataPoint(tn, payload)));
             } catch (final Throwable e) {
                 protocolAdapterMetricsService.increment(Constants.METRIC_SUBSCRIPTION_DATA_ERROR_COUNT);