[JEP-222] WebSocket-based agents

pom.xml

@@ -94,6 +94,11 @@ THE SOFTWARE.
       <artifactId>animal-sniffer-annotations</artifactId>
       <scope>provided</scope>
     </dependency>
+    <dependency>
+      <groupId>org.glassfish.tyrus.bundles</groupId>
+      <artifactId>tyrus-standalone-client-jdk</artifactId>
+      <version>1.12</version>
+    </dependency>
     <!-- test dependencies -->
     <dependency>
       <groupId>junit</groupId>
@@ -343,6 +348,17 @@ THE SOFTWARE.
               <includeGroupIds>org.jenkins-ci</includeGroupIds>
             </configuration>
           </execution>
+          <execution>
+            <id>bundle-tyrus</id>
+            <phase>process-classes</phase>
+            <goals>
+              <goal>unpack-dependencies</goal>
+            </goals>
+            <configuration>
+              <outputDirectory>${project.build.outputDirectory}</outputDirectory>
+              <includeArtifactIds>tyrus-standalone-client-jdk</includeArtifactIds>
+            </configuration>
+          </execution>
         </executions>
       </plugin>
       <plugin>

src/findbugs/excludeFilter.xml

@@ -6,6 +6,12 @@
   <Match>
     <Package name="~org\.jenkinsci\.constant_pool_scanner.*"/>
   </Match>
+  <Match>
+    <Package name="~javax\.websocket.*"/>
+  </Match>
+  <Match>
+    <Package name="~org\.glassfish\.tyrus.*"/>
+  </Match>
 
   <Match>
     <!--We do not want do break API by converting potentially usefull classes-->

src/main/java/hudson/remoting/Capability.java

