onClose callback for LogWatch

Author: C

kubernetes-client-api/src/main/java/io/fabric8/kubernetes/client/dsl/LogWatch.java

@@ -18,6 +18,7 @@
 import java.io.Closeable;
 import java.io.InputStream;
 import java.io.OutputStream;
+import java.util.concurrent.CompletionStage;
 
 public interface LogWatch extends Closeable {
 
@@ -29,6 +30,15 @@ public interface LogWatch extends Closeable {
    */
   InputStream getOutput();
 
+  /**
+   * Returns a {@link CompletionStage} released when the log stream is closed.
+   * If the stream is closed due to an exception (cf onFailure),
+   * this exception will be passed as parameter, null otherwise
+   *
+   * @return a {@link CompletionStage} released when the log stream is closed
+   */
+  CompletionStage<Throwable> onClose();
+
   /**
    * Close the Watch.
    */

kubernetes-client/src/main/java/io/fabric8/kubernetes/client/dsl/internal/LogWatchCallback.java

@@ -32,6 +32,7 @@
 import java.nio.channels.Channels;
 import java.nio.channels.WritableByteChannel;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
 import java.util.concurrent.atomic.AtomicBoolean;
 
 public class LogWatchCallback implements LogWatch, AutoCloseable {
@@ -44,6 +45,7 @@ public class LogWatchCallback implements LogWatch, AutoCloseable {
 
   private final AtomicBoolean closed = new AtomicBoolean(false);
   private final CompletableFuture<AsyncBody> asyncBody = new CompletableFuture<>();
+  private final CompletableFuture<Throwable> onCloseFuture = new CompletableFuture<>();
   private final SerialExecutor serialExecutor;
 
   public LogWatchCallback(OutputStream out, OperationContext context) {
@@ -54,16 +56,22 @@ public LogWatchCallback(OutputStream out, OperationContext context) {
     this.serialExecutor = new SerialExecutor(context.getExecutor());
   }
 
+  @Override
+  public CompletionStage<Throwable> onClose() {
+    return onCloseFuture.minimalCompletionStage();
+  }
+
   @Override
   public void close() {
-    cleanUp();
+    cleanUp(null);
   }
 
-  private void cleanUp() {
+  private void cleanUp(Throwable u) {
     if (!closed.compareAndSet(false, true)) {
       return;
     }
     asyncBody.thenAccept(AsyncBody::cancel);
+    onCloseFuture.complete(u);
     serialExecutor.shutdownNow();
   }
 
@@ -111,7 +119,7 @@ public LogWatchCallback callAndWait(HttpClient client, URL url) {
             if (t != null) {
               onFailure(t);
             } else {
-              cleanUp();
+              cleanUp(null);
             }
           }, serialExecutor));
         }
@@ -133,7 +141,7 @@ public void onFailure(Throwable u) {
     }
 
     LOGGER.error("Log Callback Failure.", u);
-    cleanUp();
+    cleanUp(u);
   }
 
 }