Add websockets to Jetty adapter

ring-jetty-adapter/project.clj

@@ -7,11 +7,13 @@
   :dependencies [[org.clojure/clojure "1.7.0"]
                  [ring/ring-core "1.10.0"]
                  [ring/ring-jakarta-servlet "1.10.0"]
-                 [org.eclipse.jetty/jetty-server "11.0.15"]]
+                 [org.eclipse.jetty/jetty-server "11.0.15"]
+                 [org.eclipse.jetty.websocket/websocket-jetty-server "11.0.15"]]
   :aliases {"test-all" ["with-profile" "default:+1.8:+1.9:+1.10:+1.11" "test"]}
   :profiles
   {:dev  {:dependencies [[clj-http "3.12.3"]
-                         [less-awful-ssl "1.0.6"]]
+                         [less-awful-ssl "1.0.6"]
+                         [hato "0.9.0"]]
           :jvm-opts ["-Dorg.eclipse.jetty.server.HttpChannelState.DEFAULT_TIMEOUT=500"]}
    :1.8  {:dependencies [[org.clojure/clojure "1.8.0"]]}
    :1.9  {:dependencies [[org.clojure/clojure "1.9.0"]]}

ring-jetty-adapter/src/ring/adapter/jetty.clj

@@ -2,8 +2,10 @@
   "A Ring adapter that uses the Jetty 9 embedded web server.
 
   Adapters are used to convert Ring handlers into running web servers."
-  (:require [ring.util.jakarta.servlet :as servlet])
-  (:import [org.eclipse.jetty.server
+  (:require [ring.util.jakarta.servlet :as servlet]
+            [ring.websocket :as ws])
+  (:import [java.nio ByteBuffer]
+           [org.eclipse.jetty.server
             Request
             Server
             ServerConnector
@@ -16,32 +18,91 @@
            [org.eclipse.jetty.util BlockingArrayQueue]
            [org.eclipse.jetty.util.thread ThreadPool QueuedThreadPool]
            [org.eclipse.jetty.util.ssl SslContextFactory$Server KeyStoreScanner]
+           [org.eclipse.jetty.websocket.server
+            JettyWebSocketServerContainer
+            JettyWebSocketCreator]
+           [org.eclipse.jetty.websocket.api
+            Session
+            WebSocketConnectionListener
+            WebSocketListener
+            WebSocketPingPongListener]
            [jakarta.servlet AsyncContext DispatcherType AsyncEvent AsyncListener]
            [jakarta.servlet.http HttpServletRequest HttpServletResponse]))
 
+(defn- websocket-socket [^Session session]
+  (let [remote (.getRemote session)]
+    (reify ws/Socket
+      (-send [_ message]
+        (if (string? message)
+          (.sendString remote message)
+          (.sendBytes remote message)))
+      (-ping [_ data]
+        (.sendPing remote data))
+      (-pong [_ data]
+        (.sendPong remote data))
+      (-close [_ status reason]
+        (.close session status reason)))))
+
+(defn- websocket-listener [listener]
+  (let [socket (volatile! nil)]
+    (reify
+      WebSocketConnectionListener
+      (onWebSocketConnect [_ session]
+        (vswap! socket (websocket-socket session))
+        (ws/on-open listener @socket))
+      (onWebSocketClose [_ status reason]
+        (ws/on-close listener @socket status reason))
+      (onWebSocketError [_ throwable]
+        (ws/on-error listener @socket throwable))
+      WebSocketListener
+      (onWebSocketText [_ message]
+        (ws/on-message listener @socket message))
+      (onWebSocketBinary [_ payload offset length]
+        (let [buffer (ByteBuffer/wrap payload offset length)]
+          (ws/on-message listener @socket buffer)))
+      WebSocketPingPongListener
+      (onWebSocketPing [_ _])
+      (onWebSocketPong [_ payload]
+        (ws/on-pong listener @socket payload)))))
+
+(defn- websocket-creator [{listener ::ws/listener}]
+  (reify JettyWebSocketCreator
+    (createWebSocket [_ _ _]
+      (websocket-listener listener))))
+
+(defn- upgrade-to-websocket [^HttpServletRequest request response response-map]
+  (let [context   (.getServletContext request)
+        container (JettyWebSocketServerContainer/getContainer context)
+        creator   (websocket-creator response-map)]
+    (.upgrade container creator request response)))
+
 (defn- ^AbstractHandler proxy-handler [handler]
   (proxy [AbstractHandler] []
     (handle [_ ^Request base-request ^HttpServletRequest request response]
       (when-not (= (.getDispatcherType request) DispatcherType/ERROR)
         (let [request-map  (servlet/build-request-map request)
               response-map (handler request-map)]
           (servlet/update-servlet-response response response-map)
+          #_(if (ws/websocket-response? response-map)
+            (upgrade-to-websocket request response response-map))
           (.setHandled base-request true))))))
 
 (defn- async-jetty-raise [^AsyncContext context ^HttpServletResponse response]
   (fn [^Throwable exception]
     (.sendError response 500 (.getMessage exception))
     (.complete context)))
 
-(defn- async-jetty-respond [context response]
+(defn- async-jetty-respond [context request response]
   (fn [response-map]
-    (servlet/update-servlet-response response context response-map)))
+    (if (ws/websocket-response? response-map)
+      (upgrade-to-websocket request response response-map)
+      (servlet/update-servlet-response response context response-map))))
 
 (defn- async-timeout-listener [request context response handler]
   (proxy [AsyncListener] []
     (onTimeout [^AsyncEvent _]
       (handler (servlet/build-request-map request)
-               (async-jetty-respond context response)
+               (async-jetty-respond context request response)
                (async-jetty-raise context response)))
     (onComplete [^AsyncEvent _])
     (onError [^AsyncEvent _])
@@ -58,7 +119,7 @@
            (async-timeout-listener request context response timeout-handler)))
         (handler
          (servlet/build-request-map request)
-         (async-jetty-respond context response)
+         (async-jetty-respond context request response)
          (async-jetty-raise context response))
         (.setHandled base-request true)))))