async retry actually async

Author: l-trotta

java-client/src/main/java/co/elastic/clients/transport/rest_client/RequestFuture.java

@@ -0,0 +1,26 @@
+package co.elastic.clients.transport.rest_client;
+
+import org.elasticsearch.client.Cancellable;
+
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * The {@code Future} implementation returned by async requests.
+ * It wraps the RestClient's cancellable and propagates cancellation.
+ */
+public class RequestFuture<T> extends CompletableFuture<T> {
+    private volatile Cancellable cancellable;
+
+    @Override
+    public boolean cancel(boolean mayInterruptIfRunning) {
+        boolean cancelled = super.cancel(mayInterruptIfRunning);
+        if (cancelled && cancellable != null) {
+            cancellable.cancel();
+        }
+        return cancelled;
+    }
+
+    public void setCancellable(Cancellable cancellable) {
+        this.cancellable = cancellable;
+    }
+}

java-client/src/main/java/co/elastic/clients/transport/rest_client/RestClientHttpClient.java

@@ -49,23 +49,6 @@ public class RestClientHttpClient implements TransportHttpClient {
 
     private static final ConcurrentHashMap<String, ContentType> ContentTypeCache = new ConcurrentHashMap<>();
 
-    /**
-     * The {@code Future} implementation returned by async requests.
-     * It wraps the RestClient's cancellable and propagates cancellation.
-     */
-    private static class RequestFuture<T> extends CompletableFuture<T> {
-        private volatile Cancellable cancellable;
-
-        @Override
-        public boolean cancel(boolean mayInterruptIfRunning) {
-            boolean cancelled = super.cancel(mayInterruptIfRunning);
-            if (cancelled && cancellable != null) {
-                cancellable.cancel();
-            }
-            return cancelled;
-        }
-    }
-
     private final RestClient restClient;
 
     public RestClientHttpClient(RestClient restClient) {
@@ -110,7 +93,7 @@ public CompletableFuture<Response> performRequestAsync(
             return future;
         }
 
-        future.cancellable = restClient.performRequestAsync(restRequest, new ResponseListener() {
+        future.setCancellable(restClient.performRequestAsync(restRequest, new ResponseListener() {
             @Override
             public void onSuccess(org.elasticsearch.client.Response response) {
                 future.complete(new RestResponse(response));
@@ -120,7 +103,7 @@ public void onSuccess(org.elasticsearch.client.Response response) {
             public void onFailure(Exception exception) {
                 future.completeExceptionally(exception);
             }
-        });
+        }));
 
         return future;
     }

java-client/src/main/java/co/elastic/clients/transport/rest_client/RetryRestClientHttpClient.java

@@ -31,12 +31,13 @@ public Response performRequestRetry(String endpointId, @Nullable Node node, Requ
         try {
             return delegate.performRequest(endpointId, node, request, options);
         } catch (ResponseException e) {
-            if (e.getResponse().getStatusLine().getStatusCode() == 503) { // TODO list of statuses
+            if (e.getResponse().getStatusLine().getStatusCode() == 503) { // TODO list of statuses, configurable or hardcoded?
                 // synchronous retry
                 if (backoffIter.hasNext()) {
                     try {
-                        Thread.sleep(backoffIter.next()); // TODO ... no?
+                        Thread.sleep(backoffIter.next());
                     } catch (InterruptedException ie) {
+                        throw e; // TODO okay with masking IE and just returning original exception?
                     }
                     System.out.println("Retrying");
                     return performRequestRetry(endpointId, node, request, options, backoffIter);
@@ -50,33 +51,41 @@ public Response performRequestRetry(String endpointId, @Nullable Node node, Requ
     @Override
     public CompletableFuture<Response> performRequestAsync(String endpointId, @Nullable Node node,
                                                            Request request, TransportOptions options) {
-        return performRequestAsyncRetry(endpointId, node, request, options, backoffPolicy.iterator());
+        RequestFuture<Response> futureResult = new RequestFuture<>();
+        return performRequestAsyncRetry(endpointId, node, request, options, backoffPolicy.iterator(),
+            futureResult);
     }
 
     public CompletableFuture<Response> performRequestAsyncRetry(String endpointId, @Nullable Node node,
                                                                 Request request,
                                                                 TransportOptions options,
-                                                                Iterator<Long> backoffIter) {
-        CompletableFuture<Response> fut = delegate.performRequestAsync(endpointId, node, request, options);
-        try {
-            fut.get(); // TODO is this problematic?
-            return fut;
-        } catch (Exception e) {
-            if (e.getCause() instanceof ResponseException) {
-                if (((ResponseException) e.getCause()).getResponse().getStatusLine().getStatusCode() == 503) { // TODO list of statuses
-                    if (backoffIter.hasNext()) {
-                        try {
-                            Thread.sleep(backoffIter.next()); // TODO ... no?
-                        } catch (InterruptedException ie) {
-                            fut.completeExceptionally(e); // TODO masking internal errors and just returning original error okay?
+                                                                Iterator<Long> backoffIter,
+                                                                CompletableFuture<Response> futureResult) {
+        CompletableFuture<Response> res = delegate.performRequestAsync(endpointId, node, request, options);
+
+        res.whenComplete((resp, e) -> {
+            if (e != null) {
+                if (e instanceof ResponseException) {
+                    if (((ResponseException) e).getResponse().getStatusLine().getStatusCode() == 503) { // TODO list of statuses, configurable or hardcoded?
+                        if (backoffIter.hasNext()) {
+                            try {
+                                Thread.sleep(backoffIter.next());
+                            } catch (InterruptedException ie) {
+                                // TODO okay with masking IE and just returning original exception?
+                                futureResult.completeExceptionally(e);
+                            }
+                            System.out.println("Retrying");
+                            performRequestAsyncRetry(endpointId, node, request, options, backoffIter,futureResult);
                         }
-                        System.out.println("Retrying");
-                        return performRequestAsyncRetry(endpointId, node, request, options, backoffIter);
                     }
                 }
             }
-            return fut;
-        }
+            else {
+                futureResult.complete(resp);
+            }
+        });
+
+        return futureResult;
     }
 
     @Override