Add async timeout hander to Jetty adapter

Add a new :async-timeout-handler option to the Jetty adapter to
customize what happens when the :async-timeout is exceeded.

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

@@ -16,7 +16,7 @@
            [org.eclipse.jetty.util BlockingArrayQueue]
            [org.eclipse.jetty.util.thread ThreadPool QueuedThreadPool]
            [org.eclipse.jetty.util.ssl SslContextFactory$Server]
-           [javax.servlet AsyncContext DispatcherType]
+           [javax.servlet AsyncContext DispatcherType AsyncEvent AsyncListener]
            [javax.servlet.http HttpServletRequest HttpServletResponse]))
 
 (defn- ^AbstractHandler proxy-handler [handler]
@@ -28,18 +28,38 @@
           (servlet/update-servlet-response response response-map)
           (.setHandled base-request true))))))
 
-(defn- ^AbstractHandler async-proxy-handler [handler timeout]
+(defn- async-jetty-raise [^AsyncContext context ^HttpServletResponse response]
+  (fn [^Throwable exception]
+    (.sendError response 500 (.getMessage exception))
+    (.complete context)))
+
+(defn- async-jetty-respond [context response]
+  (fn [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-raise context response)))
+    (onComplete [^AsyncEvent _])
+    (onError [^AsyncEvent _])
+    (onStartAsync [^AsyncEvent _])))
+
+(defn- ^AbstractHandler async-proxy-handler [handler timeout timeout-handler]
   (proxy [AbstractHandler] []
     (handle [_ ^Request base-request ^HttpServletRequest request ^HttpServletResponse response]
       (let [^AsyncContext context (.startAsync request)]
         (.setTimeout context timeout)
+        (when timeout-handler
+          (.addListener
+           context
+           (async-timeout-listener request context response timeout-handler)))
         (handler
          (servlet/build-request-map request)
-         (fn [response-map]
-           (servlet/update-servlet-response response context response-map))
-         (fn [^Throwable exception]
-           (.sendError response 500 (.getMessage exception))
-           (.complete context)))
+         (async-jetty-respond context response)
+         (async-jetty-raise context response))
         (.setHandled base-request true)))))
 
 (defn- ^ServerConnector server-connector [^Server server & factories]
@@ -138,6 +158,7 @@
   :configurator         - a function called with the Jetty Server instance
   :async?               - if true, treat the handler as asynchronous
   :async-timeout        - async context timeout in ms (defaults to 0, no timeout)
+  :async-timeout-handler - an async handler to handle an async context timeout
   :port                 - the port to listen on (defaults to 80)
   :host                 - the hostname to listen on
   :join?                - blocks the thread until server ends (defaults to true)
@@ -178,7 +199,10 @@
   [handler options]
   (let [server (create-server (dissoc options :configurator))]
     (if (:async? options)
-      (.setHandler server (async-proxy-handler handler (:async-timeout options 0)))
+      (.setHandler server
+                   (async-proxy-handler handler
+                                        (:async-timeout options 0)
+                                        (:async-timeout-handler options)))
       (.setHandler server (proxy-handler handler)))
     (when-let [configurator (:configurator options)]
       (configurator server))

ring-jetty-adapter/test/ring/adapter/test/jetty.clj

@@ -398,6 +398,12 @@
     (error-cps request respond raise)
     (hello-world-cps request respond raise)))
 
+(defn- hello-world-slow-cps [request respond raise]
+  (future (Thread/sleep 1000)
+          (respond {:status  200
+                    :headers {"Content-Type" "text/plain"}
+                    :body    "Hello World"})))
+
 (deftest run-jetty-cps-test
   (testing "async response in future"
     (reset! thread-exceptions [])
@@ -447,7 +453,46 @@
     (with-server hello-world-streaming-long {:port test-port, :async? true}
       (let [response (http/get test-url)]
         (is (= (:body response)
-               (apply str (for [i (range 10)] (str "data: " i "\n\n")))))))))
+               (apply str (for [i (range 10)] (str "data: " i "\n\n"))))))))
+
+  (testing "async timeout handler"
+    (testing "when no timeout handler is passed, behaviour is unchanged"
+      (with-server hello-world-slow-cps {:port test-port
+                                         :async? true
+                                         :async-timeout 250}
+        (let [response (http/get test-url {:throw-exceptions false})]
+          (is (= (:status response)
+                 500)))))
+
+    (testing "with timeout handlers, ring-style responses are generated"
+      (with-server hello-world-slow-cps
+        {:port test-port
+         :async? true
+         :async-timeout 200
+         :async-timeout-handler (fn [request respond raise]
+                                  (respond
+                                   {:status 503
+                                    :headers {"Content-Type" "text/plain"}
+                                    :body "Request timed out"}))}
+        (let [response (http/get test-url {:throw-exceptions false})]
+          (is (= (:body response)
+                 "Request timed out"))
+          (is (= (:status response)
+                 503))))
+
+      (with-server hello-world-slow-cps
+        {:port test-port
+         :async? true
+         :async-timeout 200
+         :async-timeout-handler (fn [request respond raise]
+                                  (raise
+                                   (ex-info "An exception was thrown" {})))}
+        (let [response (http/get (str test-url "/test-path/testing")
+                                 {:throw-exceptions false})]
+          (is (.contains ^String (:body response)
+                         "An exception was thrown"))
+          (is (= (:status response)
+                 500)))))))
 
 (def call-count (atom 0))