Added support for $lastn operation and fixed Elasticsearch indexing.

README.md

@@ -321,3 +321,8 @@ elasticsearch.password=SomePassword
 elasticsearch.required_index_status=YELLOW
 elasticsearch.schema_management_strategy=CREATE
 ```
+
+## Enabling LastN
+
+Set `hapi.fhir.lastn_enabled=true` in the [application.yaml](https://github.com/hapifhir/hapi-fhir-jpaserver-starter/blob/master/src/main/resources/application.yaml) file to enable the $lastn operation on this server.  Note that the $lastn operation relies on Elasticsearch, so for $lastn to work, indexing must be enabled using Elasticsearch.
+

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

@@ -62,6 +62,8 @@ public class AppProperties {
   private Partitioning partitioning = null;
   private List<ImplementationGuide> implementationGuides = null;
 
+  private Boolean lastn_enabled = false;
+
   public Integer getDefer_indexing_for_codesystems_of_size() {
     return defer_indexing_for_codesystems_of_size;
   }
@@ -382,6 +384,14 @@ public class AppProperties {
     this.narrative_enabled = narrative_enabled;
   }
 
+  public Boolean getLastn_enabled() {
+    return lastn_enabled;
+  }
+
+  public void setLastn_enabled(Boolean lastn_enabled) {
+    this.lastn_enabled = lastn_enabled;
+  }
+
   public static class Cors {
     private Boolean allow_Credentials = true;
     private List<String> allowed_origin = ImmutableList.of("*");

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

@@ -360,6 +360,12 @@ public class BaseJpaRestfulServer extends RestfulServer {
           .setInstallMode(PackageInstallationSpec.InstallModeEnum.STORE_AND_INSTALL));
       }
     }
+
+    if (appProperties.getLastn_enabled()) {
+      daoConfig.setLastNEnabled(true);
+    }
+
+
   }
 
 

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

@@ -1,5 +1,8 @@
 package ca.uhn.fhir.jpa.starter;
 
+import ca.uhn.fhir.jpa.search.elastic.ElasticsearchHibernatePropertiesBuilder;
+import org.hibernate.search.elasticsearch.cfg.ElasticsearchIndexStatus;
+import org.hibernate.search.elasticsearch.cfg.IndexSchemaManagementStrategy;
 import org.springframework.core.env.CompositePropertySource;
 import org.springframework.core.env.ConfigurableEnvironment;
 import org.springframework.core.env.EnumerablePropertySource;
@@ -35,9 +38,66 @@ public class EnvironmentHelper {
         properties.put(values[0], values[1]);
       });
     }
+
+    if (environment.getProperty("elasticsearch.enabled", Boolean.class) != null
+          && environment.getProperty("elasticsearch.enabled", Boolean.class) == true ){
+      ElasticsearchHibernatePropertiesBuilder builder = new ElasticsearchHibernatePropertiesBuilder();
+      ElasticsearchIndexStatus requiredIndexStatus = environment.getProperty("elasticsearch.required_index_status", ElasticsearchIndexStatus.class);
+      if (requiredIndexStatus == null) {
+        builder.setRequiredIndexStatus(ElasticsearchIndexStatus.YELLOW);
+      } else {
+        builder.setRequiredIndexStatus(requiredIndexStatus);
+      }
+
+      builder.setRestUrl(getElasticsearchServerUrl(environment));
+      builder.setUsername(getElasticsearchServerUsername(environment));
+      builder.setPassword(getElasticsearchServerPassword(environment));
+      IndexSchemaManagementStrategy indexSchemaManagementStrategy = environment.getProperty("elasticsearch.schema_management_strategy", IndexSchemaManagementStrategy.class);
+      if (indexSchemaManagementStrategy == null) {
+        builder.setIndexSchemaManagementStrategy(IndexSchemaManagementStrategy.CREATE);
+      } else {
+        builder.setIndexSchemaManagementStrategy(indexSchemaManagementStrategy);
+      }
+      //    pretty_print_json_log: false
+      Boolean refreshAfterWrite = environment.getProperty("elasticsearch.debug.refresh_after_write", Boolean.class);
+      if (refreshAfterWrite == null) {
+        builder.setDebugRefreshAfterWrite(false);
+      } else {
+        builder.setDebugRefreshAfterWrite(refreshAfterWrite);
+      }
+      //    pretty_print_json_log: false
+      Boolean prettyPrintJsonLog = environment.getProperty("elasticsearch.debug.pretty_print_json_log", Boolean.class);
+      if (prettyPrintJsonLog == null) {
+        builder.setDebugPrettyPrintJsonLog(false);
+      } else {
+        builder.setDebugPrettyPrintJsonLog(prettyPrintJsonLog);
+      }
+      builder.apply(properties);
+    }
+
     return properties;
   }
 
+  public static String getElasticsearchServerUrl(ConfigurableEnvironment environment) {
+    return environment.getProperty("elasticsearch.rest_url", String.class);
+  }
+
+  public static String getElasticsearchServerUsername(ConfigurableEnvironment environment) {
+    return environment.getProperty("elasticsearch.username");
+  }
+
+  public static String getElasticsearchServerPassword(ConfigurableEnvironment environment) {
+    return environment.getProperty("elasticsearch.password");
+  }
+
+  public static Boolean isElasticsearchEnabled(ConfigurableEnvironment environment) {
+    if (environment.getProperty("elasticsearch.enabled", Boolean.class) != null) {
+      return environment.getProperty("elasticsearch.enabled", Boolean.class);
+    } else {
+      return false;
+    }
+  }
+
   public static Map<String, Object> getPropertiesStartingWith(ConfigurableEnvironment aEnv,
                                                               String aKeyPrefix) {
     Map<String, Object> result = new HashMap<>();

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

@@ -3,6 +3,7 @@ package ca.uhn.fhir.jpa.starter;
 import ca.uhn.fhir.context.ConfigurationException;
 import ca.uhn.fhir.jpa.config.BaseJavaConfigDstu3;
 import ca.uhn.fhir.jpa.search.DatabaseBackedPagingProvider;
+import ca.uhn.fhir.jpa.search.lastn.ElasticsearchSvcImpl;
 import ca.uhn.fhir.jpa.starter.annotations.OnDSTU3Condition;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.*;
@@ -63,4 +64,18 @@ public class FhirServerConfigDstu3 extends BaseJavaConfigDstu3 {
     return retVal;
   }
 
+  @Bean()
+  public ElasticsearchSvcImpl elasticsearchSvc() {
+    if (EnvironmentHelper.isElasticsearchEnabled(configurableEnvironment)) {
+      String elasticsearchUrl = EnvironmentHelper.getElasticsearchServerUrl(configurableEnvironment);
+      String elasticsearchHost = elasticsearchUrl.substring(elasticsearchUrl.indexOf("://")+3, elasticsearchUrl.lastIndexOf(":"));
+      String elasticsearchUsername = EnvironmentHelper.getElasticsearchServerUsername(configurableEnvironment);
+      String elasticsearchPassword = EnvironmentHelper.getElasticsearchServerPassword(configurableEnvironment);
+      int elasticsearchPort = Integer.parseInt(elasticsearchUrl.substring(elasticsearchUrl.lastIndexOf(":")+1));
+      return new ElasticsearchSvcImpl(elasticsearchHost, elasticsearchPort, elasticsearchUsername, elasticsearchPassword);
+    } else {
+      return null;
+    }
+  }
+
 }

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

@@ -3,6 +3,7 @@ package ca.uhn.fhir.jpa.starter;
 import ca.uhn.fhir.context.ConfigurationException;
 import ca.uhn.fhir.jpa.config.BaseJavaConfigR4;
 import ca.uhn.fhir.jpa.search.DatabaseBackedPagingProvider;
+import ca.uhn.fhir.jpa.search.lastn.ElasticsearchSvcImpl;
 import ca.uhn.fhir.jpa.starter.annotations.OnR4Condition;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Bean;
@@ -66,4 +67,18 @@ public class FhirServerConfigR4 extends BaseJavaConfigR4 {
     return retVal;
   }
 
+  @Bean()
+  public ElasticsearchSvcImpl elasticsearchSvc() {
+    if (EnvironmentHelper.isElasticsearchEnabled(configurableEnvironment)) {
+      String elasticsearchUrl = EnvironmentHelper.getElasticsearchServerUrl(configurableEnvironment);
+      String elasticsearchHost = elasticsearchUrl.substring(elasticsearchUrl.indexOf("://")+3, elasticsearchUrl.lastIndexOf(":"));
+      String elasticsearchUsername = EnvironmentHelper.getElasticsearchServerUsername(configurableEnvironment);
+      String elasticsearchPassword = EnvironmentHelper.getElasticsearchServerPassword(configurableEnvironment);
+      int elasticsearchPort = Integer.parseInt(elasticsearchUrl.substring(elasticsearchUrl.lastIndexOf(":")+1));
+      return new ElasticsearchSvcImpl(elasticsearchHost, elasticsearchPort, elasticsearchUsername, elasticsearchPassword);
+    } else {
+      return null;
+    }
+  }
+
 }

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

@@ -3,6 +3,7 @@ package ca.uhn.fhir.jpa.starter;
 import ca.uhn.fhir.context.ConfigurationException;
 import ca.uhn.fhir.jpa.config.BaseJavaConfigR5;
 import ca.uhn.fhir.jpa.search.DatabaseBackedPagingProvider;
+import ca.uhn.fhir.jpa.search.lastn.ElasticsearchSvcImpl;
 import ca.uhn.fhir.jpa.starter.annotations.OnR5Condition;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Bean;
@@ -66,4 +67,19 @@ public class FhirServerConfigR5 extends BaseJavaConfigR5 {
     return retVal;
   }
 
+  @Bean()
+  public ElasticsearchSvcImpl elasticsearchSvc() {
+    if (EnvironmentHelper.isElasticsearchEnabled(configurableEnvironment)) {
+      String elasticsearchUrl = EnvironmentHelper.getElasticsearchServerUrl(configurableEnvironment);
+      String elasticsearchHost = elasticsearchUrl.substring(elasticsearchUrl.indexOf("://")+3, elasticsearchUrl.lastIndexOf(":"));
+      String elasticsearchUsername = EnvironmentHelper.getElasticsearchServerUsername(configurableEnvironment);
+      String elasticsearchPassword = EnvironmentHelper.getElasticsearchServerPassword(configurableEnvironment);
+      int elasticsearchPort = Integer.parseInt(elasticsearchUrl.substring(elasticsearchUrl.lastIndexOf(":")+1));
+      return new ElasticsearchSvcImpl(elasticsearchHost, elasticsearchPort, elasticsearchUsername, elasticsearchPassword);
+    } else {
+      return null;
+    }
+  }
+
+
 }

src/main/resources/application.yaml

@@ -120,6 +120,7 @@ hapi:
 #        startTlsEnable:
 #        startTlsRequired:
 #        quitWait:
+#    lastn_enabled: true
 
 
 #

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

@@ -0,0 +1,144 @@
+package ca.uhn.fhir.jpa.starter;
+
+import ca.uhn.fhir.context.ConfigurationException;
+import ca.uhn.fhir.context.FhirContext;
+import ca.uhn.fhir.jpa.search.lastn.ElasticsearchSvcImpl;
+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 org.hl7.fhir.instance.model.api.IIdType;
+import org.hl7.fhir.r4.model.Bundle;
+import org.hl7.fhir.r4.model.DateTimeType;
+import org.hl7.fhir.r4.model.IntegerType;
+import org.hl7.fhir.r4.model.Observation;
+import org.hl7.fhir.r4.model.Parameters;
+import org.hl7.fhir.r4.model.Patient;
+import org.hl7.fhir.r4.model.StringType;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.boot.test.util.TestPropertyValues;
+import org.springframework.boot.web.server.LocalServerPort;
+import org.springframework.context.ApplicationContextInitializer;
+import org.springframework.context.ConfigurableApplicationContext;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit.jupiter.SpringExtension;
+import pl.allegro.tech.embeddedelasticsearch.EmbeddedElastic;
+import pl.allegro.tech.embeddedelasticsearch.PopularProperties;
+
+import javax.annotation.PreDestroy;
+import java.io.IOException;
+import java.util.Date;
+import java.util.GregorianCalendar;
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+@ExtendWith(SpringExtension.class)
+@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT, classes = Application.class, properties =
+  {
+    "spring.batch.job.enabled=false",
+    "spring.datasource.url=jdbc:h2:mem:dbr4",
+    "hapi.fhir.fhir_version=r4",
+    "hapi.fhir.lastn_enabled=true",
+    "elasticsearch.enabled=true",
+    // Because the port is set randomly, we will set the rest_url using the Initializer.
+    // "elasticsearch.rest_url='http://localhost:9200'",
+    "elasticsearch.username=SomeUsername",
+    "elasticsearch.password=SomePassword"
+  })
+@ContextConfiguration(initializers = ElasticsearchLastNR4IT.Initializer.class)
+public class ElasticsearchLastNR4IT {
+
+  private IGenericClient ourClient;
+  private FhirContext ourCtx;
+
+  private static final String ELASTIC_VERSION = "6.5.4";
+  private static EmbeddedElastic embeddedElastic;
+
+  @Autowired
+  private ElasticsearchSvcImpl myElasticsearchSvc;
+
+  @BeforeAll
+  public static void beforeClass() {
+
+    embeddedElastic = null;
+    try {
+      embeddedElastic = EmbeddedElastic.builder()
+        .withElasticVersion(ELASTIC_VERSION)
+        .withSetting(PopularProperties.TRANSPORT_TCP_PORT, 0)
+        .withSetting(PopularProperties.HTTP_PORT, 0)
+        .withSetting(PopularProperties.CLUSTER_NAME, UUID.randomUUID())
+        .withStartTimeout(60, TimeUnit.SECONDS)
+        .build()
+        .start();
+    } catch (IOException | InterruptedException e) {
+      throw new ConfigurationException(e);
+    }
+  }
+
+  @PreDestroy
+  public void stop() {
+    embeddedElastic.stop();
+  }
+
+  @LocalServerPort
+  private int port;
+
+  @Test
+  void testLastN() throws IOException {
+
+    Patient pt = new Patient();
+    pt.addName().setFamily("Lastn").addGiven("Arthur");
+    IIdType id = ourClient.create().resource(pt).execute().getId().toUnqualifiedVersionless();
+
+    Observation obs = new Observation();
+    obs.getSubject().setReferenceElement(id);
+    String observationCode = "testobservationcode";
+    String codeSystem = "http://testobservationcodesystem";
+    obs.getCode().addCoding().setCode(observationCode).setSystem(codeSystem);
+    obs.setValue(new StringType(observationCode));
+    Date effectiveDtm = new GregorianCalendar().getTime();
+    obs.setEffective(new DateTimeType(effectiveDtm));
+    obs.getCategoryFirstRep().addCoding().setCode("testcategorycode").setSystem("http://testcategorycodesystem");
+    IIdType obsId = ourClient.create().resource(obs).execute().getId().toUnqualifiedVersionless();
+
+    myElasticsearchSvc.refreshIndex(ElasticsearchSvcImpl.OBSERVATION_INDEX);
+
+    Parameters output = ourClient.operation().onType(Observation.class).named("lastn")
+      .withParameter(Parameters.class, "max", new IntegerType(1))
+      .andParameter("subject", new StringType("Patient/" + id.getIdPart()))
+      .execute();
+    Bundle b = (Bundle) output.getParameter().get(0).getResource();
+    assertEquals(1, b.getTotal());
+    assertEquals(obsId, b.getEntry().get(0).getResource().getIdElement().toUnqualifiedVersionless());
+  }
+
+  @BeforeEach
+  void beforeEach() {
+
+    ourCtx = FhirContext.forR4();
+    ourCtx.getRestfulClientFactory().setServerValidationMode(ServerValidationModeEnum.NEVER);
+    ourCtx.getRestfulClientFactory().setSocketTimeout(1200 * 1000);
+    String ourServerBase = "http://localhost:" + port + "/fhir/";
+    ourClient = ourCtx.newRestfulGenericClient(ourServerBase);
+    ourClient.registerInterceptor(new LoggingInterceptor(true));
+  }
+
+  static class Initializer
+    implements ApplicationContextInitializer<ConfigurableApplicationContext> {
+
+    @Override
+    public void initialize(
+      ConfigurableApplicationContext configurableApplicationContext) {
+      // Since the port is dynamically generated, replace the URL with one that has the correct port
+      TestPropertyValues.of("elasticsearch.rest_url=http://localhost:" + embeddedElastic.getHttpPort())
+        .applyTo(configurableApplicationContext.getEnvironment());
+    }
+
+  }
+}