Refactoring and adding tests for BulkExport template (states for the …

…async call protocol).

bulkclient/src/main/java/au/csiro/pathling/export/BulkExportClient.java

@@ -39,6 +39,7 @@
 import au.csiro.pathling.export.utils.ExecutorServiceResource;
 import au.csiro.pathling.export.utils.TimeoutUtils;
 import au.csiro.pathling.export.ws.AsyncConfig;
+import au.csiro.pathling.export.ws.BulkExportAsyncService;
 import au.csiro.pathling.export.ws.BulkExportRequest;
 import au.csiro.pathling.export.ws.BulkExportRequest.GroupLevel;
 import au.csiro.pathling.export.ws.BulkExportRequest.Level;
@@ -210,6 +211,7 @@ public static class BulkExportClientBuilder {
 
  /**
  * Create a builder for a system-level export.
+ *
  * @return the builder configured for a system-level export
  */
  @Nonnull
@@ -219,15 +221,17 @@ public static BulkExportClientBuilder systemBuilder() {
 
  /**
  * Create a builder for a patient-level export.
+ *
  * @return the builder configured for a patient-level export
  */
  @Nonnull
  public static BulkExportClientBuilder patientBuilder() {
  return BulkExportClient.builder().withLevel(new PatientLevel());
  }
- 
+
  /**
  * Create a builder for a group-level export.
+ *
  * @param groupId the group ID to export
  * @return the builder configured for a group-level export
  */
@@ -249,8 +253,8 @@ public BulkExportResult export() {
  final CloseableHttpClient httpClient = createHttpClient(tokenProvider);
  final ExecutorServiceResource executorServiceResource = createExecutorServiceResource()
  ) {
- final BulkExportTemplate bulkExportTemplate = new BulkExportTemplate(httpClient,
- URI.create(fhirEndpointUrl),
+ final BulkExportTemplate bulkExportTemplate = new BulkExportTemplate(
+ new BulkExportAsyncService(httpClient, URI.create(fhirEndpointUrl)),
  asyncConfig);
  final UrlDownloadTemplate downloadTemplate = new UrlDownloadTemplate(httpClient,
  executorServiceResource.getExecutorService());

bulkclient/src/main/java/au/csiro/pathling/export/utils/WebUtils.java

@@ -0,0 +1,71 @@
+/*
+ * Copyright 2023 Commonwealth Scientific and Industrial Research
+ * Organisation (CSIRO) ABN 41 687 119 230.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package au.csiro.pathling.export.utils;
+
+import au.csiro.pathling.export.fhir.FhirJsonSupport;
+import java.net.URI;
+import javax.annotation.Nonnull;
+import lombok.experimental.UtilityClass;
+import org.apache.http.Consts;
+import org.apache.http.HttpEntity;
+import org.apache.http.entity.ContentType;
+import org.apache.http.entity.StringEntity;
+
+/**
+ * Utility methods for working with web resources.
+ */
+@UtilityClass
+public class WebUtils {
+
+ /**
+ * The content type for FHIR JSON.
+ */
+ public static final ContentType APPLICATION_FHIR_JSON = ContentType.create(
+ "application/fhir+json",
+ Consts.UTF_8);
+ /**
+ * HTTP status code for too many requests.
+ */
+ public static final int HTTP_TOO_MANY_REQUESTS = 429;
+
+
+ /**
+ * Converts a FHIR resource to an HTTP entity using GSON serialization.
+ *
+ * @param fhirResource the FHIR resource to convert.
+ * @return the HTTP entity.
+ */
+ @Nonnull
+ public static HttpEntity toFhirJsonEntity(@Nonnull final Object fhirResource) {
+ return new StringEntity(
+ FhirJsonSupport.toJson(fhirResource), APPLICATION_FHIR_JSON);
+ }
+
+ /**
+ * Ensures that the URI ends with a slash.
+ *
+ * @param uri the URI to ensure ends with a slash.
+ * @return the URI with a trailing slash.
+ */
+ @Nonnull
+ public static URI ensurePathEndsWithSlash(@Nonnull final URI uri) {
+ return uri.getPath().endsWith("/")
+ ? uri
+ : URI.create(uri + "/");
+ }
+}

bulkclient/src/main/java/au/csiro/pathling/export/ws/AsyncConfig.java

@@ -55,7 +55,7 @@ public class AsyncConfig {
  * The delay to retry after the HTTP 429 'Too many requests' response.
  */
  @Builder.Default
- Duration tooManyRequestsDelay = Duration.ofSeconds(2);
+ Duration tooManyRequestsDelay = Duration.ofSeconds(10);
 
  /**
  * The maximum number of consecutive transient errors to retry before giving up.

bulkclient/src/main/java/au/csiro/pathling/export/ws/BulkExportAsyncService.java

@@ -0,0 +1,79 @@
+/*
+ * Copyright 2023 Commonwealth Scientific and Industrial Research
+ * Organisation (CSIRO) ABN 41 687 119 230.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package au.csiro.pathling.export.ws;
+
+import lombok.AllArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.http.client.HttpClient;
+import org.apache.http.client.methods.HttpGet;
+import org.apache.http.client.methods.HttpPost;
+import org.apache.http.client.methods.HttpUriRequest;
+import org.apache.http.util.EntityUtils;
+
+import javax.annotation.Nonnull;
+import java.io.IOException;
+import java.net.URI;
+
+
+/**
+ * Service for handling asynchronous requests for bulk export.
+ */
+@AllArgsConstructor
+@Slf4j
+public class BulkExportAsyncService {
+
+ @Nonnull
+ final HttpClient httpClient;
+
+ @Nonnull
+ final URI fhirEndpointUri;
+
+ /**
+ * Kicks off a bulk export request.
+ *
+ * @param request the request to kick off
+ * @return the {@link AsyncResponse}response
+ * @throws IOException if an error occurs during the request
+ */
+ @Nonnull
+ AsyncResponse kickOff(@Nonnull final BulkExportRequest request) throws IOException {
+ final HttpUriRequest httpRequest = request.toHttpRequest(fhirEndpointUri);
+ log.debug("KickOff: Request: {}", httpRequest);
+ if (httpRequest instanceof HttpPost) {
+ log.debug("KickOff: Request body: {}",
+ EntityUtils.toString(((HttpPost) httpRequest).getEntity()));
+ }
+ return httpClient.execute(httpRequest, AsynResponseHandler.of(BulkExportResponse.class));
+ }
+
+ /**
+ * Checks the status of a bulk export request. Returns an {@link AsyncResponse} that may contain a
+ * {@link AcceptedAsyncResponse} or a {@link BulkExportResponse} on completion.
+ *
+ * @param statusUri the status URI
+ * @return the {@link AsyncResponse} response
+ * @throws IOException if an error occurs during the request
+ */
+ @Nonnull
+ AsyncResponse checkStatus(@Nonnull final URI statusUri) throws IOException {
+ log.debug("Poolin: Get status from: " + statusUri);
+ final HttpUriRequest statusRequest = new HttpGet(statusUri);
+ statusRequest.setHeader("accept", "application/json");
+ return httpClient.execute(statusRequest, AsynResponseHandler.of(BulkExportResponse.class));
+ }
+}

bulkclient/src/main/java/au/csiro/pathling/export/ws/BulkExportRequest.java

@@ -17,21 +17,30 @@
 
 package au.csiro.pathling.export.ws;
 
+import au.csiro.pathling.export.fhir.FhirUtils;
 import au.csiro.pathling.export.fhir.Parameters;
 import au.csiro.pathling.export.fhir.Parameters.Parameter;
 import au.csiro.pathling.export.fhir.Reference;
+import au.csiro.pathling.export.utils.WebUtils;
+import au.csiro.pathling.utilities.Lists;
+import java.net.URI;
+import java.net.URISyntaxException;
 import java.time.Instant;
 import java.util.Collections;
 import java.util.List;
+import java.util.Objects;
 import java.util.Optional;
 import java.util.function.Function;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 import javax.annotation.Nonnull;
 import javax.annotation.Nullable;
-import au.csiro.pathling.utilities.Lists;
 import lombok.Builder;
 import lombok.Value;
+import org.apache.http.client.methods.HttpGet;
+import org.apache.http.client.methods.HttpPost;
+import org.apache.http.client.methods.HttpUriRequest;
+import org.apache.http.client.utils.URIBuilder;
 
 /**
  * Represents a request to initiate a bulk export operation.
@@ -203,4 +212,55 @@ public Parameters toParameters() {
  .collect(Collectors.toUnmodifiableList());
  return Parameters.of(params);
  }
+
+ @Nonnull
+ public URI toRequestURI(@Nonnull final URI endpointUri) {
+ final URIBuilder uriBuilder = new URIBuilder(endpointUri);
+ if (get_outputFormat() != null) {
+ uriBuilder.addParameter("_outputFormat",
+ Objects.requireNonNull(get_outputFormat()));
+ }
+ if (get_since() != null) {
+ uriBuilder.addParameter("_since",
+ FhirUtils.formatFhirInstant(Objects.requireNonNull(get_since())));
+ }
+ if (!get_type().isEmpty()) {
+ uriBuilder.addParameter("_type", String.join(",", get_type()));
+ }
+ if (!get_elements().isEmpty()) {
+ uriBuilder.addParameter("_elements", String.join(",", get_elements()));
+ }
+ if (!get_typeFilter().isEmpty()) {
+ uriBuilder.addParameter("_typeFilter", String.join(",", get_typeFilter()));
+ }
+ try {
+ return uriBuilder.build();
+ } catch (final URISyntaxException ex) {
+ throw new IllegalArgumentException("Error building URI", ex);
+ }
+ }
+
+ @Nonnull
+ public HttpUriRequest toHttpRequest(@Nonnull final URI fhirEndpointUri) {
+ // check if patient is supported for the operation
+ if (!this.getLevel().isPatientSupported() && !this.getPatient().isEmpty()) {
+ throw new IllegalStateException(
+ "'patient' is not supported for operation: " + this.getLevel());
+ }
+ final URI endpointUri = WebUtils.ensurePathEndsWithSlash(fhirEndpointUri).resolve(
+ this.getLevel().getPath());
+ final HttpUriRequest httpRequest;
+
+ if (this.getPatient().isEmpty()) {
+ httpRequest = new HttpGet(this.toRequestURI(endpointUri));
+ } else {
+ final HttpPost postRequest = new HttpPost(endpointUri);
+ postRequest.setEntity(WebUtils.toFhirJsonEntity(this.toParameters()));
+ httpRequest = postRequest;
+ }
+ httpRequest.setHeader("accept", WebUtils.APPLICATION_FHIR_JSON.getMimeType());
+ httpRequest.setHeader("prefer", "respond-async");
+ return httpRequest;
+ }
+
 }