test: Add test for $. in destination

hivemq-edge/src/main/java/com/hivemq/bootstrap/services/EdgeCoreFactoryService.java

@@ -21,10 +21,23 @@
 import org.jetbrains.annotations.Nullable;
 
 public class EdgeCoreFactoryService {
+    private boolean pulseActivated;
 
     private @Nullable WritingServiceFactory writingServiceFactory;
     private @Nullable DataCombiningTransformationServiceFactory dataCombiningTransformationServiceFactory;
 
+    public EdgeCoreFactoryService() {
+        pulseActivated = false;
+    }
+
+    public boolean isPulseActivated() {
+        return pulseActivated;
+    }
+
+    public void setPulseActivated(final boolean pulseActivated) {
+        this.pulseActivated = pulseActivated;
+    }
+
     public void provideWritingServiceFactory(final @NotNull WritingServiceFactory writingServiceFactory) {
         this.writingServiceFactory = writingServiceFactory;
     }
@@ -39,6 +52,7 @@ public void provideDataCombiningTransformationServiceFactory(
     }
 
     public @Nullable DataCombiningTransformationServiceFactory getDataCombiningTransformationServiceFactory() {
-        return dataCombiningTransformationServiceFactory;
+        // If Pulse is activated, external data combining service is disable.
+        return pulseActivated ? null : dataCombiningTransformationServiceFactory;
     }
 }

hivemq-edge/src/main/java/com/hivemq/common/i18n/StringTemplate.java