@@ -120,10 +120,17 @@ public boolean supportsProxyExceptionFallback() {
 
     //TODO: ideally preamble handling needs to be reworked in order to avoid FB suppression
     /**
-     * Writes out the capacity preamble.
+     * Writes {@link #PREAMBLE} then uses {@link #write}.
      */
     void writePreamble(OutputStream os) throws IOException {
         os.write(PREAMBLE);
+        write(os);
+    }
+
+    /**
+     * Writes this capability to a stream.
+     */
+    public void write(OutputStream os) throws IOException {
         try (ObjectOutputStream oos = new ObjectOutputStream(Mode.TEXT.wrap(os)) {
             @Override
             public void close() throws IOException {
@@ -143,7 +150,7 @@ protected void annotateClass(Class<?> c) throws IOException {
     }
 
     /**
-     * The opposite operation of {@link #writePreamble(OutputStream)}.
+     * The opposite operation of {@link #write}.
      */
     public static Capability read(InputStream is) throws IOException {
         try (ObjectInputStream ois = new ObjectInputStream(Mode.TEXT.wrap(is)) {

src/main/java/hudson/remoting/Engine.java

@@ -23,13 +23,19 @@
  */
 package hudson.remoting;
 
+import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
 import hudson.remoting.Channel.Mode;
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
 import java.io.File;
 import java.io.FileInputStream;
 import java.io.FileNotFoundException;
 import java.io.IOException;
 import java.net.Socket;
+import java.net.URI;
 import java.net.URL;
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
 import java.nio.file.Path;
 import java.security.AccessController;
 import java.security.KeyManagementException;
@@ -44,13 +50,15 @@
 import java.security.cert.X509Certificate;
 import java.security.interfaces.RSAPublicKey;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.atomic.AtomicReference;
 import java.util.logging.Level;
 import java.util.logging.Logger;
 import javax.annotation.CheckForNull;
@@ -61,6 +69,13 @@
 import javax.net.ssl.SSLSocketFactory;
 import javax.net.ssl.TrustManager;
 import javax.net.ssl.TrustManagerFactory;
+import javax.websocket.ClientEndpointConfig;
+import javax.websocket.CloseReason;
+import javax.websocket.ContainerProvider;
+import javax.websocket.Endpoint;
+import javax.websocket.EndpointConfig;
+import javax.websocket.HandshakeResponse;
+import javax.websocket.Session;
 
 import org.jenkinsci.remoting.engine.JnlpEndpointResolver;
 import org.jenkinsci.remoting.engine.Jnlp4ConnectionState;
@@ -437,10 +452,105 @@ public void removeListener(EngineListener el) {
         events.remove(el);
     }
 
+    @SuppressFBWarnings(value = "REC_CATCH_EXCEPTION", justification = "checked exceptions were a mistake to begin with")
     @Override
     public void run() {
         // Create the engine
         try {
+            if (true) { // TODO conditionally enable WS support
+                AtomicReference<Channel> ch = new AtomicReference<>();
+                String localCap;
+                try (ByteArrayOutputStream baos = new ByteArrayOutputStream()) {
+                    new Capability().write(baos);
+                    localCap = baos.toString("US-ASCII"); // Base-64
+                }
+                class HeaderHandler extends ClientEndpointConfig.Configurator {
+                    Capability remoteCapability = new Capability();
+                    @Override
+                    public void beforeRequest(Map<String, List<String>> headers) {
+                        headers.put(JnlpConnectionState.CLIENT_NAME_KEY, Collections.singletonList(slaveName));
+                        headers.put(JnlpConnectionState.SECRET_KEY, Collections.singletonList(secretKey));
+                        headers.put(Capability.class.getName(), Collections.singletonList(localCap));
+                        // TODO use JnlpConnectionState.COOKIE_KEY somehow
+                        LOGGER.fine(() -> "Sending: " + headers);
+                    }
+                    @Override
+                    public void afterResponse(HandshakeResponse hr) {
+                        LOGGER.fine(() -> "Receiving: " + hr.getHeaders());
+                        try (ByteArrayInputStream bais = new ByteArrayInputStream(hr.getHeaders().get(Capability.class.getName()).get(0).getBytes(StandardCharsets.US_ASCII))) {
+                            remoteCapability = Capability.read(bais);
+                            LOGGER.fine(() -> "received " + remoteCapability);
+                        } catch (IOException x) {
+                            events.error(x);
+                        }
+                    }
+                }
+                HeaderHandler headerHandler = new HeaderHandler();
+                ContainerProvider.getWebSocketContainer().connectToServer(new Endpoint() {
+                    AbstractByteArrayCommandTransport.ByteArrayReceiver receiver;
+                    @Override
+                    public void onOpen(Session session, EndpointConfig config) {
+                        events.status("WebSocket connection open");
+                        session.addMessageHandler(byte[].class, this::onMessage);
+                        try {
+                            ch.set(new ChannelBuilder(slaveName, executor).build(new Transport(session)));
+                        } catch (IOException x) {
+                            events.error(x);
+                        }
+                    }
+                    @SuppressFBWarnings(value = "UWF_FIELD_NOT_INITIALIZED_IN_CONSTRUCTOR", justification = "just trust me here")
+                    private void onMessage(byte[] message) {
+                        LOGGER.finest(() -> "received message of length " + message.length);
+                        receiver.handle(message);
+                    }
+                    @Override
+                    public void onClose(Session session, CloseReason closeReason) {
+                        // TODO
+                    }
+                    @Override
+                    public void onError(Session session, Throwable thr) {
+                        // TODO
+                    }
+                    class Transport extends AbstractByteArrayCommandTransport {
+                        final Session session;
+                        Transport(Session session) {
+                            this.session = session;
+                        }
+                        @Override
+                        public void setup(AbstractByteArrayCommandTransport.ByteArrayReceiver _receiver) {
+                            events.status("Setting up channel");
+                            receiver = _receiver;
+                        }
+                        @Override
+                        public void writeBlock(Channel channel, byte[] payload) throws IOException {
+                            LOGGER.finest(() -> "sending message of length " + payload.length);
+                            session.getBasicRemote().sendBinary(ByteBuffer.wrap(payload));
+                        }
+                        @Override
+                        public Capability getRemoteCapability() throws IOException {
+                            return headerHandler.remoteCapability;
+                        }
+                        @Override
+                        public void closeWrite() throws IOException {
+                            events.status("Write side closed");
+                            // TODO
+                        }
+                        @Override
+                        public void closeRead() throws IOException {
+                            events.status("Read side closed");
+                            // TODO
+                        }
+                    }
+                }, ClientEndpointConfig.Builder.create().configurator(headerHandler).build(), URI.create(candidateUrls.get(0).toString().replaceFirst("^http", "ws") + "wsagents/"));
+                while (ch.get() == null) {
+                    Thread.sleep(100);
+                }
+                LOGGER.info(() -> "Waiting for channel");
+                ch.get().join();
+                // TODO handle multiple candidate URLs
+                // TODO handle reconnection
+                return;
+            }
             IOHub hub = IOHub.create(executor);
             try {
                 SSLContext context;
@@ -489,11 +599,12 @@ public void run() {
             } finally {
                 hub.close();
             }
-        } catch (IOException e) {
+        } catch (Exception e) {
             events.error(e);
         }
     }
 
+    @SuppressFBWarnings(value = "UPM_UNCALLED_PRIVATE_METHOD", justification = "TODO pending WS/TCP switch")
     @SuppressWarnings({"ThrowableInstanceNeverThrown"})
     private void innerRun(IOHub hub, SSLContext context, ExecutorService service) {
         // Create the protocols that will be attempted to connect to the master.

src/main/java/hudson/remoting/Launcher.java

@@ -383,6 +383,7 @@ public void run() throws Exception {
                 // But we set it up just in case there are overrides somewhere in the logic
                 jnlpArgs.add("-disableHttpsCertValidation");
             }
+            // TODO ws support
             try {
                 hudson.remoting.jnlp.Main._main(jnlpArgs.toArray(new String[jnlpArgs.size()]));
             } catch (CmdLineException e) {

src/main/java/hudson/remoting/jnlp/Main.java

@@ -83,6 +83,8 @@ public class Main {
             usage="Specify the Jenkins root URLs to connect to.")
     public List<URL> urls = new ArrayList<>();
 
+    // TODO option to enable WS
+
     @Option(name="-credentials",metaVar="USER:PASSWORD",
             usage="HTTP BASIC AUTH header to pass in for making HTTP requests.")
     public String credentials;