onClose callback for LogWatch

CHANGELOG.md

@@ -9,6 +9,7 @@
 #### Improvements
 * Fix #6863: ensuring SerialExecutor does not throw RejectedExecutionException to prevent unnecessary error logs
 * Fix #6763: (crd-generator) YAML output customization
+* Fix #6880: LogWatch interface provides listeners on close stream event 
 
 #### Dependency Upgrade
 

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);
   }
 
 }

kubernetes-client/src/test/java/io/fabric8/kubernetes/client/dsl/internal/LogWatchCallbackTest.java

@@ -0,0 +1,118 @@
+/*
+ * Copyright (C) 2015 Red Hat, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *         http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.fabric8.kubernetes.client.dsl.internal;
+
+import io.fabric8.kubernetes.client.http.AsyncBody;
+import io.fabric8.kubernetes.client.http.HttpClient;
+import io.fabric8.kubernetes.client.http.HttpRequest;
+import io.fabric8.kubernetes.client.http.HttpResponse;
+import io.fabric8.kubernetes.client.http.TestAsyncBody;
+import io.fabric8.kubernetes.client.http.TestHttpResponse;
+import io.fabric8.kubernetes.client.impl.BaseClient;
+import io.fabric8.kubernetes.client.utils.KubernetesSerialization;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+import java.io.ByteArrayOutputStream;
+import java.net.HttpURLConnection;
+import java.net.MalformedURLException;
+import java.net.URL;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executor;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.when;
+
+class LogWatchCallbackTest {
+  private OperationContext context;
+  private Executor executor = Executors.newFixedThreadPool(2);
+  private URL url;
+  private HttpClient httpClientMock;
+
+  @BeforeEach
+  void setUp() throws MalformedURLException {
+    BaseClient mock = mock(BaseClient.class, Mockito.RETURNS_SELF);
+    Mockito.when(mock.adapt(BaseClient.class).getKubernetesSerialization()).thenReturn(new KubernetesSerialization());
+    final OperationContext operationContext = new OperationContext().withClient(mock);
+    when(mock.getExecutor()).thenReturn(this.executor);
+    this.context = operationContext;
+
+    this.url = new URL("http://url_called");
+    this.httpClientMock = spy(HttpClient.class);
+    var httpRequestMock = mock(HttpRequest.class);
+    var builderMock = mock(HttpRequest.Builder.class);
+
+    Mockito.when(httpClientMock.newHttpRequestBuilder()).thenReturn(builderMock);
+    Mockito.when(builderMock.url(url)).thenReturn(builderMock);
+    Mockito.when(builderMock.build()).thenReturn(httpRequestMock);
+
+  }
+
+  @Test
+  void withOutputStreamCloseEventTest() throws InterruptedException {
+
+    var future = new CompletableFuture<HttpResponse<AsyncBody>>();
+    var reached = new CountDownLatch(1);
+
+    Mockito.when(httpClientMock.consumeBytes(Mockito.any(), Mockito.any())).thenReturn(future);
+
+    ByteArrayOutputStream baos = new ByteArrayOutputStream();
+    LogWatchCallback logWatch = new LogWatchCallback(baos, this.context);
+    logWatch.callAndWait(httpClientMock, url);
+
+    logWatch.onClose().thenAccept((Throwable t) -> {
+      reached.countDown();
+    });
+    future.complete(
+        new TestHttpResponse<AsyncBody>().withCode(HttpURLConnection.HTTP_GONE).withBody(new TestAsyncBody()));
+
+    assertThat(reached.await(1, TimeUnit.SECONDS)).isTrue();
+    logWatch.close();
+  }
+
+  @Test
+  void withOutputStreamCloseEventOnFailureTest() throws InterruptedException {
+
+    var future = new CompletableFuture<HttpResponse<AsyncBody>>();
+    var reached = new CountDownLatch(1);
+
+    Mockito.when(httpClientMock.consumeBytes(Mockito.any(), Mockito.any())).thenReturn(future);
+
+    LogWatchCallback logWatch = new LogWatchCallback(new ByteArrayOutputStream(), this.context);
+    logWatch.callAndWait(httpClientMock, url);
+
+    final Throwable[] tReturned = new Throwable[1];
+    logWatch.onClose().thenAccept((Throwable t) -> {
+      tReturned[0] = t;
+      reached.countDown();
+    });
+
+    var th = new Throwable("any exception");
+    future.completeExceptionally(th);
+
+    assertThat(reached.await(1, TimeUnit.SECONDS)).isTrue();
+    assertThat(tReturned[0]).isEqualTo(th);
+
+    logWatch.close();
+  }
+}