0.8.3

pom.xml

@@ -8,7 +8,7 @@
 
   <groupId>de.difuture.uds</groupId>
   <artifactId>odm2fhir</artifactId>
-  <version>0.8.2</version>
+  <version>0.8.3</version>
 
   <name>ODM2FHIR</name>
   <description>Mapper for GECCO based study/patient data in CDISC ODM to HL7 FHIR</description>
@@ -27,15 +27,17 @@
 
     <hapi-fhir.version>6.0.4</hapi-fhir.version>
     <jackson.version>2.13.3</jackson.version>
+    <kafka-fhir-serializer.version>1.0.2</kafka-fhir-serializer.version>
     <lombok.version>1.18.24</lombok.version>
     <spring-boot.version>${project.parent.version}</spring-boot.version>
+    <spring-kafka.version>2.9.0</spring-kafka.version>
     <spring-retry.version>1.3.3</spring-retry.version>
     <woodstox.version>6.3.0</woodstox.version>
 
     <maven-antrun-plugin.version>3.1.0</maven-antrun-plugin.version>
     <maven-compiler-plugin.version>3.10.1</maven-compiler-plugin.version>
-    <maven-project-info-reports-plugin.version>3.3.0</maven-project-info-reports-plugin.version>
-    <maven-site-plugin.version>3.12.0</maven-site-plugin.version>
+    <maven-project-info-reports-plugin.version>3.4.0</maven-project-info-reports-plugin.version>
+    <maven-site-plugin.version>3.12.1</maven-site-plugin.version>
   </properties>
 
   <dependencies>
@@ -104,6 +106,16 @@
       <artifactId>hapi-fhir-validation-resources-r4</artifactId>
       <version>${hapi-fhir.version}</version>
     </dependency>
+    <dependency>
+      <groupId>org.springframework.kafka</groupId>
+      <artifactId>spring-kafka</artifactId>
+      <version>${spring-kafka.version}</version>
+    </dependency>
+    <dependency>
+      <groupId>org.miracum</groupId>
+      <artifactId>kafka-fhir-serializer</artifactId>
+      <version>${kafka-fhir-serializer.version}</version>
+    </dependency>
   </dependencies>
 
   <build>

src/main/java/de/difuture/uds/odm2fhir/fhir/mapper/anamnesis/item/ImmunizationStatus.java

@@ -77,8 +77,8 @@ private Immunization createImmunization(ItemData immunizationCoding, ItemData da
         case "410594000": //NO
           immunization.setStatus(NOTDONE);
           break;
-        case "385432009": //Answer = Sonstige/Other
-          diseaseCodeableCoding.addCoding(coding.setDisplay("Not applicable (qualifier value)"));
+        case "74964007": //Answer = Sonstige/Other
+          diseaseCodeableCoding.addCoding(coding.setDisplay("Other (qualifier value)"));
           if (!textValue.isEmpty()) {
             diseaseCodeableCoding.setText(textValue.getValue());
           }
@@ -112,8 +112,8 @@ private Immunization createImmunization(ItemData immunizationCoding, ItemData da
                                                    "No known immunizations"));
     }
 
