Feature/28823 persist schema in southbound (#686)

* add schema to SouthboundMapping * persist schema in config update tests * fix ProtocolAdaptersResourceImpl * fix * fix after rebase * add validation for SouthboundMappingEntity * fix missing contents in xml * update milo to latest version
hivemq · Dec 10, 2024 · c179bf0 · c179bf0
1 parent d840087
commit c179bf0
Show file tree

Hide file tree

Showing 15 changed files with 125 additions and 41 deletions.
diff --git a/gradle/libs.versions.toml b/gradle/libs.versions.toml
@@ -32,7 +32,7 @@ jsonSchemaValidator = "1.5.2"
 junit-jupiter = "5.11.2"
 junit = "4.13.2"
 logback = "1.5.11"
-milo = "0.6.14"
+milo = "0.6.15"
 mockito = "5.14.2"
 modbus = "1.2.2"
 mqtt-sn-codec = "838f51d691"

diff --git a/...q-edge/src/main/java/com/hivemq/api/model/mappings/southbound/SouthboundMappingModel.java b/...q-edge/src/main/java/com/hivemq/api/model/mappings/southbound/SouthboundMappingModel.java
@@ -61,10 +61,13 @@ public SouthboundMappingModel(
         return fieldMapping;
     }
 
-    public @NotNull SouthboundMapping toToEdgeMapping() {
+    public @NotNull SouthboundMapping toToEdgeMapping(final @NotNull String schema) {
         return new SouthboundMapping(this.tagName,
                 this.topicFilter,
-                this.fieldMapping != null ? FieldMapping.fromModel(this.fieldMapping) : null);
+                this.fieldMapping != null ?
+                        FieldMapping.fromModel(this.fieldMapping) :
+                        FieldMapping.DEFAULT_FIELD_MAPPING,
+                schema);
     }
 
     public static SouthboundMappingModel from(final @NotNull SouthboundMapping southboundMapping) {

diff --git a/hivemq-edge/src/main/java/com/hivemq/api/resources/impl/ProtocolAdaptersResourceImpl.java b/hivemq-edge/src/main/java/com/hivemq/api/resources/impl/ProtocolAdaptersResourceImpl.java
@@ -64,6 +64,8 @@
 import com.hivemq.persistence.domain.DomainTagUpdateResult;
 import com.hivemq.persistence.mappings.NorthboundMapping;
 import com.hivemq.persistence.mappings.SouthboundMapping;
+import com.hivemq.persistence.topicfilter.TopicFilter;
+import com.hivemq.persistence.topicfilter.TopicFilterPersistence;
 import com.hivemq.protocols.InternalProtocolAdapterWritingService;
 import com.hivemq.protocols.ProtocolAdapterConfig;
 import com.hivemq.protocols.ProtocolAdapterConfigConverter;
@@ -86,6 +88,7 @@
 import java.net.URLDecoder;
 import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
+import java.util.Base64;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
@@ -106,6 +109,7 @@ public class ProtocolAdaptersResourceImpl extends AbstractApi implements Protoco
     private final @NotNull ProtocolAdapterManager protocolAdapterManager;
     private final @NotNull InternalProtocolAdapterWritingService protocolAdapterWritingService;
     private final @NotNull ProtocolAdapterConfigConverter configConverter;
+    private final @NotNull TopicFilterPersistence topicFilterPersistence;
     private final @NotNull ObjectMapper objectMapper;
     private final @NotNull VersionProvider versionProvider;
     private final @NotNull CustomConfigSchemaGenerator customConfigSchemaGenerator = new CustomConfigSchemaGenerator();
@@ -118,14 +122,16 @@ public ProtocolAdaptersResourceImpl(
             final @NotNull InternalProtocolAdapterWritingService protocolAdapterWritingService,
             final @NotNull ObjectMapper objectMapper,
             final @NotNull VersionProvider versionProvider,
-            final @NotNull ProtocolAdapterConfigConverter configConverter) {
+            final @NotNull ProtocolAdapterConfigConverter configConverter,
+            final @NotNull TopicFilterPersistence topicFilterPersistence) {
         this.remoteService = remoteService;
         this.configurationService = configurationService;
         this.protocolAdapterManager = protocolAdapterManager;
         this.objectMapper = ProtocolAdapterUtils.createProtocolAdapterMapper(objectMapper);
         this.versionProvider = versionProvider;
         this.protocolAdapterWritingService = protocolAdapterWritingService;
         this.configConverter = configConverter;
+        this.topicFilterPersistence = topicFilterPersistence;
     }
 
     @Override
@@ -645,9 +651,10 @@ protected void validateAdapterSchema(
                     .map(NorthboundMappingModel::to)
                     .collect(Collectors.toList());
 
+
             final List<SouthboundMapping> southboundMappings = adapter.getSouthboundMappingModels()
                     .stream()
-                    .map(SouthboundMappingModel::toToEdgeMapping)
+                    .map(this::parseAndEnrichWithSchema)
                     .collect(Collectors.toList());
 
             protocolAdapterManager.addAdapter(new ProtocolAdapterConfig(adapterId,
@@ -672,6 +679,7 @@ protected void validateAdapterSchema(
         return Response.ok().build();
     }
 
+
     @Override
     public Response getNorthboundMappingsForAdapter(final @NotNull String adapterId) {
         return protocolAdapterManager.getAdapterById(adapterId)
@@ -758,7 +766,7 @@ public Response updateSouthboundMappingsForAdapter(
             final Set<String> requiredTags = new HashSet<>();
             final List<SouthboundMapping> converted = southboundMappingListModel.getItems().stream().map(mapping -> {
                 requiredTags.add(mapping.getTagName());
-                return mapping.toToEdgeMapping();
+                return parseAndEnrichWithSchema(mapping);
             }).collect(Collectors.toList());
             adapter.getTags().forEach(tag -> requiredTags.remove(tag.getName()));
 
@@ -778,4 +786,24 @@ public Response updateSouthboundMappingsForAdapter(
             }
         }).orElseGet(() -> ApiErrorUtils.notFound("Adapter not found"));
     }
+
+
+    private @NotNull SouthboundMapping parseAndEnrichWithSchema(final @NotNull SouthboundMappingModel model) {
+        final TopicFilter topicFilter = topicFilterPersistence.getTopicFilter(model.getTopicFilter());
+        if (topicFilter == null) {
+            throw new IllegalStateException("Southbound mapping contained a topic filter '" +
+                    model.getTopicFilter() +
+                    "', which is unknown to Edge. Southbound mapping can not be created.");
+        }
+
+        final DataUrl schemaAsDataUrl = topicFilter.getSchema();
+        if (schemaAsDataUrl == null) {
+            throw new IllegalStateException("Southbound mapping contained a topic filter '" +
+                    model.getTopicFilter() +
+                    "', which has no schema attached. Southbound mapping can not be created.");
+        }
+
+        final String schema = new String(Base64.getDecoder().decode(schemaAsDataUrl.getData()));
+        return model.toToEdgeMapping(schema);
+    }
 }
diff --git a/...q-edge/src/main/java/com/hivemq/configuration/entity/adapter/SouthboundMappingEntity.java b/...q-edge/src/main/java/com/hivemq/configuration/entity/adapter/SouthboundMappingEntity.java
@@ -17,9 +17,8 @@
 
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.hivemq.configuration.entity.adapter.fieldmapping.FieldMappingEntity;
-import org.jetbrains.annotations.NotNull;
-import org.jetbrains.annotations.Nullable;
 import com.hivemq.persistence.mappings.SouthboundMapping;
+import org.jetbrains.annotations.NotNull;
 
 import javax.xml.bind.ValidationEvent;
 import javax.xml.bind.annotation.XmlElement;
@@ -35,23 +34,29 @@ public class SouthboundMappingEntity {
     @XmlElement(name = "tagName", required = true)
     private final @NotNull String tagName;
 
-    @XmlElement(name = "fieldMapping")
-    private final @Nullable FieldMappingEntity fieldMapping;
+    @XmlElement(name = "fieldMapping", required = true)
+    private final @NotNull FieldMappingEntity fieldMapping;
+
+    @XmlElement(name = "fromNorthSchema", required = true)
+    private final @NotNull String fromNorthSchema;
 
     // no-arg constructor for JaxB
     public SouthboundMappingEntity() {
         topicFilter = "";
         tagName = "";
         fieldMapping = null;
+        fromNorthSchema = "";
     }
 
     public SouthboundMappingEntity(
             final @NotNull String tagName,
             final @NotNull String topicFilter,
-            final @Nullable FieldMappingEntity fieldMapping) {
+            final @NotNull FieldMappingEntity fieldMapping,
+            final @NotNull String fromNorthSchema) {
         this.tagName = tagName;
         this.topicFilter = topicFilter;
         this.fieldMapping = fieldMapping;
+        this.fromNorthSchema = fromNorthSchema;
     }
 
     public @NotNull String getTagName() {
@@ -69,22 +74,24 @@ public void validate(final @NotNull List<ValidationEvent> validationEvents) {
         if (tagName == null || tagName.isEmpty()) {
             validationEvents.add(new ValidationEventImpl(ValidationEvent.FATAL_ERROR, "tagName is missing", null));
         }
+        if (fromNorthSchema == null || fromNorthSchema.isEmpty()) {
+            validationEvents.add(new ValidationEventImpl(ValidationEvent.FATAL_ERROR, "fromNorthSchema is missing", null));
+        }
     }
 
 
     public @NotNull SouthboundMapping to(final @NotNull ObjectMapper mapper) {
-        return new SouthboundMapping(
-                this.getTagName(),
+        return new SouthboundMapping(this.getTagName(),
                 this.getTopicFilter(),
-                this.fieldMapping != null ? this.fieldMapping.to(mapper) : null);
+                this.fieldMapping != null ? this.fieldMapping.to(mapper) : null,
+                this.fromNorthSchema);
     }
 
     public static @NotNull SouthboundMappingEntity from(final @NotNull SouthboundMapping southboundMapping) {
-        return new SouthboundMappingEntity(
-                southboundMapping.getTagName(),
+        return new SouthboundMappingEntity(southboundMapping.getTagName(),
                 southboundMapping.getTopicFilter(),
-                FieldMappingEntity.from(southboundMapping.getFieldMapping())
-        );
+                FieldMappingEntity.from(southboundMapping.getFieldMapping()),
+                southboundMapping.getSchema());
     }
 
 }
diff --git a/hivemq-edge/src/main/java/com/hivemq/persistence/mappings/SouthboundMapping.java b/hivemq-edge/src/main/java/com/hivemq/persistence/mappings/SouthboundMapping.java
@@ -21,19 +21,22 @@
 import org.jetbrains.annotations.NotNull;
 import org.jetbrains.annotations.Nullable;
 
-public class SouthboundMapping implements WritingContext {
+public class SouthboundMapping implements InternalWritingContext {
 
     private final @NotNull String topicFilter;
     private final @NotNull String tagName;
-    private final @Nullable FieldMapping fieldMapping;
+    private final @NotNull FieldMapping fieldMapping;
+    private final @NotNull String schema;
 
     public SouthboundMapping(
             final @NotNull String tagName,
             final @NotNull String topicFilter,
-            final @Nullable FieldMapping fieldMapping) {
+            final @NotNull FieldMapping fieldMapping,
+            final @NotNull String schema) {
         this.tagName = tagName;
         this.topicFilter = topicFilter;
         this.fieldMapping = fieldMapping;
+        this.schema = schema;
     }
 
     public @NotNull String getTopicFilter() {
@@ -44,16 +47,13 @@ public SouthboundMapping(
         return tagName;
     }
 
-    public @Nullable FieldMapping getFieldMapping() {
+    public @NotNull FieldMapping getFieldMapping() {
         return fieldMapping;
     }
 
-    public static @NotNull SouthboundMapping from(
-            final @NotNull InternalWritingContext writingContext) {
-        return new SouthboundMapping(
-                writingContext.getTagName(),
-                writingContext.getTopicFilter(),
-                writingContext.getFieldMapping());
+    @Override
+    public @NotNull String getSchema() {
+        return schema;
     }
 
     @Override

diff --git a/hivemq-edge/src/main/java/com/hivemq/persistence/mappings/fieldmapping/FieldMapping.java b/hivemq-edge/src/main/java/com/hivemq/persistence/mappings/fieldmapping/FieldMapping.java
@@ -23,6 +23,9 @@
 
 public class FieldMapping {
 
+    public static @NotNull FieldMapping DEFAULT_FIELD_MAPPING =
+            new FieldMapping(List.of(new Instruction("value", "value")));
+
     private final @NotNull List<Instruction> instructions;
 
     public FieldMapping(final @NotNull List<Instruction> instructions) {

diff --git a/hivemq-edge/src/main/java/com/hivemq/protocols/InternalWritingContext.java b/hivemq-edge/src/main/java/com/hivemq/protocols/InternalWritingContext.java
@@ -26,4 +26,6 @@ public interface InternalWritingContext extends WritingContext {
    @NotNull
    FieldMapping getFieldMapping();
 
+   @NotNull
+    String getSchema();
 }
diff --git a/hivemq-edge/src/main/java/com/hivemq/protocols/InternalWritingContextImpl.java b/hivemq-edge/src/main/java/com/hivemq/protocols/InternalWritingContextImpl.java
@@ -33,6 +33,11 @@ public FieldMapping getFieldMapping() {
         return southboundMapping.getFieldMapping();
     }
 
+    @Override
+    public @NotNull String getSchema() {
+        return southboundMapping.getSchema();
+    }
+
     @Override
     public @NotNull String getTagName() {
         return southboundMapping.getTagName();
-Original file line number
+Diff line change
@@ Expand Up @@
        @NotNull
        FieldMapping getFieldMapping();
+       @NotNull
+        String getSchema();
     }