DsfClient: Improved async search impl, refactored operations with retry

New DelayStrategy to configure first and later delays. Constant 200ms
and exponential (100ms, 200ms, 400ms, 800ms, 800ms, ...) backoff
implementations.

Author: hhund

dsf-bpe/dsf-bpe-process-api-v2-impl/src/main/java/dev/dsf/bpe/v2/client/dsf/AbstractDsfClientJerseyWithRetry.java

@@ -34,18 +34,20 @@ public abstract class AbstractDsfClientJerseyWithRetry
 
 	protected final DsfClientJersey delegate;
 	private final int nTimes;
-	private final Duration delay;
+	private final DelayStrategy delayStrategy;
 
-	protected AbstractDsfClientJerseyWithRetry(DsfClientJersey delegate, int nTimes, Duration delay)
+	protected AbstractDsfClientJerseyWithRetry(DsfClientJersey delegate, int nTimes, DelayStrategy delayStrategy)
 	{
 		this.delegate = delegate;
 		this.nTimes = nTimes;
-		this.delay = delay;
+		this.delayStrategy = delayStrategy;
 	}
 
 	protected final <R> R retry(Supplier<R> supplier)
 	{
 		RuntimeException caughtException = null;
+		Duration delay = delayStrategy.getFirstDelay();
+
 		for (int tryNumber = 0; tryNumber <= nTimes || nTimes == RetryClient.RETRY_FOREVER; tryNumber++)
 		{
 			try
@@ -63,8 +65,11 @@ else if (nTimes != RetryClient.RETRY_FOREVER)
 				{
 					if (tryNumber < nTimes || nTimes == RetryClient.RETRY_FOREVER)
 					{
-						logger.warn("Caught {} - {}; trying again in {}s{}", e.getClass(), e.getMessage(),
-								delay.toSeconds(),
+						if (tryNumber > 0)
+							delay = delayStrategy.getNextDelay(delay);
+
+						logger.warn("Caught {} - {}; trying again in {}{}", e.getClass(), e.getMessage(),
+								delay.toString(),
 								nTimes == RetryClient.RETRY_FOREVER ? " (retry " + (tryNumber + 1) + ")" : "");
 
 						try

dsf-bpe/dsf-bpe-process-api-v2-impl/src/main/java/dev/dsf/bpe/v2/client/dsf/BasicDsfClientWithRetryImpl.java

@@ -16,10 +16,9 @@
 package dev.dsf.bpe.v2.client.dsf;
 
 import java.io.InputStream;
-import java.time.Duration;
 import java.util.List;
 import java.util.Map;
-import java.util.concurrent.Future;
+import java.util.concurrent.CompletableFuture;
 
 import org.hl7.fhir.r4.model.Binary;
 import org.hl7.fhir.r4.model.Bundle;
@@ -32,9 +31,9 @@
 
 class BasicDsfClientWithRetryImpl extends AbstractDsfClientJerseyWithRetry implements BasicDsfClient
 {
-	BasicDsfClientWithRetryImpl(DsfClientJersey delegate, int nTimes, Duration delay)
+	BasicDsfClientWithRetryImpl(DsfClientJersey delegate, int nTimes, DelayStrategy delayStrategy)
 	{
-		super(delegate, nTimes, delay);
+		super(delegate, nTimes, delayStrategy);
 	}
 
 	@Override
@@ -92,29 +91,29 @@ public Bundle search(Class<? extends Resource> resourceType, Map<String, List<St
 	}
 
 	@Override
-	public Future<Bundle> searchAsync(Duration initialPollingInterval, Class<? extends Resource> resourceType,
+	public CompletableFuture<Bundle> searchAsync(DelayStrategy delayStrategy, Class<? extends Resource> resourceType,
 			Map<String, List<String>> parameters)
 	{
-		return retry(() -> delegate.searchAsync(initialPollingInterval, resourceType, parameters));
+		return retry(() -> delegate.searchAsync(delayStrategy, resourceType, parameters));
 	}
 
 	@Override
-	public Future<Bundle> searchAsync(Duration initialPollingInterval, String url)
+	public CompletableFuture<Bundle> searchAsync(DelayStrategy delayStrategy, String url)
 	{
-		return retry(() -> delegate.searchAsync(initialPollingInterval, url));
+		return retry(() -> delegate.searchAsync(delayStrategy, url));
 	}
 
 	@Override
-	public Future<Bundle> searchAsyncWithStrictHandling(Duration initialPollingInterval,
+	public CompletableFuture<Bundle> searchAsyncWithStrictHandling(DelayStrategy delayStrategy,
 			Class<? extends Resource> resourceType, Map<String, List<String>> parameters)
 	{
-		return retry(() -> delegate.searchAsyncWithStrictHandling(initialPollingInterval, resourceType, parameters));
+		return retry(() -> delegate.searchAsyncWithStrictHandling(delayStrategy, resourceType, parameters));
 	}
 
 	@Override
-	public Future<Bundle> searchAsyncWithStrictHandling(Duration initialPollingInterval, String url)
+	public CompletableFuture<Bundle> searchAsyncWithStrictHandling(DelayStrategy delayStrategy, String url)
 	{
-		return retry(() -> delegate.searchAsyncWithStrictHandling(initialPollingInterval, url));
+		return retry(() -> delegate.searchAsyncWithStrictHandling(delayStrategy, url));
 	}
 
 	@Override

dsf-bpe/dsf-bpe-process-api-v2-impl/src/main/java/dev/dsf/bpe/v2/client/dsf/DsfClientJersey.java

@@ -29,7 +29,6 @@
 import java.util.TimeZone;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.Executors;
-import java.util.concurrent.Future;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Supplier;
@@ -968,7 +967,7 @@ public Bundle searchWithStrictHandling(Class<? extends Resource> resourceType, M
 	}
 
 	@Override
-	public Future<Bundle> searchAsync(Duration initialPollingInterval, Class<? extends Resource> resourceType,
+	public CompletableFuture<Bundle> searchAsync(DelayStrategy delayStrategy, Class<? extends Resource> resourceType,
 			Map<String, List<String>> parameters)
 	{
 		Objects.requireNonNull(resourceType, "resourceType");
@@ -980,19 +979,19 @@ public Future<Bundle> searchAsync(Duration initialPollingInterval, Class<? exten
 				target = target.queryParam(entry.getKey(), entry.getValue().toArray());
 		}
 
-		return doSearchAsync(initialPollingInterval, target, false);
+		return doSearchAsync(delayStrategy, target, false);
 	}
 
 	@Override
-	public Future<Bundle> searchAsync(Duration initialPollingInterval, String url)
+	public CompletableFuture<Bundle> searchAsync(DelayStrategy delayStrategy, String url)
 	{
 		checkUri(url);
 
-		return doSearchAsync(initialPollingInterval, client.target(url), false);
+		return doSearchAsync(delayStrategy, client.target(url), false);
 	}
 
 	@Override
-	public Future<Bundle> searchAsyncWithStrictHandling(Duration initialPollingInterval,
+	public CompletableFuture<Bundle> searchAsyncWithStrictHandling(DelayStrategy delayStrategy,
 			Class<? extends Resource> resourceType, Map<String, List<String>> parameters)
 	{
 		Objects.requireNonNull(resourceType, "resourceType");
@@ -1004,15 +1003,15 @@ public Future<Bundle> searchAsyncWithStrictHandling(Duration initialPollingInter
 				target = target.queryParam(entry.getKey(), entry.getValue().toArray());
 		}
 
-		return doSearchAsync(initialPollingInterval, target, true);
+		return doSearchAsync(delayStrategy, target, true);
 	}
 
 	@Override
-	public Future<Bundle> searchAsyncWithStrictHandling(Duration initialPollingInterval, String url)
+	public CompletableFuture<Bundle> searchAsyncWithStrictHandling(DelayStrategy delayStrategy, String url)
 	{
 		checkUri(url);
 
-		return doSearchAsync(initialPollingInterval, client.target(url), true);
+		return doSearchAsync(delayStrategy, client.target(url), true);
 	}
 
 	private void checkUri(String url)
@@ -1022,9 +1021,11 @@ private void checkUri(String url)
 			throw new RuntimeException("url is blank");
 		if (!url.startsWith(baseUrl))
 			throw new RuntimeException("url not starting with client base url");
+		if (url.startsWith(baseUrl + "@"))
+			throw new RuntimeException("url starting with client base url + @");
 	}
 
-	private Future<Bundle> doSearchAsync(Duration initialPollingInterval, WebTarget target, boolean strict)
+	private CompletableFuture<Bundle> doSearchAsync(DelayStrategy delayStrategy, WebTarget target, boolean strict)
 	{
 		Builder requestBuilder = target.request().header(Constants.HEADER_PREFER,
 				Constants.HEADER_PREFER_RESPOND_ASYNC);
@@ -1044,11 +1045,22 @@ public void completed(Response response)
 				else if (Status.ACCEPTED.getStatusCode() == response.getStatus())
 				{
 					String location = response.getHeaderString(HttpHeaders.LOCATION);
-					if (location == null)
-						throw new RuntimeException("202 Accepted without Location header");
+					if (location == null || location.isBlank())
+					{
+						logger.warn("202 Accepted without Location header");
+
+						location = response.getHeaderString(HttpHeaders.CONTENT_LOCATION);
+						if (location == null || location.isBlank())
+							throw new RuntimeException(
+									"202 Accepted without Location and without Content-Location header");
+						else
+							logger.warn("202 Accepted with Content-Location header");
+					}
+
+					checkUri(location);
 
 					response.close();
-					pollUntilComplete(location, initialPollingInterval.toMillis(), resultFuture);
+					pollUntilComplete(location, delayStrategy, resultFuture);
 				}
 				else
 					resultFuture.completeExceptionally(handleError(response));
@@ -1064,36 +1076,32 @@ public void failed(Throwable throwable)
 		return resultFuture;
 	}
 
-	private void pollUntilComplete(String pollUrl, long pollIntervalMillis, CompletableFuture<Bundle> resultFuture)
+	private void pollUntilComplete(String location, DelayStrategy delayStrategy, CompletableFuture<Bundle> resultFuture)
 	{
 		ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();
 
 		Runnable poll = new Runnable()
 		{
-			private long pollInterval = pollIntervalMillis;
+			private Duration delay = delayStrategy.getFirstDelay();
 
 			@Override
 			public void run()
 			{
+				if (resultFuture.isCancelled())
+					return;
+
 				try
 				{
-					checkUri(pollUrl);
-
-					Response response = client.target(pollUrl).request().get();
+					Response response = client.target(location).request().get();
 
 					if (Status.OK.getStatusCode() == response.getStatus())
 						resultFuture.complete(response.readEntity(Bundle.class));
 					else if (Status.ACCEPTED.getStatusCode() == response.getStatus())
 					{
-						String location = response.getHeaderString(HttpHeaders.LOCATION);
-						if (location == null)
-							throw new RuntimeException("202 Accepted without Location header");
-
 						response.close();
 
-						pollInterval *= 2;
-
-						executor.schedule(this, pollInterval, TimeUnit.MILLISECONDS);
+						delay = delayStrategy.getNextDelay(delay);
+						executor.schedule(this, delay.toMillis(), TimeUnit.MILLISECONDS);
 					}
 					else
 						resultFuture.completeExceptionally(handleError(response));
@@ -1106,7 +1114,7 @@ else if (Status.ACCEPTED.getStatusCode() == response.getStatus())
 			}
 		};
 
-		executor.schedule(poll, pollIntervalMillis, TimeUnit.MILLISECONDS);
+		executor.schedule(poll, delayStrategy.getFirstDelay().toMillis(), TimeUnit.MILLISECONDS);
 	}
 
 	@Override
@@ -1165,23 +1173,21 @@ public StructureDefinition generateSnapshot(StructureDefinition differential)
 	}
 
 	@Override
-	public BasicDsfClient withRetry(int nTimes, Duration delay)
+	public BasicDsfClient withRetry(int nTimes, DelayStrategy delayStrategy)
 	{
 		if (nTimes < 0)
 			throw new IllegalArgumentException("nTimes < 0");
-		if (delay == null || delay.isNegative())
-			throw new IllegalArgumentException("delay null or negative");
+		Objects.requireNonNull(delayStrategy, "delayStrategy");
 
-		return new BasicDsfClientWithRetryImpl(this, nTimes, delay);
+		return new BasicDsfClientWithRetryImpl(this, nTimes, delayStrategy);
 	}
 
 	@Override
-	public BasicDsfClient withRetryForever(Duration delay)
+	public BasicDsfClient withRetryForever(DelayStrategy delayStrategy)
 	{
-		if (delay == null || delay.isNegative())
-			throw new IllegalArgumentException("delay null or negative");
+		Objects.requireNonNull(delayStrategy, "delayStrategy");
 
-		return new BasicDsfClientWithRetryImpl(this, RETRY_FOREVER, delay);
+		return new BasicDsfClientWithRetryImpl(this, RETRY_FOREVER, delayStrategy);
 	}
 
 	@Override

dsf-bpe/dsf-bpe-process-api-v2-impl/src/main/java/dev/dsf/bpe/v2/client/dsf/PreferReturnMinimalRetryImpl.java

@@ -16,7 +16,6 @@
 package dev.dsf.bpe.v2.client.dsf;
 
 import java.io.InputStream;
-import java.time.Duration;
 import java.util.List;
 import java.util.Map;
 
@@ -28,9 +27,9 @@
 
 class PreferReturnMinimalRetryImpl extends AbstractDsfClientJerseyWithRetry implements PreferReturnMinimal
 {
-	PreferReturnMinimalRetryImpl(DsfClientJersey delegate, int nTimes, Duration delay)
+	PreferReturnMinimalRetryImpl(DsfClientJersey delegate, int nTimes, DelayStrategy delayStrategy)
 	{
-		super(delegate, nTimes, delay);
+		super(delegate, nTimes, delayStrategy);
 	}
 
 	@Override

dsf-bpe/dsf-bpe-process-api-v2-impl/src/main/java/dev/dsf/bpe/v2/client/dsf/PreferReturnMinimalWithRetryImpl.java

@@ -16,7 +16,6 @@
 package dev.dsf.bpe.v2.client.dsf;
 
 import java.io.InputStream;
-import java.time.Duration;
 import java.util.List;
 import java.util.Map;
 
@@ -78,22 +77,22 @@ public Bundle postBundle(Bundle bundle)
 	}
 
 	@Override
-	public PreferReturnMinimal withRetry(int nTimes, Duration delay)
+	public PreferReturnMinimal withRetry(int nTimes, DelayStrategy delayStrategy)
 	{
 		if (nTimes < 0)
 			throw new IllegalArgumentException("nTimes < 0");
-		if (delay == null || delay.isNegative())
-			throw new IllegalArgumentException("delay null or negative");
+		if (delayStrategy == null)
+			throw new IllegalArgumentException("delayStrategy null");
 
-		return new PreferReturnMinimalRetryImpl(delegate, nTimes, delay);
+		return new PreferReturnMinimalRetryImpl(delegate, nTimes, delayStrategy);
 	}
 
 	@Override
-	public PreferReturnMinimal withRetryForever(Duration delay)
+	public PreferReturnMinimal withRetryForever(DelayStrategy delayStrategy)
 	{
-		if (delay == null || delay.isNegative())
-			throw new IllegalArgumentException("delay null or negative");
+		if (delayStrategy == null)
+			throw new IllegalArgumentException("delayStrategy null");
 
-		return new PreferReturnMinimalRetryImpl(delegate, RETRY_FOREVER, delay);
+		return new PreferReturnMinimalRetryImpl(delegate, RETRY_FOREVER, delayStrategy);
 	}
 }

dsf-bpe/dsf-bpe-process-api-v2-impl/src/main/java/dev/dsf/bpe/v2/client/dsf/PreferReturnOutcomeRetryImpl.java

@@ -16,7 +16,6 @@
 package dev.dsf.bpe.v2.client.dsf;
 
 import java.io.InputStream;
-import java.time.Duration;
 import java.util.List;
 import java.util.Map;
 
@@ -28,9 +27,9 @@
 
 class PreferReturnOutcomeRetryImpl extends AbstractDsfClientJerseyWithRetry implements PreferReturnOutcome
 {
-	PreferReturnOutcomeRetryImpl(DsfClientJersey delegate, int nTimes, Duration delay)
+	PreferReturnOutcomeRetryImpl(DsfClientJersey delegate, int nTimes, DelayStrategy delayStrategy)
 	{
-		super(delegate, nTimes, delay);
+		super(delegate, nTimes, delayStrategy);
 	}
 
 	@Override