Merge pull request #277 from hivemq/bugfix/eip-warnings

fix intermittent warnings in adapter
hivemq · Feb 5, 2024 · 5a63e26 · 5a63e26
2 parents fa0ab67 + a2987f2
commit 5a63e26
Show file tree

Hide file tree

Showing 2 changed files with 34 additions and 3 deletions.
diff --git a/...-module-plc4x/src/main/java/com/hivemq/edge/adapters/plc4x/impl/AbstractPlc4xAdapter.java b/...-module-plc4x/src/main/java/com/hivemq/edge/adapters/plc4x/impl/AbstractPlc4xAdapter.java
@@ -15,6 +15,7 @@
 import org.apache.plc4x.java.api.messages.PlcReadResponse;
 import org.apache.plc4x.java.api.types.PlcResponseCode;
 import org.apache.plc4x.java.api.value.PlcValue;
+import org.apache.plc4x.java.spi.messages.DefaultPlcReadResponse;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -167,7 +168,7 @@ protected CompletableFuture<ProtocolAdapterDataSample<T>> onSamplerInvoked(
                 return CompletableFuture.failedFuture(e);
             }
         }
-        return CompletableFuture.completedFuture(null);
+        return CompletableFuture.completedFuture( new ProtocolAdapterDataSample(subscription));
     }
 
 
@@ -238,7 +239,7 @@ public static String nullSafe(final @Nullable Object o) {
     protected ProtocolAdapterDataSample processReadResponse(
             final @NotNull T.Subscription subscription, final @NotNull PlcReadResponse readEvent) {
         //it is possible that the read response does not contain any values at all, leading to unexpected error states, especially with EIP adapter
-        if (readEvent.getNumberOfValues(subscription.getTagName()) > 0) {
+        if (!(readEvent instanceof DefaultPlcReadResponse) || ((DefaultPlcReadResponse) readEvent).getValues().containsKey(subscription.getTagName())) {
             PlcResponseCode responseCode = readEvent.getResponseCode(subscription.getTagName());
             if (responseCode == PlcResponseCode.OK) {
                 if (getConnectionStatus() == ConnectionStatus.ERROR) {
@@ -248,7 +249,7 @@ protected ProtocolAdapterDataSample processReadResponse(
                 return processPlcFieldData(subscription, Plc4xDataUtils.readDataFromReadResponse(readEvent));
             }
         }
-        return null;
+        return new ProtocolAdapterDataSample(subscription);
     }
 
     protected ProtocolAdapterDataSample processPlcFieldData(

diff --git a/...dule-plc4x/src/main/java/com/hivemq/edge/adapters/plc4x/types/eip/EIPProtocolAdapter.java b/...dule-plc4x/src/main/java/com/hivemq/edge/adapters/plc4x/types/eip/EIPProtocolAdapter.java
@@ -18,11 +18,15 @@
 import com.codahale.metrics.MetricRegistry;
 import com.hivemq.edge.adapters.plc4x.impl.AbstractPlc4xAdapter;
 import com.hivemq.edge.adapters.plc4x.model.Plc4xAdapterConfig;
+import com.hivemq.edge.modules.adapters.data.ProtocolAdapterDataSample;
 import com.hivemq.edge.modules.api.adapters.ProtocolAdapterInformation;
+import com.hivemq.edge.modules.config.impl.AbstractProtocolAdapterConfig;
 import com.hivemq.extension.sdk.api.annotations.NotNull;
+import org.apache.plc4x.java.api.messages.PlcReadResponse;
 
 import java.util.HashMap;
 import java.util.Map;
+import java.util.concurrent.CompletableFuture;
 
 /**
  * @author HiveMQ Adapter Generator
@@ -61,4 +65,30 @@ protected Map<String, String> createQueryStringParams(final @NotNull EIPAdapterC
         map.put("bigEndian", "false");
         return map;
     }
+
+    @Override
+    protected CompletableFuture<ProtocolAdapterDataSample<EIPAdapterConfig>> onSamplerInvoked(
+            final EIPAdapterConfig config, final AbstractProtocolAdapterConfig.Subscription subscription) {
+        if (!(subscription instanceof EIPAdapterConfig.Subscription)) {
+            throw new IllegalStateException("Subscription configuration is not of correct type Ethernet/IP");
+        }
+        if (connection.isConnected()) {
+            try {
+                CompletableFuture<? extends PlcReadResponse> request =
+                        connection.read((Plc4xAdapterConfig.Subscription) subscription);
+                return request.thenApply(response -> (ProtocolAdapterDataSample<EIPAdapterConfig>) processReadResponse((EIPAdapterConfig.Subscription) subscription,
+                        response)).exceptionally(throwable -> {
+                    if (throwable instanceof InterruptedException ||
+                            throwable.getCause() instanceof InterruptedException) {
+                        return new ProtocolAdapterDataSample<EIPAdapterConfig>(subscription);
+                    }
+                    throw new RuntimeException(throwable);
+                });
+
+            } catch (Exception e) {
+                return CompletableFuture.failedFuture(e);
+            }
+        }
+        return CompletableFuture.completedFuture(new ProtocolAdapterDataSample<EIPAdapterConfig>(subscription));
+    }
 }