Error states were incorrectly handled in ADS. There is also a library bug complicating things

Author: codepitbull

modules/hivemq-edge-module-plc4x/src/main/java/com/hivemq/edge/adapters/plc4x/impl/AbstractPlc4xAdapter.java

@@ -45,14 +45,14 @@
 import org.slf4j.LoggerFactory;
 
 import java.util.Collections;
-import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 import java.util.function.Function;
 import java.util.stream.Collectors;
 
 import static com.hivemq.adapter.sdk.api.state.ProtocolAdapterState.ConnectionStatus.CONNECTED;
+import static com.hivemq.adapter.sdk.api.state.ProtocolAdapterState.ConnectionStatus.ERROR;
 
 /**
  * Abstract PLC4X implementation. Exposes core abstractions of the underlying framework so instances can be exposes
@@ -135,9 +135,28 @@ public void poll(final @NotNull BatchPollingInput pollingInput, final @NotNull B
     public void start(
             final @NotNull ProtocolAdapterStartInput input, final @NotNull ProtocolAdapterStartOutput output) {
         try {
-            // we do not subscribe anymore as no current adapter type supports it anyway
-            initConnection();
-            output.startedSuccessfully();
+            if (connection == null) {
+                synchronized (lock) {
+                    if (connection == null) {
+                        // we do not subscribe anymore as no current adapter type supports it anyway
+                        if (log.isTraceEnabled()) {
+                            log.trace("Creating new instance of Plc4x connector with {}.", adapterConfig);
+                        }
+                        final Plc4xConnection<T> connection = createConnection();
+                        if(connection.isConnected()) {
+                            protocolAdapterState.setConnectionStatus(CONNECTED);
+                            output.startedSuccessfully();
+                        } else {
+                            protocolAdapterState.setConnectionStatus(ERROR);
+                            output.failStart(new Plc4xException("Unable to connect to device"), "Unable to connect to device");
+                        }
+
+                        this.connection = connection;
+                    }
+                }
+            } else {
+                output.startedSuccessfully();
+            }
         } catch (final Exception e) {
             output.failStart(e, null);
         }
@@ -164,26 +183,6 @@ public void stop(final @NotNull ProtocolAdapterStopInput input, final @NotNull P
         return adapterInformation;
     }
 
-    private Plc4xConnection<T> initConnection() {
-        if (connection == null) {
-            synchronized (lock) {
-                if (connection == null) {
-                    try {
-                        if (log.isTraceEnabled()) {
-                            log.trace("Creating new instance of Plc4x connector with {}.", adapterConfig);
-                        }
-                        connection = createConnection();
-                        protocolAdapterState.setConnectionStatus(CONNECTED);
-                        return connection;
-                    } catch (final Plc4xException e) {
-                        throw new RuntimeException(e);
-                    }
-                }
-            }
-        }
-        return connection;
-    }
-
     protected @NotNull Plc4xConnection<T> createConnection() throws Plc4xException {
         return new Plc4xConnection<>(driverManager,
                 adapterConfig,

modules/hivemq-edge-module-plc4x/src/main/java/com/hivemq/edge/adapters/plc4x/impl/Plc4xConnection.java

@@ -32,7 +32,12 @@
 import org.slf4j.LoggerFactory;
 
 import java.util.List;
+import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 import java.util.function.Consumer;
 
 public abstract class Plc4xConnection<T extends Plc4XSpecificAdapterConfig<?>> {
@@ -76,23 +81,30 @@ public Plc4xConnection(
     }
 
     protected void initConnection() throws Plc4xException {
-        try {
-            if (plcConnection == null) {
-                synchronized (lock) {
-                    if (plcConnection == null) {
-                        final String connectionString = createConnectionString(config);
-                        if (log.isTraceEnabled()) {
-                            log.trace("Connecting via PLC4X to {}.", connectionString);
-                        }
-                        plcConnection = plcDriverManager.getConnectionManager().getConnection(connectionString);
+        if (plcConnection == null) {
+            synchronized (lock) {
+                if (plcConnection == null) {
+                    final String connectionString = createConnectionString(config);
+                    if (log.isTraceEnabled()) {
+                        log.trace("Connecting via PLC4X to {}.", connectionString);
+                    }
+                    try {
+                        plcConnection = CompletableFuture.supplyAsync(() -> {
+                                try {
+                                    return Optional.of(plcDriverManager.getConnectionManager().getConnection(connectionString));
+                                } catch (Throwable e) {
+                                    log.info("Error encountered connecting to external device", e);
+                                }
+                                return Optional.<PlcConnection>empty();
+                        })
+                        .get(2_000, TimeUnit.MILLISECONDS)
+                        .orElseThrow(() -> new Plc4xException("Unable to connect to device."));
+                    } catch (InterruptedException | ExecutionException | TimeoutException e) {
+                        log.error("Error encountered connecting to external device", e);
+                        throw new Plc4xException(e);
                     }
                 }
             }
-        } catch (PlcConnectionException e) {
-            if (log.isInfoEnabled()) {
-                log.info("Error encountered connecting to external device.", e);
-            }
-            throw new Plc4xException("Error connecting", e);
         }
     }