Use CompletableFuture in JarCache (#715)

src/main/java/hudson/remoting/JarCache.java

@@ -5,6 +5,7 @@
 import java.io.File;
 import java.io.IOException;
 import java.net.URL;
+import java.util.concurrent.CompletableFuture;
 
 /**
  * Jar file cache.
@@ -55,5 +56,5 @@ public abstract class JarCache {
      *      URL of the jar file.
      */
     @NonNull
-    public abstract Future<URL> resolve(@NonNull Channel channel, long sum1, long sum2) throws IOException, InterruptedException;
+    public abstract CompletableFuture<URL> resolve(@NonNull Channel channel, long sum1, long sum2) throws IOException, InterruptedException;
 }

src/main/java/hudson/remoting/JarCacheSupport.java

@@ -2,6 +2,7 @@
 
 import java.io.IOException;
 import java.net.URL;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.ExecutorService;
@@ -21,7 +22,7 @@ public abstract class JarCacheSupport extends JarCache {
     /**
      * Remember in-progress jar file resolution to avoid retrieving the same jar file twice.
      */
-    private final ConcurrentMap<Checksum,Future<URL>> inprogress = new ConcurrentHashMap<>();
+    private final ConcurrentMap<Checksum,CompletableFuture<URL>> inprogress = new ConcurrentHashMap<>();
 
     /**
      * Look up the local cache and return URL if found.
@@ -45,11 +46,11 @@ public abstract class JarCacheSupport extends JarCache {
 
     @Override
     @NonNull
-    public Future<URL> resolve(@NonNull final Channel channel, final long sum1, final long sum2) throws IOException, InterruptedException {
+    public CompletableFuture<URL> resolve(@NonNull final Channel channel, final long sum1, final long sum2) throws IOException, InterruptedException {
         URL jar = lookInCache(channel,sum1, sum2);
         if (jar!=null) {
             // already in the cache
-            return new AsyncFutureImpl<>(jar);
+            return CompletableFuture.completedFuture(jar);
         }
 
         final Checksum key = new Checksum(sum1, sum2);
@@ -58,8 +59,8 @@ public Future<URL> resolve(@NonNull final Channel channel, final long sum1, fina
 
     @NonNull
     @SuppressFBWarnings(value = "RV_RETURN_VALUE_IGNORED_BAD_PRACTICE", justification = "API compatibility")
-    private Future<URL> submitDownload(Channel channel, long sum1, long sum2, Checksum key) {
-        final AsyncFutureImpl<URL> promise = new AsyncFutureImpl<>();
+    private CompletableFuture<URL> submitDownload(Channel channel, long sum1, long sum2, Checksum key) {
+        final CompletableFuture<URL> promise = new CompletableFuture<>();
         downloader.submit(new DownloadRunnable(channel, sum1, sum2, key, promise));
         return promise;
     }
@@ -70,9 +71,9 @@ private class DownloadRunnable implements Runnable {
         final long sum1;
         final long sum2;
         final Checksum key;
-        final AsyncFutureImpl<URL> promise;
+        final CompletableFuture<URL> promise;
 
-        public DownloadRunnable(Channel channel, long sum1, long sum2, Checksum key, AsyncFutureImpl<URL> promise) {
+        public DownloadRunnable(Channel channel, long sum1, long sum2, Checksum key, CompletableFuture<URL> promise) {
             this.channel = channel;
             this.sum1 = sum1;
             this.sum2 = sum2;
@@ -85,7 +86,7 @@ public void run() {
             try {
                 URL url = retrieve(channel, sum1, sum2);
                 inprogress.remove(key);
-                promise.set(url);
+                promise.complete(url);
             } catch (ChannelClosedException | RequestAbortedException e) {
                 // the connection was killed while we were still resolving the file
                 bailout(e);
@@ -101,7 +102,7 @@ public void run() {
                 } else {
                     // in other general failures, we aren't retrying
                     // TODO: or should we?
-                    promise.set(e);
+                    promise.completeExceptionally(e);
                     LOGGER.log(Level.WARNING, String.format("Failed to resolve a jar %016x%016x", sum1, sum2), e);
                 }
             }
@@ -112,7 +113,7 @@ public void run() {
          */
         private void bailout(Throwable e) {
             inprogress.remove(key);     // this lets another thread to retry later
-            promise.set(e);             // then tell those who are waiting that we aborted
+            promise.completeExceptionally(e);             // then tell those who are waiting that we aborted
         }
     }
 

src/main/java/hudson/remoting/RemoteClassLoader.java

@@ -47,6 +47,7 @@
 import java.util.Map;
 import java.util.Set;
 import java.util.Vector;
+import java.util.concurrent.Future;
 import java.util.concurrent.ExecutionException;
 import java.util.logging.Level;
 import java.util.logging.Logger;

