diff --git a/hapi-fhir-base/src/main/java/ca/uhn/fhir/validation/FhirValidator.java b/hapi-fhir-base/src/main/java/ca/uhn/fhir/validation/FhirValidator.java index 842b8fd8ac2..2d2706f32b0 100644 --- a/hapi-fhir-base/src/main/java/ca/uhn/fhir/validation/FhirValidator.java +++ b/hapi-fhir-base/src/main/java/ca/uhn/fhir/validation/FhirValidator.java @@ -27,9 +27,12 @@ import ca.uhn.fhir.interceptor.api.Pointcut; import ca.uhn.fhir.rest.server.exceptions.InternalErrorException; import ca.uhn.fhir.util.BundleUtil; import ca.uhn.fhir.validation.schematron.SchematronProvider; +import org.apache.commons.lang3.SerializationUtils; import org.apache.commons.lang3.Validate; +import org.hl7.fhir.instance.model.api.IAnyResource; import org.hl7.fhir.instance.model.api.IBaseBundle; import org.hl7.fhir.instance.model.api.IBaseResource; +import org.hl7.fhir.instance.model.api.IDomainResource; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -38,6 +41,7 @@ import java.util.List; import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorService; import java.util.concurrent.Future; +import java.util.function.Consumer; import java.util.stream.Collectors; import java.util.stream.IntStream; @@ -250,6 +254,14 @@ public class FhirValidator { } } + /*if (myConcurrentBundleValidation) { + if (myExecutorService != null) { + return validateResourceConcurrently(theResource, theOptions); + } else { + ourLog.error("Concurrent Bundle Validation is enabled but ExecutorService is null. Reverting to serial validation."); + } + }*/ + return validateResource(theResource, theOptions); } @@ -311,6 +323,58 @@ public class FhirValidator { return theValidationResult; } + private ValidationResult validateResourceConcurrently(IBaseResource theResource, ValidationOptions theOptions) { + List validationTasks = new ArrayList<>(); + populateConcurrentValidationTasks(theResource, theOptions, "", validationTasks); + List messages = new ArrayList<>(); + for (ConcurrentValidationTask validationTask : validationTasks) { + try { + ValidationResult result = validationTask.getFuture().get(); + messages.addAll(result.getMessages().stream() + .map(message -> { + String currentPath = message.getLocationString().substring(message.getLocationString().indexOf('.')); + message.setLocationString(validationTask.getResourcePathPrefix() + currentPath); + return message; + }).collect(Collectors.toList())); + } catch (InterruptedException | ExecutionException exp) { + throw new InternalErrorException(exp); + } + } + return new ValidationResult(myContext, messages); + } + + private void populateConcurrentValidationTasks(IBaseResource theResource, ValidationOptions theOptions, String resourcePathPrefix, List theValidationTasks) { + if (theResource instanceof IBaseBundle) { + List bundleEntries = BundleUtil.toListOfResources(myContext, (IBaseBundle) theResource); + for (int i = 0; i < bundleEntries.size(); i++) { + IBaseResource entry = bundleEntries.get(i); + String bundleEntryPath = String.format("Bundle.entry[%d].resource.ofType(%s)", i, entry.fhirType()); + populateConcurrentValidationTasks(entry, theOptions, bundleEntryPath, theValidationTasks); + } + } else if (theResource instanceof IDomainResource) { + IDomainResource domainResource = (IDomainResource) theResource; + if (domainResource.getContained().size() > 0) { + // validate contained resources + List containedResources = domainResource.getContained(); + for (int i = 0; i < containedResources.size(); i++) { + IBaseResource containedResource = containedResources.get(i); + String containedResourcePath = String.format("%s.contained[%d].ofType(%s)", resourcePathPrefix, i, containedResource.fhirType()); + populateConcurrentValidationTasks(containedResource, theOptions, containedResourcePath, theValidationTasks); + } + } else { + theValidationTasks.add(new ConcurrentValidationTask( + resourcePathPrefix, + myExecutorService.submit(() -> validateResource(domainResource, theOptions)) + )); + } + } else { + theValidationTasks.add(new ConcurrentValidationTask( + resourcePathPrefix, + myExecutorService.submit(() -> validateResource(theResource, theOptions)) + )); + } + } + /** * Optionally supplies an interceptor broadcaster that will be used to invoke validation related Pointcut events * @@ -342,4 +406,22 @@ public class FhirValidator { myConcurrentBundleValidation = theConcurrentBundleValidation; return this; } + + private static class ConcurrentValidationTask { + private final String myResourcePathPrefix; + private final Future myFuture; + + private ConcurrentValidationTask(String theResourcePathPrefix, Future theFuture) { + myResourcePathPrefix = theResourcePathPrefix; + myFuture = theFuture; + } + + public String getResourcePathPrefix() { + return myResourcePathPrefix; + } + + public Future getFuture() { + return myFuture; + } + } }