Fix publish-on-change (#1073)

Author: codepitbull

hivemq-edge/src/main/java/com/hivemq/edge/modules/adapters/impl/polling/PollingTask.java

@@ -82,7 +82,7 @@ public void run() {
                     }
                 }
             });
-        } catch (Throwable t) {
+        } catch (final Throwable t) {
             // the sampler shouldn't throw a exception, but better safe than sorry as we might to miss rescheduling the task otherwise.
             handleExceptionDuringPolling(t);
         }
@@ -96,10 +96,10 @@ private void handleInterruptionException(final @NotNull Throwable throwable) {
         //-- Job was killed by the framework as it took too long
         //-- Do not call back to the job here (notify) since it will
         //-- Not respond and we dont want to block other polls
-        int errorCountTotal = watchdogErrorCount.incrementAndGet();
-        boolean stopBecauseOfTooManyErrors =
+        final var errorCountTotal = watchdogErrorCount.incrementAndGet();
+        final var stopBecauseOfTooManyErrors =
                 errorCountTotal > InternalConfigurations.ADAPTER_RUNTIME_WATCHDOG_TIMEOUT_ERRORS_BEFORE_INTERRUPT.get();
-        final long milliSecondsSinceLastPoll = TimeUnit.NANOSECONDS.toMillis(nanoTimeProvider.nanoTime() - nanosOfLastPolling);
+        final var milliSecondsSinceLastPoll = TimeUnit.NANOSECONDS.toMillis(nanoTimeProvider.nanoTime() - nanosOfLastPolling);
         if (stopBecauseOfTooManyErrors) {
             log.warn(
                     "Detected bad system process {} in sampler {} - terminating process to maintain health ({}ms runtime)",
@@ -123,7 +123,7 @@ private void handleInterruptionException(final @NotNull Throwable throwable) {
 
 
     private void handleExceptionDuringPolling(final @NotNull Throwable throwable) {
-        int errorCountTotal = applicationErrorCount.incrementAndGet();
+        final int errorCountTotal = applicationErrorCount.incrementAndGet();
         final int maxErrorsBeforeRemoval = sampler.getMaxErrorsBeforeRemoval();
         // case 1: Unlimited retry (maxErrorsBeforeRemoval < 0) or less errors than the limit
         if (maxErrorsBeforeRemoval < 0 || errorCountTotal <= maxErrorsBeforeRemoval) {
@@ -149,18 +149,18 @@ private void handleExceptionDuringPolling(final @NotNull Throwable throwable) {
     }
 
     private void notifyOnError(
-            final @NotNull ProtocolAdapterPollingSampler sampler, final @NotNull Throwable t, boolean continuing) {
+            final @NotNull ProtocolAdapterPollingSampler sampler, final @NotNull Throwable t, final boolean continuing) {
         try {
             sampler.error(t, continuing);
-        } catch (Throwable samplerError) {
+        } catch (final Throwable samplerError) {
             if (log.isInfoEnabled()) {
                 log.info("Sampler Encountered Error In Notification", samplerError);
             }
         }
     }
 
-    private void reschedule(int errorCountTotal) {
-        long pollDuration = TimeUnit.NANOSECONDS.toMillis(nanoTimeProvider.nanoTime() - nanosOfLastPolling);
+    private void reschedule(final int errorCountTotal) {
+        final long pollDuration = TimeUnit.NANOSECONDS.toMillis(nanoTimeProvider.nanoTime() - nanosOfLastPolling);
         final long delayInMillis = sampler.getPeriod() - pollDuration;
         // a negative delay means that the last polling attempt took longer to be processed than the specified delay between polls
         if (delayInMillis < 0) {
@@ -177,24 +177,24 @@ private void reschedule(int errorCountTotal) {
                     .fire();
         }
 
-        long nonNegativeDelay = Math.max(0, delayInMillis);
+        final long nonNegativeDelay = Math.max(0, delayInMillis);
 
         if (errorCountTotal == 0) {
             schedule(nonNegativeDelay);
         } else {
-            long backoff = getBackoff(errorCountTotal,
+            final long backoff = getBackoff(errorCountTotal,
                     InternalConfigurations.ADAPTER_RUNTIME_MAX_APPLICATION_ERROR_BACKOFF.get());
-            long effectiveDelay = Math.max(nonNegativeDelay, backoff);
+            final long effectiveDelay = Math.max(nonNegativeDelay, backoff);
             schedule(effectiveDelay);
         }
     }
 
     @VisibleForTesting
-    void schedule(long nonNegativeDelay) {
+    void schedule(final long nonNegativeDelay) {
         if (continueScheduling.get()) {
             try {
                 scheduledExecutorService.schedule(this, nonNegativeDelay, TimeUnit.MILLISECONDS);
-            } catch (RejectedExecutionException rejectedExecutionException) {
+            } catch (final RejectedExecutionException rejectedExecutionException) {
                 // ignore. This is fine during shutdown.
             }
         }
@@ -205,7 +205,7 @@ private void resetErrorStats() {
         watchdogErrorCount.set(0);
     }
 
-    private static long getBackoff(int errorCount, long max) {
+    private static long getBackoff(final int errorCount, final long max) {
         //-- This will backoff up to a max of about a day (unless the max provided is less)
         long f = (long) (Math.pow(2, Math.min(errorCount, 20)) * 100);
         f += ThreadLocalRandom.current().nextInt(0, errorCount * 100);

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

@@ -18,28 +18,37 @@
 import com.hivemq.adapter.sdk.api.data.DataPoint;
 import org.jetbrains.annotations.NotNull;
 
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 /**
  * This class is here TEMPORARY, the functionality will be moved into NorthboundMappings
  */
 public class PublishChangedDataOnlyHandler {
-    private final @NotNull Map<String, List<DataPoint>> lastSamples = new ConcurrentHashMap<>();
+    private final @NotNull Map<String, Set<DataPoint>> lastSamples = new ConcurrentHashMap<>();
 
-    public boolean replaceIfValueIsNew(final @NotNull String tagName, final @NotNull List<DataPoint> newValue) {
-        final var computedValue = lastSamples.compute(tagName, (key,value) -> {
+    public boolean checkIfValuesHaveChangedSinceLastInvocation(final @NotNull String tagName, final @NotNull List<DataPoint> newValue) {
+        final var replaced = new AtomicBoolean(false);
+        lastSamples.compute(tagName, (key,value) -> {
             if (value == null) {
-                return newValue;
-            } else if (value.equals(newValue)) {
-                return value;
+                replaced.set(true);
+                return new HashSet<>(newValue);
             } else {
-                return newValue;
+                final var newSet = new HashSet<>(newValue);
+                if(!newSet.equals(value)) {
+                    replaced.set(true);
+                    return new HashSet<>(newValue);
+                } else {
+                    return value;
+                }
             }
         });
 
-        return newValue != computedValue;
+        return replaced.get();
     }
 
     public void clear() {

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

@@ -111,7 +111,7 @@ public void poll(final @NotNull BatchPollingInput pollingInput, final @NotNull B
                                         .collect(Collectors.groupingBy(DataPoint::getTagName,
                                                 Collectors.mapping(Function.identity(), Collectors.toList())));
                                 tagsToValueList.forEach((tagName,tagValues) -> {
-                                    if (lastSamples.replaceIfValueIsNew(tagName, tagValues)) {
+                                    if (lastSamples.checkIfValuesHaveChangedSinceLastInvocation(tagName, tagValues)) {
                                         tagValues.forEach(pollingOutput::addDataPoint);
                                     }
                                 });

modules/hivemq-edge-module-plc4x/src/test/java/com/hivemq/edge/adapters/plc4x/PublishChangedDataOnlyHandlerTest.java

@@ -0,0 +1,38 @@
+/*
+ * Copyright 2023-present HiveMQ GmbH
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.hivemq.edge.adapters.plc4x;
+
+import com.hivemq.edge.modules.adapters.impl.factories.AdapterFactoriesImpl;
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+public class PublishChangedDataOnlyHandlerTest {
+
+    @Test
+    public void test() {
+        final var factory = new AdapterFactoriesImpl();
+        final var dataPointFactory = factory.dataPointFactory();
+        final var toTest = new PublishChangedDataOnlyHandler();
+        final var initial = toTest.checkIfValuesHaveChangedSinceLastInvocation("tag1", List.of(dataPointFactory.create("tag1", "value1")));
+        final var secondTry = toTest.checkIfValuesHaveChangedSinceLastInvocation("tag1", List.of(dataPointFactory.create("tag1", "value1")));
+
+        assertThat(initial).isTrue();
+        assertThat(secondTry).isFalse();
+    }
+}