@@ -0,0 +1,50 @@
+/*
+ *  Copyright 2019-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.common.i18n;
+
+import freemarker.template.Configuration;
+import freemarker.template.Template;
+import org.jetbrains.annotations.NotNull;
+
+import java.io.StringWriter;
+import java.nio.charset.StandardCharsets;
+import java.util.Locale;
+import java.util.Map;
+
+public final class StringTemplate {
+    private final static @NotNull Configuration CONFIGURATION = new Configuration(Configuration.VERSION_2_3_22);
+
+    static {
+        CONFIGURATION.setDefaultEncoding(StandardCharsets.UTF_8.name());
+        CONFIGURATION.setLocale(Locale.US);
+    }
+
+    private StringTemplate() {
+    }
+
+    public static @NotNull String format(
+            final @NotNull String stringTemplate,
+            final @NotNull Map<String, Object> arguments) {
+        try (final StringWriter stringWriter = new StringWriter();) {
+            final Template template = new Template(stringTemplate, stringTemplate, CONFIGURATION);
+            template.process(arguments, stringWriter);
+            return stringWriter.toString();
+        } catch (final Exception e) {
+            return e.getMessage();
+        }
+    }
+}

hivemq-edge/src/main/java/com/hivemq/pulse/status/PulseAgentStatusChangedListener.java

@@ -17,6 +17,7 @@
 package com.hivemq.pulse.status;
 
 import com.hivemq.api.model.capabilities.Capability;
+import com.hivemq.bootstrap.services.EdgeCoreFactoryService;
 import com.hivemq.edge.HiveMQCapabilityService;
 import org.jetbrains.annotations.NotNull;
 
@@ -27,10 +28,14 @@ public class PulseAgentStatusChangedListener implements StatusProvider.StatusCha
             "HiveMQ Pulse Agent Asset Management",
             "This enables HiveMQ Edge to be a HiveMQ Pulse Agent.");
     private final @NotNull HiveMQCapabilityService capabilityService;
+    private final @NotNull EdgeCoreFactoryService edgeCoreFactoryService;
     private @NotNull Status status;
 
-    public PulseAgentStatusChangedListener(final @NotNull HiveMQCapabilityService capabilityService) {
+    public PulseAgentStatusChangedListener(
+            final @NotNull HiveMQCapabilityService capabilityService,
+            final @NotNull EdgeCoreFactoryService edgeCoreFactoryService) {
         this.capabilityService = capabilityService;
+        this.edgeCoreFactoryService = edgeCoreFactoryService;
         this.status = new Status(Status.ActivationStatus.DEACTIVATED, Status.ConnectionStatus.DISCONNECTED, List.of());
     }
 
@@ -43,8 +48,10 @@ public void onStatusChanged(@NotNull final Status status) {
         this.status = status;
         if (this.status.activationStatus() == Status.ActivationStatus.ACTIVATED) {
             capabilityService.addCapability(CAPABILITY);
+            edgeCoreFactoryService.setPulseActivated(true);
         } else {
             capabilityService.removeCapability(CAPABILITY);
+            edgeCoreFactoryService.setPulseActivated(false);
         }
     }
 }

hivemq-edge/src/test/java/com/hivemq/combining/vanilla/VanillaDataCombiningTransformationServiceTest.java

@@ -20,6 +20,7 @@
 import com.hivemq.combining.model.DataCombining;
 import com.hivemq.combining.model.DataCombiningDestination;
 import com.hivemq.combining.model.DataIdentifierReference;
+import com.hivemq.common.i18n.StringTemplate;
 import com.hivemq.mqtt.handler.publish.PublishingResult;
 import com.hivemq.mqtt.message.QoS;
 import com.hivemq.mqtt.message.mqtt5.Mqtt5UserProperties;
@@ -37,6 +38,7 @@
 import org.mockito.quality.Strictness;
 
 import java.util.List;
+import java.util.Map;
 import java.util.UUID;
 
 import static org.assertj.core.api.Assertions.assertThat;
@@ -113,7 +115,8 @@ public void when1FilterMatches_thenPublishPasses() {
                 new DataIdentifierReference("topic/a", DataIdentifierReference.Type.TOPIC_FILTER))));
         assertThat(service.applyMappings(publish, dataCombining).isDone()).isFalse();
         verify(prePublishProcessorService, times(1)).publish(publishCaptor.capture(), any(), any());
-        assertThat(new String(publishCaptor.getValue().getPayload())).isEqualTo("{\"dest\":{\"a\":1}}");
+        assertThat(new String(publishCaptor.getValue().getPayload())).isEqualTo("""
+                {"dest":{"a":1}}""");
     }
 
     @Test
@@ -135,7 +138,8 @@ public void when2FiltersMatch_thenPublishPasses() {
                         new DataIdentifierReference("topic/b", DataIdentifierReference.Type.TOPIC_FILTER))));
         assertThat(service.applyMappings(publish, dataCombining).isDone()).isFalse();
         verify(prePublishProcessorService, times(1)).publish(publishCaptor.capture(), any(), any());
-        assertThat(new String(publishCaptor.getValue().getPayload())).isEqualTo("{\"dest\":{\"a\":1,\"b\":2}}");
+        assertThat(new String(publishCaptor.getValue().getPayload())).isEqualTo("""
+                {"dest":{"a":1,"b":2}}""");
     }
 
     @Test
@@ -146,7 +150,8 @@ public void when1TagMatches_thenPublishPasses() {
                 new DataIdentifierReference("TAG1", DataIdentifierReference.Type.TAG))));
         assertThat(service.applyMappings(publish, dataCombining).isDone()).isFalse();
         verify(prePublishProcessorService, times(1)).publish(publishCaptor.capture(), any(), any());
-        assertThat(new String(publishCaptor.getValue().getPayload())).isEqualTo("{\"dest\":{\"tag1\":\"TAG1\"}}");
+        assertThat(new String(publishCaptor.getValue().getPayload())).isEqualTo("""
+                {"dest":{"tag1":"TAG1"}}""");
     }
 
     @Test
@@ -160,8 +165,8 @@ public void when2TagsMatch_thenPublishPasses() {
                         new DataIdentifierReference("TAG2", DataIdentifierReference.Type.TAG))));
         assertThat(service.applyMappings(publish, dataCombining).isDone()).isFalse();
         verify(prePublishProcessorService, times(1)).publish(publishCaptor.capture(), any(), any());
-        assertThat(new String(publishCaptor.getValue().getPayload())).isEqualTo(
-                "{\"dest\":{\"tag1\":\"TAG1\",\"tag2\":\"TAG2\"}}");
+        assertThat(new String(publishCaptor.getValue().getPayload())).isEqualTo("""
+                {"dest":{"tag1":"TAG1","tag2":"TAG2"}}""");
     }
 
     @Test
@@ -173,9 +178,8 @@ public void when1AssetMatches_thenPublishPasses() {
                 new DataIdentifierReference(assetId, DataIdentifierReference.Type.PULSE_ASSET))));
         assertThat(service.applyMappings(publish, dataCombining).isDone()).isFalse();
         verify(prePublishProcessorService, times(1)).publish(publishCaptor.capture(), any(), any());
-        assertThat(new String(publishCaptor.getValue().getPayload())).isEqualTo("{\"dest\":{\"asset\":\"" +
-                assetId +
-                "\"}}");
+        assertThat(new String(publishCaptor.getValue().getPayload())).isEqualTo(StringTemplate.format("""
+                {"dest":{"asset":"${assetId}"}}""", Map.of("assetId", assetId)));
     }
 
     @Test
@@ -209,8 +213,8 @@ public void when2FiltersAnd2TagsAnd2AssetsMatch_thenPublishPasses() {
                         new DataIdentifierReference("ASSET2", DataIdentifierReference.Type.PULSE_ASSET))));
         assertThat(service.applyMappings(publish, dataCombining).isDone()).isFalse();
         verify(prePublishProcessorService, times(1)).publish(publishCaptor.capture(), any(), any());
-        assertThat(new String(publishCaptor.getValue().getPayload())).isEqualTo(
-                "{\"dest\":{\"a\":1,\"b\":2,\"tag1\":\"TAG1\",\"tag2\":\"TAG2\",\"asset1\":\"ASSET1\",\"asset2\":\"ASSET2\"}}");
+        assertThat(new String(publishCaptor.getValue().getPayload())).isEqualTo("""
+                {"dest":{"a":1,"b":2,"tag1":"TAG1","tag2":"TAG2","asset1":"ASSET1","asset2":"ASSET2"}}""");
     }
 
     @Test
@@ -232,7 +236,8 @@ public void when2FiltersOverlap_thenLast1WinsAndPublishPasses() {
                         new DataIdentifierReference("topic/b", DataIdentifierReference.Type.TOPIC_FILTER))));
         assertThat(service.applyMappings(publish, dataCombining).isDone()).isFalse();
         verify(prePublishProcessorService, times(1)).publish(publishCaptor.capture(), any(), any());
-        assertThat(new String(publishCaptor.getValue().getPayload())).isEqualTo("{\"dest\":{\"x\":2}}");
+        assertThat(new String(publishCaptor.getValue().getPayload())).isEqualTo("""
+                {"dest":{"x":2}}""");
     }
 
     @Test
@@ -246,7 +251,8 @@ public void when2TagsOverlap_thenLast1WinsAndPublishPasses() {
                         new DataIdentifierReference("TAG2", DataIdentifierReference.Type.TAG))));
         assertThat(service.applyMappings(publish, dataCombining).isDone()).isFalse();
         verify(prePublishProcessorService, times(1)).publish(publishCaptor.capture(), any(), any());
-        assertThat(new String(publishCaptor.getValue().getPayload())).isEqualTo("{\"dest\":{\"tag\":\"TAG2\"}}");
+        assertThat(new String(publishCaptor.getValue().getPayload())).isEqualTo("""
+                {"dest":{"tag":"TAG2"}}""");
     }
 
     @Test
@@ -269,6 +275,30 @@ public void whenSingleAndDoubleQuotesAreInTopic_thenPublishPasses() {
                                 DataIdentifierReference.Type.TOPIC_FILTER))));
         assertThat(service.applyMappings(publish, dataCombining).isDone()).isFalse();
         verify(prePublishProcessorService, times(1)).publish(publishCaptor.capture(), any(), any());
-        assertThat(new String(publishCaptor.getValue().getPayload())).isEqualTo("{\"dest\":{\"a\":1,\"b\":2}}");
+        assertThat(new String(publishCaptor.getValue().getPayload())).isEqualTo("""
+                {"dest":{"a":1,"b":2}}""");
+    }
+
+    @Test
+    public void whenDestinationIsFullJsonPath_thenNoDollarIsInTheResult() {
+        when(publish.getPayload()).thenReturn("""
+                {
+                  "TOPIC_FILTER:topic/a": {
+                    "a": 1
+                  },
+                  "TOPIC_FILTER:topic/b": {
+                    "b": 2
+                  }
+                }""".getBytes());
+        when(dataCombining.instructions()).thenReturn(List.of(new Instruction("$.a",
+                        "$.dest.a",
+                        new DataIdentifierReference("topic/a", DataIdentifierReference.Type.TOPIC_FILTER)),
+                new Instruction("$.b",
+                        "$.dest.b",
+                        new DataIdentifierReference("topic/b", DataIdentifierReference.Type.TOPIC_FILTER))));
+        assertThat(service.applyMappings(publish, dataCombining).isDone()).isFalse();
+        verify(prePublishProcessorService, times(1)).publish(publishCaptor.capture(), any(), any());
+        assertThat(new String(publishCaptor.getValue().getPayload())).isEqualTo("""
+                {"dest":{"a":1,"b":2}}""");
     }
 }