onClose callback for LogWatch

Author: C

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

@@ -18,6 +18,8 @@
 import java.io.Closeable;
 import java.io.InputStream;
 import java.io.OutputStream;
+import java.util.concurrent.CompletionStage;
+
 
 public interface LogWatch extends Closeable {
 
@@ -29,9 +31,21 @@ 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.
    */
   @Override
   void close();
+  
+  
 }

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

@@ -32,7 +32,9 @@
 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;
+import java.util.function.Consumer;
 
 public class LogWatchCallback implements LogWatch, AutoCloseable {
 
@@ -41,9 +43,11 @@ public class LogWatchCallback implements LogWatch, AutoCloseable {
   private final OutputStream out;
   private WritableByteChannel outChannel;
   private volatile InputStream output;
+ 
 
   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) {
@@ -53,18 +57,26 @@ 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();
+   
   }
 
   public LogWatchCallback callAndWait(HttpClient client, URL url) {
@@ -111,7 +123,7 @@ public LogWatchCallback callAndWait(HttpClient client, URL url) {
             if (t != null) {
               onFailure(t);
             } else {
-              cleanUp();
+              cleanUp(null);
             }
           }, serialExecutor));
         }
@@ -131,9 +143,9 @@ public void onFailure(Throwable u) {
     if (closed.get()) {
       return;
     }
-
+    
     LOGGER.error("Log Callback Failure.", u);
-    cleanUp();
+    cleanUp(u);
   }
 
 }