src/main/java/hudson/remoting/ResourceImageBoth.java

@@ -4,6 +4,7 @@
 
 import java.io.IOException;
 import java.net.URL;
+import java.util.concurrent.Future;
 import java.util.logging.Level;
 import java.util.logging.Logger;
 

src/main/java/hudson/remoting/ResourceImageDirect.java

@@ -4,6 +4,8 @@
 
 import java.io.IOException;
 import java.net.URL;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Future;
 import java.util.logging.Level;
 import java.util.logging.Logger;
 
@@ -35,12 +37,12 @@ class ResourceImageDirect extends ResourceImageRef {
     @Override
     Future<byte[]> resolve(Channel channel, String resourcePath) throws IOException, InterruptedException {
         LOGGER.log(Level.FINE, resourcePath+" image is direct");
-        return new AsyncFutureImpl<>(payload);
+        return CompletableFuture.completedFuture(payload);
     }
 
     @Override
     Future<URLish> resolveURL(Channel channel, String resourcePath) throws IOException, InterruptedException {
-        return new AsyncFutureImpl<>(URLish.from(makeResource(resourcePath, payload)));
+        return CompletableFuture.completedFuture(URLish.from(makeResource(resourcePath, payload)));
     }
 
     private static final Logger LOGGER = Logger.getLogger(ResourceImageDirect.class.getName());

src/main/java/hudson/remoting/ResourceImageInJar.java

@@ -4,9 +4,12 @@
 import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
 
 import java.io.IOException;
+import java.io.InputStream;
+import java.io.UncheckedIOException;
 import java.net.URL;
 import java.net.URLClassLoader;
-import java.util.concurrent.ExecutionException;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Future;
 
 /**
  * {@link ResourceImageRef} that points to a resource inside a jar file.
@@ -42,31 +45,29 @@ class ResourceImageInJar extends ResourceImageRef {
 
     @Override
     Future<byte[]> resolve(Channel channel, final String resourcePath) throws IOException, InterruptedException {
-        return new FutureAdapter<>(_resolveJarURL(channel)) {
-            @Override
-            @SuppressFBWarnings(value = "URLCONNECTION_SSRF_FD", justification = "This is only used for managing the jar cache as files.")
-            protected byte[] adapt(URL jar) throws ExecutionException {
-                try {
-                    return Util.readFully(toResourceURL(jar, resourcePath).openStream());
-                } catch (IOException e) {
-                    throw new ExecutionException(e);
-                }
-            }
-        };
+        return _resolveJarURL(channel).thenApply(jar -> readContents(jar, resourcePath));
+    }
+
+    @SuppressFBWarnings(value = "URLCONNECTION_SSRF_FD", justification = "This is only used for managing the jar cache as files.")
+    private byte[] readContents(URL jar, String resourcePath) {
+        try (InputStream in = toResourceURL(jar, resourcePath).openStream()) {
+            return Util.readFully(in);
+        } catch (IOException e) {
+            throw new UncheckedIOException(e);
+        }
     }
 
     @Override
     Future<URLish> resolveURL(Channel channel, final String resourcePath) throws IOException, InterruptedException {
-        return new FutureAdapter<>(_resolveJarURL(channel)) {
-            @Override
-            protected URLish adapt(URL jar) throws ExecutionException {
-                try {
-                    return URLish.from(toResourceURL(jar, resourcePath));
-                } catch (IOException e) {
-                    throw new ExecutionException(e);
-                }
-            }
-        };
+        return _resolveJarURL(channel).thenApply(jar -> getUrlish(jar, resourcePath));
+    }
+
+    private URLish getUrlish(URL jar, String resourcePath) {
+        try {
+            return URLish.from(toResourceURL(jar, resourcePath));
+        } catch (IOException e) {
+            throw new UncheckedIOException(e);
+        }
     }
 
     @NonNull
@@ -83,7 +84,7 @@ open jar file (see sun.net.www.protocol.jar.JarURLConnection.JarURLInputStream.c
         return new URL("jar:"+ jar +"!/"+resourcePath);
     }
 
-    Future<URL> _resolveJarURL(Channel channel) throws IOException, InterruptedException {
+    CompletableFuture<URL> _resolveJarURL(Channel channel) throws IOException, InterruptedException {
         JarCache c = channel.getJarCache();
         if (c == null) {
             throw new IOException(String.format("Failed to resolve a jar %016x%016x. JAR Cache is disabled for the channel %s",

src/main/java/hudson/remoting/ResourceImageRef.java

@@ -4,6 +4,7 @@
 
 import java.io.IOException;
 import java.io.Serializable;
+import java.util.concurrent.Future;
 
 /**
  * Wire protocol data representation that encapsulates the access to a resource inside a {@link ClassLoader}.