Merge pull request #95 from hapifhir/ks-20200427-empi1

Changes for empi (as well as subscriptions and partitions)

README.md

@@ -158,6 +158,10 @@ The server may be configured with subscription support by enabling properties in
 
 - `subscription.websocket.enabled` - Enables websocket subscriptions. With this enabled, your server will accept incoming websocket connections on the following URL (this example uses the default context path and port, you may need to tweak depending on your deployment environment): [ws://localhost:8080/hapi-fhir-jpaserver/websocket](ws://localhost:8080/hapi-fhir-jpaserver/websocket)
 
+## Enabling EMPI
+
+Set `empi.enabled=true` in the [hapi.properties](https://github.com/hapifhir/hapi-fhir-jpaserver-starter/blob/master/src/main/resources/hapi.properties) file to enable EMPI on this server.  The EMPI matching rules are configured in [empi-rules.json](https://github.com/hapifhir/hapi-fhir-jpaserver-starter/blob/master/src/main/resources/empi-rules.json).  The rules in this example file should be replaced with actual matching rules appropriate to your data.
+
 ## Using Elasticsearch
 
 By default, the server will use embedded lucene indexes for terminology and fulltext indexing purposes. You can switch this to using lucene by editing the properties in [hapi.properties](https://github.com/hapifhir/hapi-fhir-jpaserver-starter/blob/master/src/main/resources/hapi.properties)

pom.xml

@@ -11,7 +11,7 @@
     <parent>
         <groupId>ca.uhn.hapi.fhir</groupId>
         <artifactId>hapi-fhir</artifactId>
-        <version>5.0.0</version>
+        <version>5.1.0-SNAPSHOT</version>
     </parent>
 
     <artifactId>hapi-fhir-jpaserver-starter</artifactId>
@@ -84,7 +84,12 @@
                 </exclusion>
             </exclusions>
         </dependency>
-
+        <!-- This dependency includes the JPA EMPI Server -->
+        <dependency>
+            <groupId>ca.uhn.hapi.fhir</groupId>
+            <artifactId>hapi-fhir-jpaserver-empi</artifactId>
+            <version>${project.version}</version>
+        </dependency>
         <!-- This dependency is used for the "FHIR Tester" web app overlay -->
         <dependency>
             <groupId>ca.uhn.hapi.fhir</groupId>
@@ -249,7 +254,12 @@
             <scope>test</scope>
         </dependency>
 
-
+        <dependency>
+            <groupId>org.awaitility</groupId>
+            <artifactId>awaitility</artifactId>
+            <version>4.0.0-rc1</version>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 
     <packaging>war</packaging>

src/main/java/ca/uhn/fhir/jpa/starter/ApplicationContext.java

@@ -1,6 +1,8 @@
 package ca.uhn.fhir.jpa.starter;
 
 import ca.uhn.fhir.context.FhirVersionEnum;
+import ca.uhn.fhir.jpa.empi.config.EmpiConsumerConfig;
+import ca.uhn.fhir.jpa.empi.config.EmpiSubmitterConfig;
 import ca.uhn.fhir.jpa.subscription.channel.config.SubscriptionChannelConfig;
 import ca.uhn.fhir.jpa.subscription.match.config.SubscriptionProcessorConfig;
 import ca.uhn.fhir.jpa.subscription.match.config.WebsocketDispatcherConfig;
@@ -35,6 +37,12 @@ public class ApplicationContext extends AnnotationConfigWebApplicationContext {
       register(SubscriptionChannelConfig.class);
     }
 
+    if (HapiProperties.getEmpiEnabled()) {
+      register(EmpiSubmitterConfig.class);
+      register(EmpiConsumerConfig.class);
+      register(EmpiConfig.class);
+    }
+
   }
 
 }

src/main/java/ca/uhn/fhir/jpa/starter/EmpiConfig.java

@@ -0,0 +1,23 @@
+package ca.uhn.fhir.jpa.starter;
+
+import ca.uhn.fhir.empi.api.IEmpiSettings;
+import ca.uhn.fhir.empi.rules.config.EmpiSettings;
+import com.google.common.base.Charsets;
+import org.apache.commons.io.IOUtils;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.core.io.DefaultResourceLoader;
+import org.springframework.core.io.Resource;
+
+import java.io.IOException;
+
+@Configuration
+public class EmpiConfig {
+  @Bean
+  IEmpiSettings empiSettings() throws IOException {
+    DefaultResourceLoader resourceLoader = new DefaultResourceLoader();
+    Resource resource = resourceLoader.getResource("empi-rules.json");
+    String json = IOUtils.toString(resource.getInputStream(), Charsets.UTF_8);
+    return new EmpiSettings().setEnabled(HapiProperties.getEmpiEnabled()).setScriptText(json);
+  }
+}

src/main/java/ca/uhn/fhir/jpa/starter/FhirServerConfigCommon.java

@@ -10,7 +10,6 @@ import ca.uhn.fhir.jpa.subscription.match.deliver.email.IEmailSender;
 import ca.uhn.fhir.jpa.subscription.match.deliver.email.JavaMailEmailSender;
 import org.apache.commons.dbcp2.BasicDataSource;
 import org.hl7.fhir.dstu2.model.Subscription;
-import org.springframework.beans.factory.BeanFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
@@ -25,7 +24,7 @@ import java.sql.Driver;
  * This is the primary configuration file for the example server
  */
 @Configuration
-@EnableTransactionManagement()
+@EnableTransactionManagement
 public class FhirServerConfigCommon {
 
   private static final org.slf4j.Logger ourLog = org.slf4j.LoggerFactory.getLogger(FhirServerConfigCommon.class);
@@ -198,11 +197,10 @@ public class FhirServerConfigCommon {
       retVal.setSmtpServerPort(this.emailPort);
       retVal.setSmtpServerUsername(this.emailUsername);
       retVal.setSmtpServerPassword(this.emailPassword);
-      // TODO KHS add these when HAPI 4.2.0 is released
+      retVal.setAuth(this.emailAuth);
-//      retVal.setAuth(this.emailAuth);
+      retVal.setStartTlsEnable(this.emailStartTlsEnable);
-//      retVal.setStartTlsEnable(this.emailStartTlsEnable);
+      retVal.setStartTlsRequired(this.emailStartTlsRequired);
-//      retVal.setStartTlsRequired(this.emailStartTlsRequired);
+      retVal.setQuitWait(this.emailQuitWait);
-//      retVal.setQuitWait(this.emailQuitWait);
 
       SubscriptionDeliveryHandlerFactory subscriptionDeliveryHandlerFactory = myAppCtx.getBean(SubscriptionDeliveryHandlerFactory.class);
       Validate.notNull(subscriptionDeliveryHandlerFactory, "No subscription delivery handler");

src/main/java/ca/uhn/fhir/jpa/starter/HapiProperties.java

@@ -57,6 +57,11 @@ public class HapiProperties {
   static final String SUBSCRIPTION_EMAIL_ENABLED = "subscription.email.enabled";
   static final String SUBSCRIPTION_RESTHOOK_ENABLED = "subscription.resthook.enabled";
   static final String SUBSCRIPTION_WEBSOCKET_ENABLED = "subscription.websocket.enabled";
+  static final String EMPI_ENABLED = "empi.enabled";
+  static final String PARTITIONING_ENABLED = "partitioning.enabled";
+  static final String PARTITIONING_CROSS_PARTITION_REFERENCE_MODE = "partitioning.cross_partition_reference_mode";
+  private static final String PARTITIONING_INCLUDE_PARTITION_IN_SEARCH_HASHES = "partitioning.partitioning_include_in_search_hashes";
+
   static final String ALLOWED_BUNDLE_TYPES = "allowed_bundle_types";
   static final String TEST_PORT = "test.port";
   static final String TESTER_CONFIG_REFUSE_TO_FETCH_THIRD_PARTY_URLS = "tester.config.refuse_to_fetch_third_party_urls";
@@ -373,6 +378,22 @@ public class HapiProperties {
     return HapiProperties.getBooleanProperty(SUBSCRIPTION_WEBSOCKET_ENABLED, false);
   }
 
+  public static Boolean getEmpiEnabled() {
+    return HapiProperties.getBooleanProperty(EMPI_ENABLED, false);
+  }
+
+  public static Boolean getPartitioningEnabled() {
+    return HapiProperties.getBooleanProperty(PARTITIONING_ENABLED, false);
+  }
+
+  public static String getPartitioningCrossPartitionReferenceMode() {
+    return HapiProperties.getProperty(PARTITIONING_CROSS_PARTITION_REFERENCE_MODE, "NOT_ALLOWED");
+  }
+
+  public static Boolean getIncludePartitionInSearchHashes() {
+    return HapiProperties.getBooleanProperty(PARTITIONING_INCLUDE_PARTITION_IN_SEARCH_HASHES, true);
+  }
+
   public static Boolean getAllowContainsSearches() {
     return HapiProperties.getBooleanProperty(ALLOW_CONTAINS_SEARCHES, true);
   }

src/main/java/ca/uhn/fhir/jpa/starter/JpaRestfulServer.java

@@ -7,10 +7,11 @@ import ca.uhn.fhir.interceptor.api.IInterceptorService;
 import ca.uhn.fhir.jpa.api.config.DaoConfig;
 import ca.uhn.fhir.jpa.api.dao.DaoRegistry;
 import ca.uhn.fhir.jpa.api.dao.IFhirSystemDao;
-import ca.uhn.fhir.jpa.api.rp.ResourceProviderFactory;
 import ca.uhn.fhir.jpa.binstore.BinaryStorageInterceptor;
 import ca.uhn.fhir.jpa.bulk.BulkDataExportProvider;
 import ca.uhn.fhir.jpa.interceptor.CascadingDeleteInterceptor;
+import ca.uhn.fhir.jpa.model.config.PartitionSettings;
+import ca.uhn.fhir.jpa.partition.PartitionManagementProvider;
 import ca.uhn.fhir.jpa.provider.GraphQLProvider;
 import ca.uhn.fhir.jpa.provider.JpaConformanceProviderDstu2;
 import ca.uhn.fhir.jpa.provider.JpaSystemProviderDstu2;
@@ -35,6 +36,7 @@ import ca.uhn.fhir.rest.server.interceptor.LoggingInterceptor;
 import ca.uhn.fhir.rest.server.interceptor.RequestValidatingInterceptor;
 import ca.uhn.fhir.rest.server.interceptor.ResponseHighlighterInterceptor;
 import ca.uhn.fhir.rest.server.interceptor.ResponseValidatingInterceptor;
+import ca.uhn.fhir.rest.server.provider.ResourceProviderFactory;
 import ca.uhn.fhir.validation.IValidatorModule;
 import ca.uhn.fhir.validation.ResultSeverityEnum;
 import org.hl7.fhir.dstu3.model.Bundle;
@@ -318,6 +320,13 @@ public class JpaRestfulServer extends RestfulServer {
       registerProvider(appCtx.getBean(BulkDataExportProvider.class));
     }
 
+    if (HapiProperties.getPartitioningEnabled()) {
+      PartitionSettings partitionSettings = appCtx.getBean(PartitionSettings.class);
+      partitionSettings.setPartitioningEnabled(true);
+      PartitionSettings.CrossPartitionReferenceMode mode = PartitionSettings.CrossPartitionReferenceMode.valueOf(HapiProperties.getPartitioningCrossPartitionReferenceMode());
+      partitionSettings.setAllowReferencesAcrossPartitions(mode);
+      partitionSettings.setIncludePartitionInSearchHashes(HapiProperties.getIncludePartitionInSearchHashes());
+      registerProvider(appCtx.getBean(PartitionManagementProvider.class));
+  }
   }
-
 }

src/main/resources/empi-rules.json

@@ -0,0 +1,35 @@
+{
+	"candidateSearchParams" : [ {
+		"resourceType" : "Patient",
+		"searchParam" : "birthdate"
+	}, {
+		"resourceType" : "*",
+		"searchParam" : "identifier"
+	},{
+		"resourceType" : "Patient",
+		"searchParam" : "general-practitioner"
+	} ],
+	"candidateFilterSearchParams" : [ {
+		"resourceType" : "*",
+		"searchParam" : "active",
+		"fixedValue" : "true"
+	} ],
+	"matchFields" : [ {
+		"name" : "given-name",
+		"resourceType" : "*",
+		"resourcePath" : "name.given",
+		"metric" : "COSINE",
+		"matchThreshold" : 0.8
+	}, {
+		"name" : "last-name",
+		"resourceType" : "*",
+		"resourcePath" : "name.family",
+		"metric" : "JARO_WINKLER",
+		"matchThreshold" : 0.8
+	}],
+	"matchResultMap" : {
+		"given-name" : "POSSIBLE_MATCH",
+		"given-name,last-name" : "MATCH"
+	},
+	"eidSystem": "http://company.io/fhir/NamingSystem/custom-eid-system"
+}

src/main/resources/hapi.properties

@@ -149,3 +149,11 @@ email.password=
 
 # Enable Websocket Subscription Channel
 subscription.websocket.enabled=false
+
+# EMPI
+empi.enabled=false
+
+# Partitioning
+partitioning.enabled=false
+partitioning.cross_partition_reference_mode=NOT_ALLOWED
+partitioning.partitioning_include_in_search_hashes=true

src/main/resources/logback.xml

@@ -9,6 +9,25 @@
 		</encoder>
 	</appender>
 
+	<appender name="EMPI_TROUBLESHOOTING" class="ch.qos.logback.core.rolling.RollingFileAppender">
+		<filter class="ch.qos.logback.classic.filter.ThresholdFilter"><level>DEBUG</level></filter>
+		<file>${smile.basedir}/log/empi-troubleshooting.log</file>
+		<rollingPolicy class="ch.qos.logback.core.rolling.FixedWindowRollingPolicy">
+			<fileNamePattern>${smile.basedir}/log/empi-troubleshooting.log.%i.gz</fileNamePattern>
+			<minIndex>1</minIndex>
+			<maxIndex>9</maxIndex>
+		</rollingPolicy>
+		<triggeringPolicy class="ch.qos.logback.core.rolling.SizeBasedTriggeringPolicy">
+			<maxFileSize>5MB</maxFileSize>
+		</triggeringPolicy>
+		<encoder>
+			<pattern>%d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n${log.stackfilter.pattern}</pattern>
+		</encoder>
+	</appender>
+	<logger name="ca.uhn.fhir.log.empi_troubleshooting" level="TRACE">
+		<appender-ref ref="EMPI_TROUBLESHOOTING"/>
+	</logger>
+
 	<root level="INFO">
 		<appender-ref ref="STDOUT" />
 	</root>

src/test/java/ca/uhn/fhir/jpa/starter/ExampleServerR4IT.java

@@ -8,6 +8,7 @@ import ca.uhn.fhir.rest.client.api.IGenericClient;
 import ca.uhn.fhir.rest.client.api.ServerValidationModeEnum;
 import ca.uhn.fhir.rest.client.interceptor.LoggingInterceptor;
 import ca.uhn.fhir.test.utilities.JettyUtil;
+import ca.uhn.fhir.util.BundleUtil;
 import org.eclipse.jetty.server.Server;
 import org.eclipse.jetty.webapp.WebAppContext;
 import org.eclipse.jetty.websocket.api.Session;
@@ -17,6 +18,8 @@ import org.hl7.fhir.instance.model.api.IIdType;
 import org.hl7.fhir.r4.model.Bundle;
 import org.hl7.fhir.r4.model.Observation;
 import org.hl7.fhir.r4.model.Patient;
+import org.hl7.fhir.r4.model.Person;
+import org.hl7.fhir.r4.model.Reference;
 import org.hl7.fhir.r4.model.Subscription;
 import org.junit.AfterClass;
 import org.junit.BeforeClass;
@@ -24,11 +27,16 @@ import org.junit.Test;
 
 import java.net.URI;
 import java.nio.file.Paths;
+import java.util.Collection;
+import java.util.List;
+import java.util.Optional;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 
 import static ca.uhn.fhir.util.TestUtil.waitForSize;
+import static org.awaitility.Awaitility.await;
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
 
 public class ExampleServerR4IT {
 
@@ -43,6 +51,7 @@ public class ExampleServerR4IT {
         HapiProperties.setProperty(HapiProperties.DATASOURCE_URL, "jdbc:h2:mem:dbr4");
         HapiProperties.setProperty(HapiProperties.FHIR_VERSION, "R4");
         HapiProperties.setProperty(HapiProperties.SUBSCRIPTION_WEBSOCKET_ENABLED, "true");
+        HapiProperties.setProperty(HapiProperties.EMPI_ENABLED, "true");
         ourCtx = FhirContext.forR4();
     }
 
@@ -57,6 +66,27 @@ public class ExampleServerR4IT {
 
         Patient pt2 = ourClient.read().resource(Patient.class).withId(id).execute();
         assertEquals(methodName, pt2.getName().get(0).getFamily());
+
+        // Test EMPI
+
+        // Wait until the EMPI message has been processed
+        await().until(() -> getPeople().size() > 0);
+        List<Person> persons = getPeople();
+
+        // Verify a Person was created that links to our Patient
+        Optional<String> personLinkToCreatedPatient = persons.stream()
+          .map(Person::getLink)
+          .flatMap(Collection::stream)
+          .map(Person.PersonLinkComponent::getTarget)
+          .map(Reference::getReference)
+          .filter(pid -> id.toUnqualifiedVersionless().getValue().equals(pid))
+          .findAny();
+        assertTrue(personLinkToCreatedPatient.isPresent());
+    }
+
+  private List<Person> getPeople() {
+    Bundle bundle = ourClient.search().forResource(Person.class).cacheControl(new CacheControlDirective().setNoCache(true)).returnBundle(Bundle.class).execute();
+    return BundleUtil.toListOfResourcesOfType(ourCtx, bundle, Person.class);
   }
 
   @Test
@@ -78,7 +108,7 @@ public class ExampleServerR4IT {
         IIdType mySubscriptionId = methodOutcome.getId();
 
         // Wait for the subscription to be activated
-        waitForSize(1, () -> ourClient.search().forResource(Subscription.class).where(Subscription.STATUS.exactly().code("active")).cacheControl(new CacheControlDirective().setNoCache(true)).returnBundle(Bundle.class).execute().getEntry().size());
+        await().until(() -> activeSubscriptionCount() == 3);
 
         /*
          * Attach websocket
@@ -117,6 +147,10 @@ public class ExampleServerR4IT {
         ourClient.delete().resourceById(mySubscriptionId).execute();
     }
 
+  private int activeSubscriptionCount() {
+    return ourClient.search().forResource(Subscription.class).where(Subscription.STATUS.exactly().code("active")).cacheControl(new CacheControlDirective().setNoCache(true)).returnBundle(Bundle.class).execute().getEntry().size();
+  }
+
   @AfterClass
     public static void afterClass() throws Exception {
         ourServer.stop();