-    if (ENVIRONMENT.getProperty("fhir.notapplicables.removed", Boolean.class, true) &&
-        diseaseCodeableCoding.getCoding().stream().map(Coding::getCode).anyMatch("385432009"::equals)) {
+    if (ENVIRONMENT.getProperty("fhir.others.removed", Boolean.class, true) &&
+        diseaseCodeableCoding.getCoding().stream().map(Coding::getCode).anyMatch("74964007"::equals)) {
       diseaseCodeableCoding.setCoding(null).setText(null);
     }
 

src/main/java/de/difuture/uds/odm2fhir/fhir/mapper/medication/item/Anticoagulation.java

@@ -84,8 +84,8 @@ private MedicationStatement createMedicationStatement(ItemData generalCoding, It
         case "261665006": //Answer = Unknown
           medicationStatement.setStatus(MedicationStatementStatus.UNKNOWN);
           break;
-        case "385432009": //Answer = Sonstige/Other
-          medicationCodeableConcept.addCoding(coding.setDisplay("Not applicable (qualifier value)"));
+        case "74964007": //Answer = Sonstige/Other
+          medicationCodeableConcept.addCoding(coding.setDisplay("Other (qualifier value)"));
           if (endsWith(accurateCodingOrText.getItemOID(), "_textfeld") && !accurateCodingOrText.isEmpty()) {
             medicationCodeableConcept.setText(accurateCodingOrText.getValue());
           }
@@ -105,8 +105,8 @@ private MedicationStatement createMedicationStatement(ItemData generalCoding, It
       }
     }
 
-    if (ENVIRONMENT.getProperty("fhir.notapplicables.removed", Boolean.class, true) &&
-        medicationCodeableConcept.getCoding().stream().map(Coding::getCode).anyMatch("385432009"::equals)) {
+    if (ENVIRONMENT.getProperty("fhir.others.removed", Boolean.class, true) &&
+        medicationCodeableConcept.getCoding().stream().map(Coding::getCode).anyMatch("74964007"::equals)) {
       medicationCodeableConcept.setCoding(null).setText(null);
     }
 

src/main/java/de/difuture/uds/odm2fhir/fhir/mapper/medication/item/COVID19Therapy.java

@@ -77,8 +77,8 @@ private MedicationStatement createMedicationStatement(ItemData specificCoding, I
         case "261665006":
           medicationStatement.setStatus(MedicationStatementStatus.UNKNOWN);
           break;
-        case "385432009": //Answer = Sonstige/Other
-          medicationCodeableConcept.addCoding(coding.setDisplay("Not applicable (qualifier value)"));
+        case "74964007": //Answer = Sonstige/Other
+          medicationCodeableConcept.addCoding(coding.setDisplay("Other (qualifier value)"));
           if (endsWith(accurateCodingOrText.getItemOID(), "_textfeld") && !accurateCodingOrText.isEmpty()) {
             medicationCodeableConcept.setText(accurateCodingOrText.getValue());
           }
@@ -98,8 +98,8 @@ private MedicationStatement createMedicationStatement(ItemData specificCoding, I
       }
     }
 
-    if (ENVIRONMENT.getProperty("fhir.notapplicables.removed", Boolean.class, true) &&
-        medicationCodeableConcept.getCoding().stream().map(Coding::getCode).anyMatch("385432009"::equals)) {
+    if (ENVIRONMENT.getProperty("fhir.others.removed", Boolean.class, true) &&
+        medicationCodeableConcept.getCoding().stream().map(Coding::getCode).anyMatch("74964007"::equals)) {
       medicationCodeableConcept.setCoding(null).setText(null);
     }
 

src/main/java/de/difuture/uds/odm2fhir/fhir/mapper/symptoms/item/SymptomConditions.java

@@ -78,8 +78,8 @@ private Stream<DomainResource> createConditions(FormData formData, ItemData gene
         case "261665006": //UNKNOWN
           condition.addModifierExtension(DATA_PRESENCE_UNKNOWN);
           break;
-        case "385432009": //Answer = Sonstige/Other
-          codeCodeableConcept.addCoding(coding.setDisplay("Not applicable (qualifier value)"));
+        case "74964007": //Answer = Sonstige/Other
+          codeCodeableConcept.addCoding(coding.setDisplay("Other (qualifier value)"));
           if (!formData.getItemData("symptome_andere_symptome_textfeld").isEmpty()) {
             codeCodeableConcept.setText(formData.getItemData("symptome_andere_symptome_textfeld").getValue());
           }
@@ -108,8 +108,8 @@ private Stream<DomainResource> createConditions(FormData formData, ItemData gene
                                 .setIdentifier(List.of(identifier.copy().setValue(identifier.getValue() + "_geruch"))));
     }
 
-    if (ENVIRONMENT.getProperty("fhir.notapplicables.removed", Boolean.class, true) &&
-        codeCodeableConcept.getCoding().stream().map(Coding::getCode).anyMatch("385432009"::equals)) {
+    if (ENVIRONMENT.getProperty("fhir.others.removed", Boolean.class, true) &&
+        codeCodeableConcept.getCoding().stream().map(Coding::getCode).anyMatch("74964007"::equals)) {
       codeCodeableConcept.setCoding(null).setText(null);
     }
 

src/main/java/de/difuture/uds/odm2fhir/fhir/writer/FHIRBundler.java

@@ -64,6 +64,7 @@ public class FHIRBundler {
   @Autowired(required = false)
   private ResourceValidator resourceValidator;
 
+  @SuppressWarnings("unchecked")
   private void removeCodingDisplays(Object object) {
     if (codingdisplaysRemoved) {
       var children = findMethod(object.getClass(), "children");

src/main/java/de/difuture/uds/odm2fhir/fhir/writer/FileFHIRBundleWriter.java

@@ -37,7 +37,7 @@
 import static java.nio.file.Files.createDirectories;
 import static java.nio.file.Files.writeString;
 
-@ConditionalOnExpression("'${fhir.server.url:}'.empty")
+@ConditionalOnExpression("'${fhir.server.url:}'.empty and '${kafka.broker.url:}'.empty")
 @Service
 @Slf4j
 public class FileFHIRBundleWriter implements FHIRBundleWriter {
@@ -59,8 +59,8 @@ public void write(Bundle bundle) throws IOException {
     RESOURCES_NUMBER.addAndGet(bundle.getEntry().size());
 
     var json = JSON_PARSER.encodeResourceToString(bundle);
-    var name = ((Patient) bundle.getEntryFirstRep().getResource()).getIdentifierFirstRep().getValue();
-    writeString(folderPath.resolve(name + ".json"), json);
+    var patientIdentifier = ((Patient) bundle.getEntryFirstRep().getResource()).getIdentifierFirstRep().getValue();
+    writeString(folderPath.resolve(patientIdentifier + ".json"), json);
   }
 
 }

src/main/java/de/difuture/uds/odm2fhir/fhir/writer/KafkaFHIRBundleWriter.java

@@ -0,0 +1,48 @@
+package de.difuture.uds.odm2fhir.fhir.writer;
+
+/*
+ * Copyright (C) 2021 DIFUTURE (https://difuture.de)
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated
+ * documentation files (the "Software"), to deal in the Software without restriction, including without limitation
+ * the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, copies
+ * of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following
+ * conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all copies or substantial portions of the
+ * Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE
+ * WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
+ * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR
+ * OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+ */
+
+import lombok.extern.slf4j.Slf4j;
+
+import org.hl7.fhir.r4.model.Bundle;
+
+import org.hl7.fhir.r4.model.Patient;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.stereotype.Service;
+
+@ConditionalOnExpression("'${fhir.server.url:}'.empty and !'${kafka.broker.url:}'.empty")
+@Service
+@Slf4j
+public class KafkaFHIRBundleWriter implements FHIRBundleWriter {
+
+  @Value("${spring.kafka.template.default-topic:}")
+  private String topicName;
+
+  @Autowired
+  private KafkaTemplate<String, Bundle> kafkaTemplate;
+
+  public void write(Bundle bundle) {
+    var patientIdentifier = ((Patient) bundle.getEntryFirstRep().getResource()).getIdentifierFirstRep().getValue();
+    kafkaTemplate.send(topicName, patientIdentifier, bundle);
+  }
+
+}

src/main/resources/application.yml

@@ -66,4 +66,22 @@ fhir:
 #      medicationstatement: https://ein.fiktives.krankenhaus/fhir/NamingSystem/medicationstatementId
 #      observation: https://ein.fiktives.krankenhaus/fhir/NamingSystem/observationId
 #      patient: https://ein.fiktives.krankenhaus/fhir/NamingSystem/patientId
-#      procedure: https://ein.fiktives.krankenhaus/fhir/NamingSystem/procedureId
+#      procedure: https://ein.fiktives.krankenhaus/fhir/NamingSystem/procedureId
+
+#spring:
+#  kafka:
+#    bootstrap-servers: localhost:9092
+#    producer:
+#      key-serializer: org.apache.kafka.common.serialization.StringSerializer
+#      value-serializer: org.miracum.kafka.serializers.KafkaFhirSerializer
+#      client-id: odm2fhir
+#    security.protocol: PLAINTEXT
+#    ssl:
+#      trust-store-type: PKCS12
+#      trust-store-location: file:///opt/kafka-certs/ca.p12
+#      trust-store-password: change-password
+#      key-store-type: PKCS12
+#      key-store-location: file:///opt/kafka-certs/user.p12
+#      key-store-password: change-password
+#    template:
+#      default-topic: fhir.odm-gecco