Add Async Copy methods.

Author: brendanburns

util/src/main/java/io/kubernetes/client/Copy.java

@@ -30,6 +30,9 @@
 import java.io.OutputStream;
 import java.nio.file.Path;
 import java.nio.file.Paths;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
 import org.apache.commons.codec.binary.Base64InputStream;
 import org.apache.commons.codec.binary.Base64OutputStream;
 import org.apache.commons.compress.archivers.ArchiveEntry;
@@ -153,6 +156,21 @@ public void copyDirectoryFromPod(
       createDirectoryStructureFromTree(tree, namespace, pod, container, srcPath, destination);
       return;
     }
+    Future<Integer> future =
+        copyDirectoryFromPodAsync(namespace, pod, container, srcPath, destination);
+    try {
+      int code = future.get().intValue();
+      if (code != 0) {
+        throw new IOException("Copy failed (" + code + ")");
+      }
+    } catch (InterruptedException | ExecutionException ex) {
+      throw new IOException(ex);
+    }
+  }
+
+  public Future<Integer> copyDirectoryFromPodAsync(
+      String namespace, String pod, String container, String srcPath, Path destination)
+      throws IOException, ApiException {
     final Process proc =
         this.exec(
             namespace,
@@ -187,14 +205,7 @@ public void copyDirectoryFromPod(
         }
       }
     }
-    try {
-      int status = proc.waitFor();
-      if (status != 0) {
-        throw new IOException("Copy command failed with status " + status);
-      }
-    } catch (InterruptedException ex) {
-      throw new IOException(ex);
-    }
+    return new ProcessFuture(proc);
   }
 
   // This creates directories and files using tree of files and directories under container
@@ -330,6 +341,19 @@ public void copyFileToPod(
       String namespace, String pod, String container, Path srcPath, Path destPath)
       throws ApiException, IOException {
 
+    try {
+      int exit = copyFileToPodAsync(namespace, pod, container, srcPath, destPath).get();
+      if (exit != 0) {
+        throw new IOException("Failed to copy: " + exit);
+      }
+    } catch (InterruptedException | ExecutionException ex) {
+      throw new IOException(ex);
+    }
+  }
+
+  public Future<Integer> copyFileToPodAsync(
+      String namespace, String pod, String container, Path srcPath, Path destPath)
+      throws ApiException, IOException {
     // Run decoding and extracting processes
     final Process proc = execCopyToPod(namespace, pod, container, destPath);
 
@@ -344,15 +368,27 @@ public void copyFileToPod(
       archiveOutputStream.putArchiveEntry(tarEntry);
       ByteStreams.copy(input, archiveOutputStream);
       archiveOutputStream.closeArchiveEntry();
-    } finally {
-      proc.destroy();
+
+      return new ProcessFuture(proc);
     }
   }
 
   public void copyFileToPod(
       String namespace, String pod, String container, byte[] src, Path destPath)
       throws ApiException, IOException {
+    try {
+      int exit = copyFileToPodAsync(namespace, pod, container, src, destPath).get();
+      if (exit != 0) {
+        throw new IOException("Copy failed: " + exit);
+      }
+    } catch (InterruptedException | ExecutionException ex) {
+      throw new IOException(ex);
+    }
+  }
 
+  public Future<Integer> copyFileToPodAsync(
+      String namespace, String pod, String container, byte[] src, Path destPath)
+      throws ApiException, IOException {
     // Run decoding and extracting processes
     final Process proc = execCopyToPod(namespace, pod, container, destPath);
 
@@ -365,8 +401,8 @@ public void copyFileToPod(
       archiveOutputStream.putArchiveEntry(tarEntry);
       ByteStreams.copy(new ByteArrayInputStream(src), archiveOutputStream);
       archiveOutputStream.closeArchiveEntry();
-    } finally {
-      proc.destroy();
+
+      return new ProcessFuture(proc);
     }
   }
 
@@ -397,4 +433,37 @@ private boolean isTarPresentInContainer(String namespace, String pod, String con
       proc.destroy();
     }
   }
+
+  private static class ProcessFuture implements Future<Integer> {
+    private Process proc;
+
+    ProcessFuture(Process proc) {
+      this.proc = proc;
+    }
+
+    // TODO: support cancelling?
+    public boolean cancel(boolean interupt) {
+      return false;
+    }
+
+    public boolean isCancelled() {
+      return false;
+    }
+
+    public Integer get() throws InterruptedException {
+      proc.waitFor();
+      proc.destroy();
+      return proc.exitValue();
+    }
+
+    public Integer get(long timeout, TimeUnit unit) throws InterruptedException {
+      proc.waitFor(timeout, unit);
+      proc.destroy();
+      return proc.exitValue();
+    }
+
+    public boolean isDone() {
+      return proc.isAlive();
+    }
+  }
 }