Move exit value parsing to message handling for channel 3

Author: rjeberhard

examples/src/main/java/io/kubernetes/client/examples/ExecExample.java

@@ -47,7 +47,7 @@ public static void main(String[] args) throws IOException, ApiException, Interru
     Configuration.setDefaultApiClient(client);
 
     Exec exec = new Exec();
-    boolean tty = true; // Optional: System.console() != null;
+    boolean tty = System.console() != null;
     // final Process proc = exec.exec("default", "nginx-4217019353-k5sn9", new String[]
     //   {"sh", "-c", "echo foo"}, true, tty);
     final Process proc =
@@ -93,6 +93,6 @@ public void run() {
 
     proc.destroy();
 
-    System.exit(0);
+    System.exit(proc.exitValue());
   }
 }

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

@@ -12,6 +12,7 @@
 */
 package io.kubernetes.client;
 
+import com.google.common.io.CharStreams;
 import com.google.gson.reflect.TypeToken;
 import io.kubernetes.client.models.V1Pod;
 import io.kubernetes.client.models.V1Status;
@@ -21,7 +22,9 @@
 import io.kubernetes.client.util.WebSockets;
 import java.io.IOException;
 import java.io.InputStream;
+import java.io.InputStreamReader;
 import java.io.OutputStream;
+import java.io.Reader;
 import java.lang.reflect.Type;
 import java.util.HashMap;
 import java.util.List;
@@ -192,16 +195,12 @@ public Process exec(
 
   static int parseExitCode(ApiClient client, InputStream inputStream) {
     try {
-      int available = inputStream.available();
-
-      // Kubernetes returns no content when the exit code is 0
-      if (available == 0) return 0;
-
-      byte[] b = new byte[available];
-      inputStream.read(b);
-      String body = new String(b, "UTF-8");
-
       Type returnType = new TypeToken<V1Status>() {}.getType();
+      String body;
+      try (final Reader reader = new InputStreamReader(inputStream)) {
+        body = CharStreams.toString(reader);
+      }
+
       V1Status status = client.getJSON().deserialize(body, returnType);
       if ("Success".equals(status.getStatus())) return 0;
 
@@ -232,21 +231,34 @@ static int parseExitCode(ApiClient client, InputStream inputStream) {
 
   private class ExecProcess extends Process {
     private final WebSocketStreamHandler streamHandler;
-    private volatile int statusCode;
+    private int statusCode = -1;
+    private boolean isAlive = true;
     private final Map<Integer, InputStream> input = new HashMap<>();
 
     public ExecProcess() throws IOException {
-      this.statusCode = -1;
       this.streamHandler =
           new WebSocketStreamHandler() {
             @Override
-            public void close() {
-              if (statusCode == -1) {
-                int exitCode = parseExitCode(apiClient, ExecProcess.this.getInputStream(3));
+            protected void handleMessage(int stream, InputStream inStream) {
+              if (stream == 3) {
+                int exitCode = parseExitCode(apiClient, inStream);
+                if (exitCode >= 0) {
+                  // notify of process completion
+                  synchronized (ExecProcess.this) {
+                    statusCode = exitCode;
+                    isAlive = false;
+                    ExecProcess.this.notifyAll();
+                  }
+                }
+              } else super.handleMessage(stream, inStream);
+            }
 
-                // notify of process completion
-                synchronized (ExecProcess.this) {
-                  statusCode = exitCode;
+            @Override
+            public void close() {
+              // notify of process completion
+              synchronized (ExecProcess.this) {
+                if (isAlive) {
+                  isAlive = false;
                   ExecProcess.this.notifyAll();
                 }
               }
@@ -286,24 +298,29 @@ private synchronized InputStream getInputStream(int stream) {
     public int waitFor() throws InterruptedException {
       synchronized (this) {
         this.wait();
+        return statusCode;
       }
-      return statusCode;
     }
 
     @Override
     public boolean waitFor(long timeout, TimeUnit unit) throws InterruptedException {
       synchronized (this) {
         this.wait(TimeUnit.MILLISECONDS.convert(timeout, unit));
+        return !isAlive();
       }
-      return !isAlive();
     }
 
     @Override
-    public int exitValue() {
-      if (statusCode == -1) throw new IllegalThreadStateException();
+    public synchronized int exitValue() {
+      if (isAlive) throw new IllegalThreadStateException();
       return statusCode;
     }
 
+    @Override
+    public synchronized boolean isAlive() {
+      return isAlive;
+    }
+
     @Override
     public void destroy() {
       streamHandler.close();

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

@@ -67,25 +67,31 @@ public synchronized void open(String protocol, WebSocket socket) {
   @Override
   public void bytesMessage(InputStream in) {
     try {
-      OutputStream out = getSocketInputOutputStream(in.read());
-      ByteStreams.copy(in, out);
+      handleMessage(in.read(), in);
     } catch (IOException ex) {
-      log.error("Error writing message", ex);
+      log.error("Error reading message channel", ex);
     }
   }
 
   @Override
   public void textMessage(Reader in) {
     try {
-      OutputStream out = getSocketInputOutputStream(in.read());
-      InputStream inStream =
-          new ByteArrayInputStream(CharStreams.toString(in).getBytes(Charsets.UTF_8));
-      ByteStreams.copy(inStream, out);
+      handleMessage(
+          in.read(), new ByteArrayInputStream(CharStreams.toString(in).getBytes(Charsets.UTF_8)));
     } catch (IOException ex) {
       log.error("Error writing message", ex);
     }
   }
 
+  protected void handleMessage(int stream, InputStream inStream) {
+    try {
+      OutputStream out = getSocketInputOutputStream(stream);
+      ByteStreams.copy(inStream, out);
+    } catch (IOException ex) {
+      log.error("Error handling message", ex);
+    }
+  }
+
   @Override
   public synchronized void close() {
     if (state != State.CLOSED) {