Run schema validation in separate stage (#589)

helm/antu/templates/rbac.yaml

@@ -37,7 +37,7 @@ rules:
     verbs: ["create"]
   - apiGroups: ["coordination.k8s.io"]
     resources: ["leases"]
-    resourceNames: ["antu-leaders-lockonantureportaggregationqueue", "antu-leaders-lockonanturefreshstopcacheperiodically", "antu-leaders-lockonanturefreshstopcacheatstartup", "antu-leaders-lockonantucommonfilesaggregationqueue", "antu-leaders-lockonanturefreshorganisationcache"]
+    resourceNames: ["antu-leaders-lockonantuxmlschemavalidationaggregationqueue", "antu-leaders-lockonantureportaggregationqueue", "antu-leaders-lockonanturefreshstopcacheperiodically", "antu-leaders-lockonanturefreshstopcacheatstartup", "antu-leaders-lockonantucommonfilesaggregationqueue", "antu-leaders-lockonanturefreshorganisationcache"]
     verbs: ["get", "update", "list"]
   - apiGroups: ["apps"]
     resources: [ "deployments" ]

src/main/java/no/entur/antu/Constants.java

@@ -43,6 +43,11 @@ public final class Constants {
 
   public static final String JOB_TYPE = "JOB_TYPE";
   public static final String JOB_TYPE_SPLIT = "SPLIT";
+  public static final String JOB_TYPE_VALIDATE_XML_SCHEMA =
+    "VALIDATE_XML_SCHEMA";
+  public static final String JOB_TYPE_AGGREGATE_XML_SCHEMA_VALIDATION =
+    "AGGREGATE_XML_SCHEMA_VALIDATION";
+
   public static final String JOB_TYPE_VALIDATE = "VALIDATE";
   public static final String JOB_TYPE_VALIDATE_DATASET = "VALIDATE_DATASET";
   public static final String JOB_TYPE_COMPLETE_VALIDATION =
@@ -108,5 +113,13 @@ public final class Constants {
   public static final String PEN_XMLNS = "PEN";
   public static final String PEN_XMLNSURL = "http://www.rutebanken.org/ns/pen";
 
+  public static final String PROP_DATASET_NETEX_FILE_NAMES_STRING =
+    "EnturDatasetNetexFileNamesString";
+  public static final String PROP_DATASET_NETEX_FILE_NAMES_SET =
+    "EnturDatasetNetexFileNamesSet";
+  public static final String PROP_STOP_WATCH = "PROP_STOP_WATCH";
+  public static final String PROP_NETEX_VALIDATION_CALLBACK =
+    "PROP_NETEX_VALIDATION_CALLBACK";
+
   private Constants() {}
 }

src/main/java/no/entur/antu/config/ValidatorConfig.java

@@ -21,7 +21,7 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
-import no.entur.antu.validation.NetexValidationProfile;
+import no.entur.antu.validation.NetexValidationWorkflow;
 import no.entur.antu.validation.validator.id.NetexIdValidator;
 import no.entur.antu.validation.validator.id.ReferenceToNsrValidator;
 import no.entur.antu.validation.validator.id.TrainElementRegistryIdValidator;
@@ -160,7 +160,7 @@ public NetexIdUniquenessValidator netexIdUniquenessValidator(
   }
 
   @Bean
-  public NetexValidationProfile netexValidationProfile(
+  public NetexValidationWorkflow netexValidationWorkflow(
     @Qualifier(
       "timetableDataValidatorsRunner"
     ) NetexValidatorsRunner timetableDataValidatorsRunner,
@@ -186,7 +186,7 @@ public NetexValidationProfile netexValidationProfile(
       "${antu.netex.validation.validators.skip:false}"
     ) boolean skipNetexValidators
   ) {
-    return new NetexValidationProfile(
+    return new NetexValidationWorkflow(
       Map.of(
         TIMETABLE,
         timetableDataValidatorsRunner,

src/main/java/no/entur/antu/routes/validation/AggregateValidationReportsAggregationStrategy.java

@@ -0,0 +1,85 @@
+package no.entur.antu.routes.validation;
+
+import static no.entur.antu.Constants.*;
+
+import java.time.LocalDateTime;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+import org.apache.camel.Exchange;
+import org.apache.camel.processor.aggregate.GroupedMessageAggregationStrategy;
+import org.entur.netex.validation.validator.ValidationReport;
+import org.entur.netex.validation.validator.ValidationReportEntry;
+
+/**
+ * Strategy that aggregates the validation reports produced for individual files into a single validation report.
+ */
+class AggregateValidationReportsAggregationStrategy
+  extends GroupedMessageAggregationStrategy {
+
+  @Override
+  public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
+    if (
+      oldExchange == null ||
+      oldExchange.getIn().getBody(ValidationReport.class) == null
+    ) {
+      return newExchange;
+    }
+
+    ValidationReport oldValidationReport = oldExchange
+      .getIn()
+      .getBody(ValidationReport.class);
+    ValidationReport newValidationReport = newExchange
+      .getIn()
+      .getBody(ValidationReport.class);
+
+    List<ValidationReportEntry> validationReportEntries = Stream
+      .concat(
+        oldValidationReport.getValidationReportEntries().stream(),
+        newValidationReport.getValidationReportEntries().stream()
+      )
+      .toList();
+
+    Map<String, Long> numberOfValidationEntriesPerRule = Stream
+      .concat(
+        oldValidationReport
+          .getNumberOfValidationEntriesPerRule()
+          .entrySet()
+          .stream(),
+        newValidationReport
+          .getNumberOfValidationEntriesPerRule()
+          .entrySet()
+          .stream()
+      )
+      .collect(
+        Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue, Long::sum)
+      );
+
+    oldExchange
+      .getIn()
+      .setBody(
+        new ValidationReport(
+          oldExchange.getIn().getHeader(DATASET_CODESPACE, String.class),
+          oldExchange
+            .getIn()
+            .getHeader(VALIDATION_REPORT_ID_HEADER, String.class),
+          validationReportEntries,
+          numberOfValidationEntriesPerRule
+        )
+      );
+
+    LocalDateTime reportCreationDate = oldExchange
+      .getIn()
+      .getHeader(REPORT_CREATION_DATE, LocalDateTime.class);
+    if (
+      reportCreationDate == null ||
+      reportCreationDate.isAfter(oldValidationReport.getCreationDate())
+    ) {
+      oldExchange
+        .getIn()
+        .setHeader(REPORT_CREATION_DATE, oldValidationReport.getCreationDate());
+    }
+    return oldExchange;
+  }
+}