Improved handling of stream closure.

Making interrupt of `build -s` work in `-http` mode.
Does not solve `ReadPendingException`s.

cli/src/main/java/hudson/cli/CLI.java

@@ -728,8 +728,12 @@ public class CLI implements AutoCloseable {
             protected void onStderr(byte[] chunk) throws IOException {
                 System.err.write(chunk);
             }
+            @Override
+            protected synchronized void handleClose() {
+                notifyAll();
             }
-        final ClientSideImpl connection = new ClientSideImpl(streams.getInputStream(), streams.getOutputStream());
+        }
+        try (final ClientSideImpl connection = new ClientSideImpl(streams.getInputStream(), streams.getOutputStream())) {
             for (String arg : args) {
                 connection.sendArg(arg);
             }
@@ -757,6 +761,7 @@ public class CLI implements AutoCloseable {
             }
             return connection.exit;
         }
+    }
 
     private static String computeVersion() {
         Properties props = new Properties();

cli/src/main/java/hudson/cli/PlainCLIProtocol.java

@@ -25,12 +25,14 @@
 package hudson.cli;
 
 import java.io.ByteArrayOutputStream;
+import java.io.Closeable;
 import java.io.DataInputStream;
 import java.io.DataOutputStream;
 import java.io.EOFException;
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
+import java.nio.channels.ReadPendingException;
 import java.util.logging.Level;
 import java.util.logging.Logger;
 import org.apache.commons.io.IOUtils;
@@ -68,7 +70,7 @@ class PlainCLIProtocol {
         STDERR
     }
 
-    static abstract class EitherSide {
+    static abstract class EitherSide implements Closeable {
 
         private final CountingInputStream cis;
         private final FlightRecorderInputStream flightRecorder;
@@ -83,7 +85,15 @@ class PlainCLIProtocol {
         }
 
         final void begin() {
-            new Thread("PlainCLIProtocol") { // TODO set distinctive Thread.name
+            new Reader().start();
+        }
+
+        private class Reader extends Thread {
+
+            Reader() {
+                super("PlainCLIProtocol"); // TODO set distinctive Thread.name
+            }
+
             @Override
             public void run() {
                 try {
@@ -93,7 +103,14 @@ class PlainCLIProtocol {
                         try {
                             framelen = dis.readInt();
                         } catch (EOFException x) {
+                            handleClose();
                             break; // TODO verify that we hit EOF immediately, not partway into framelen
+                        } catch (ReadPendingException x) {
+                            LOGGER.log(Level.FINE, null, x);
+                            // TODO what does this signify? Seems to be thrown randomly by org.eclipse.jetty.io.FillInterest.register. No obvious impact.
+                            // Check https://github.com/eclipse/jetty.project/issues/1047 in 9.4.3.v20170317 but cf. https://github.com/joakime/jetty-async-bug/issues/1
+                            handleClose();
+                            break;
                         }
                         if (framelen < 0) {
                             throw new IOException("corrupt stream: negative frame length");
@@ -123,11 +140,16 @@ class PlainCLIProtocol {
                     }
                 } catch (IOException x) {
                     LOGGER.log(Level.WARNING, null, flightRecorder.analyzeCrash(x, "broken stream"));
+                } catch (ReadPendingException x) { // as above
+                    LOGGER.log(Level.FINE, null, x);
+                    handleClose();
                 }
             }
-            }.start();
+
         }
 
+        protected abstract void handleClose();
+
         protected abstract boolean handle(Op op, int framelen) throws IOException;
 
         private void writeOp(Op op) throws IOException {
@@ -165,6 +187,7 @@ class PlainCLIProtocol {
         }
 
         protected final byte[] readChunk(int framelen) throws IOException {
+            assert Thread.currentThread() instanceof EitherSide.Reader;
             byte[] buf = new byte[framelen];
             dis.readFully(buf);
             return buf;
@@ -187,6 +210,11 @@ class PlainCLIProtocol {
             };
         }
 
+        @Override
+        public synchronized void close() throws IOException {
+            dos.close();
+        }
+
     }
 
     static abstract class ServerSide extends EitherSide {
@@ -197,6 +225,7 @@ class PlainCLIProtocol {
 
         @Override
         protected final boolean handle(Op op, int framelen) throws IOException {
+            assert Thread.currentThread() instanceof EitherSide.Reader;
             switch (op) {
             case ARG:
                 onArg(dis.readUTF());
@@ -255,6 +284,7 @@ class PlainCLIProtocol {
 
         @Override
         protected boolean handle(Op op, int framelen) throws IOException {
+            assert Thread.currentThread() instanceof EitherSide.Reader;
             switch (op) {
             case EXIT:
                 onExit(dis.readInt());

core/src/main/java/hudson/cli/CLIAction.java

@@ -55,6 +55,7 @@ import java.nio.charset.UnsupportedCharsetException;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Locale;
+import java.util.concurrent.atomic.AtomicReference;
 import java.util.logging.Level;
 import java.util.logging.Logger;
 import jenkins.util.FullDuplexHttpService;
@@ -132,6 +133,7 @@ public class CLIAction implements UnprotectedRootAction, StaplerProxy {
             return new FullDuplexHttpService(uuid) {
                 @Override
                 protected void run(InputStream upload, OutputStream download) throws IOException, InterruptedException {
+                    final AtomicReference<Thread> runningThread = new AtomicReference<>();
                     class ServerSideImpl extends PlainCLIProtocol.ServerSide {
                         List<String> args = new ArrayList<>();
                         Locale locale = Locale.getDefault();
@@ -176,14 +178,27 @@ public class CLIAction implements UnprotectedRootAction, StaplerProxy {
                         protected void onEndStdin() throws IOException {
                             stdinMatch.close();
                         }
+                        @Override
+                        protected synchronized void handleClose() {
+                            notifyAll();
+                            Thread t = runningThread.get();
+                            if (t != null) {
+                                t.interrupt();
+                            }
+                        }
                     }
-                    ServerSideImpl connection = new ServerSideImpl(upload, download);
+                    try (ServerSideImpl connection = new ServerSideImpl(upload, download)) {
                         connection.begin();
                         synchronized (connection) {
-                        connection.wait(); // TODO this can wait indefinitely even when the connection is broken
+                            connection.wait();
                         }
                         PrintStream stdout = new PrintStream(connection.streamStdout(), false, connection.encoding.name());
                         PrintStream stderr = new PrintStream(connection.streamStderr(), true, connection.encoding.name());
+                        if (connection.args.isEmpty()) {
+                            stderr.println("Connection closed before arguments received");
+                            connection.sendExit(2);
+                            return;
+                        }
                         String commandName = connection.args.get(0);
                         CLICommand command = CLICommand.clone(commandName);
                         if (command == null) {
@@ -195,11 +210,14 @@ public class CLIAction implements UnprotectedRootAction, StaplerProxy {
                         command.setClientCharset(connection.encoding);
                         CLICommand orig = CLICommand.setCurrent(command);
                         try {
+                            runningThread.set(Thread.currentThread());
                             int exit = command.main(connection.args.subList(1, connection.args.size()), connection.locale, connection.stdin, stdout, stderr);
                             stdout.flush();
                             connection.sendExit(exit);
                         } finally {
                             CLICommand.setCurrent(orig);
+                            runningThread.set(null);
+                        }
                     }
                 }
             };

test/src/test/java/hudson/cli/CLITest.java

@@ -120,9 +120,7 @@ public class CLITest {
         p.getBuildersList().add(new SleepBuilder(TimeUnit.MINUTES.toMillis(5)));
         doInterrupt(jar, p, "-remoting", "-i", privkey.getAbsolutePath());
         doInterrupt(jar, p, "-ssh", "-user", "admin", "-i", privkey.getAbsolutePath());
-        /* TODO does not yet work in HTTP mode:
         doInterrupt(jar, p, "-http", "-auth", "admin:admin");
-        */
     }
     private void doInterrupt(File jar, FreeStyleProject p, String... modeArgs) throws Exception {
         ByteArrayOutputStream baos = new ByteArrayOutputStream();