tag metadata in northbound mappings

Author: codepitbull

hivemq-edge/src/main/java/com/hivemq/protocols/northbound/NorthboundTagConsumer.java

@@ -27,7 +27,6 @@
 import com.hivemq.adapter.sdk.api.data.JsonPayloadCreator;
 import com.hivemq.adapter.sdk.api.events.EventService;
 import com.hivemq.adapter.sdk.api.events.model.Payload;
-import com.hivemq.adapter.sdk.api.factories.DataPointFactory;
 import com.hivemq.adapter.sdk.api.services.ProtocolAdapterMetricsService;
 import com.hivemq.edge.modules.adapters.data.DataPointImpl;
 import com.hivemq.edge.modules.adapters.impl.ProtocolAdapterPublishServiceImpl;
@@ -59,7 +58,6 @@ public class NorthboundTagConsumer implements TagConsumer{
     private final @NotNull ProtocolAdapterMetricsService protocolAdapterMetricsService;
     private final @NotNull EventService eventService;
     private final @NotNull AtomicInteger publishCount = new AtomicInteger(0);
-    private final @NotNull DataPointFactory dataPointFactory;
 
     public NorthboundTagConsumer(
             final @NotNull PollingContext pollingContext,
@@ -76,19 +74,6 @@ public NorthboundTagConsumer(
         this.protocolAdapterPublishService = protocolAdapterPublishService;
         this.protocolAdapterMetricsService = protocolAdapterMetricsService;
         this.eventService = eventService;
-        this.dataPointFactory = new DataPointFactory() {
-            @Override
-            public @NotNull DataPoint create(final @NotNull String tagName, final @NotNull Object tagValue) {
-                return new DataPointImpl(tagName, tagValue);
-            }
-
-            @Override
-            public @NotNull DataPoint createJsonDataPoint(
-                    final @NotNull String tagName,
-                    final @NotNull Object tagValue) {
-                return new DataPointImpl(tagName, tagValue, true);
-            }
-        };
     }
 
     public void accept(final @NotNull List<DataPoint> dataPoints) {
@@ -101,8 +86,6 @@ public void accept(final @NotNull List<DataPoint> dataPoints) {
         try {
             final ImmutableList.Builder<CompletableFuture<?>> publishFutures = ImmutableList.builder();
 
-            final List<byte[]> jsonPayloadsAsBytes = new ArrayList<>();
-
             final JsonPayloadCreator jsonPayloadCreatorOverride = pollingContext.getJsonPayloadCreator();
 
             final List<DataPoint> jsonDataPoints =
@@ -111,23 +94,25 @@ public void accept(final @NotNull List<DataPoint> dataPoints) {
             final var preparedJsonDataPoints = jsonDataPoints.stream().map(jsonDataPoint -> {
                 try {
                     final var jsonMap=objectMapper.readValue((String)jsonDataPoint.getTagValue(), typeRef);
-                    final var value = jsonMap.get("value");
-                    if(value!=null) {
-                        return dataPointFactory.create(jsonDataPoint.getTagName(), value);
+                    if (jsonMap.size() > 1 && jsonMap.containsKey("value")) {
+                        return new DataPointImpl(jsonDataPoint.getTagName(), jsonMap);
+                    } else if (jsonMap.containsKey("value")) {
+                        return new DataPointImpl(jsonDataPoint.getTagName(), jsonMap.get("value"), true);
                     } else {
                         throw new RuntimeException("No value entry in JSON message");
                     }
-                } catch (JsonProcessingException e) {
+                } catch (final JsonProcessingException e) {
                     throw new RuntimeException(e);
                 }
             }).toList();
 
             final var dataPointsCopied = new ArrayList<>(dataPoints);
             dataPointsCopied.removeAll(jsonDataPoints);
             dataPointsCopied.addAll(preparedJsonDataPoints);
-            jsonPayloadsAsBytes
-                    .addAll(Objects.requireNonNullElse(jsonPayloadCreatorOverride, jsonPayloadCreator)
-                    .convertToJson(dataPointsCopied, pollingContext, objectMapper));
+
+            final List<byte[]> jsonPayloadsAsBytes =
+                    new ArrayList<>(Objects.requireNonNullElse(jsonPayloadCreatorOverride, jsonPayloadCreator)
+                            .convertToJson(dataPointsCopied, pollingContext, objectMapper));
 
             for (final byte[] json : jsonPayloadsAsBytes) {
                 final ProtocolAdapterPublishBuilder publishBuilder = protocolAdapterPublishService.createPublish()

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

@@ -86,14 +86,14 @@ public class OpcUaToJsonConverter {
                 jsonObject.add("statusCode", convertStatusCode(dataValue.getStatusCode()));
             }
             if (dataValue.getSourceTime() != null) {
-                jsonObject.add("sourceTimestamp",
+                jsonObject.add("sourceTime",
                         new JsonPrimitive(DateTimeFormatter.ISO_INSTANT.format(dataValue.getSourceTime().getJavaInstant())));
             }
             if (dataValue.getSourcePicoseconds() != null) {
                 jsonObject.add("sourcePicoseconds", new JsonPrimitive(dataValue.getSourcePicoseconds().intValue()));
             }
             if (dataValue.getServerTime() != null) {
-                jsonObject.add("serverTimestamp",
+                jsonObject.add("serverTime",
                         new JsonPrimitive(DateTimeFormatter.ISO_INSTANT.format(dataValue.getServerTime().getJavaInstant())));
             }
             if (dataValue.getServerPicoseconds() != null) {