Support user-defined response executor

Author: ferdinand-beyer

src/lsp4clj/server.clj

@@ -9,6 +9,7 @@
    [lsp4clj.protocols.endpoint :as protocols.endpoint]
    [lsp4clj.trace :as trace]
    [promesa.core :as p]
+   [promesa.exec :as p.exec]
    [promesa.protocols :as p.protocols])
   (:import
    (java.util.concurrent CancellationException)))
@@ -196,6 +197,7 @@
                        trace-ch
                        tracer*
                        ^java.time.Clock clock
+                       response-executor
                        on-close
                        request-id*
                        pending-sent-requests*
@@ -357,13 +359,13 @@
             ;; thread, whatever that thread is. Since the callbacks are not
             ;; under our control, they are under our users' control, they could
             ;; block. Therefore, we do not want the completing thread to be our
-            ;; thread. This is very easy for users to
-            ;; miss, therefore we complete the promise on the default executor.
-            (p/thread-call :default
-                           (fn []
-                             (if error
-                               (p/reject! p (ex-info "Received error response" resp))
-                               (p/resolve! p result)))))
+            ;; thread. This is very easy for users to miss, therefore we
+            ;; complete the promise using an explicit executor.
+            (p.exec/submit! response-executor
+                            (fn []
+                              (if error
+                                (p/reject! p (ex-info "Received error response" resp))
+                                (p/resolve! p result)))))
           (trace this trace/received-unmatched-response resp now)))
       (catch Throwable e
         (log-error-receiving this e resp))))
@@ -420,9 +422,10 @@
   (update server :tracer* reset! (trace/tracer-for-level trace-level)))
 
 (defn chan-server
-  [{:keys [output-ch input-ch log-ch trace? trace-level trace-ch clock on-close]
+  [{:keys [output-ch input-ch log-ch trace? trace-level trace-ch clock on-close response-executor]
     :or {clock (java.time.Clock/systemDefaultZone)
-         on-close (constantly nil)}}]
+         on-close (constantly nil)
+         response-executor :default}}]
   (let [;; before defaulting trace-ch, so that default is "off"
         tracer (trace/tracer-for-level (or trace-level
                                            (when (or trace? trace-ch) "verbose")
@@ -437,6 +440,7 @@
        :tracer* (atom tracer)
        :clock clock
        :on-close on-close
+       :response-executor response-executor
        :request-id* (atom 0)
        :pending-sent-requests* (atom {})
        :pending-received-requests* (atom {})

test/lsp4clj/server_test.clj

@@ -481,6 +481,73 @@
              (h/assert-take output-ch)))
       (server/shutdown server))))
 
+(defn- core-async-dispatch-thread? [^Thread thread]
+  (re-matches #"async-dispatch-\d+" (.getName thread)))
+
+(deftest can-determine-core-async-dispatch-thread
+  (testing "current thread"
+    (is (not (core-async-dispatch-thread? (Thread/currentThread)))))
+  (testing "thread running go blocks"
+    (let [ch (async/chan)
+          _ (async/go (async/>! ch (Thread/currentThread)))
+          thread (async/<!! ch)]
+      (is (core-async-dispatch-thread? thread))))
+  (testing "thread running core.async thread macro"
+    (let [ch (async/chan)
+          _ (async/thread (async/>!! ch (Thread/currentThread)))
+          thread (async/<!! ch)]
+      (is (not (core-async-dispatch-thread? thread))))))
+
+(deftest request-should-complete-on-a-suitable-executor
+  (testing "successful completion"
+    (let [input-ch (async/chan 3)
+          output-ch (async/chan 3)
+          server (server/chan-server {:output-ch output-ch
+                                      :input-ch input-ch})
+          _ (server/start server nil)
+          thread-p (-> (server/send-request server "req" {:body "foo"})
+                       (p/then (fn [_] (Thread/currentThread))))
+          client-rcvd-msg (h/assert-take output-ch)
+          _ (async/put! input-ch (lsp.responses/response (:id client-rcvd-msg) {:result "good"}))
+          thread (deref thread-p 100 nil)]
+      (is (not (core-async-dispatch-thread? thread)))
+      (is (instance? java.util.concurrent.ForkJoinWorkerThread thread)
+          "completes on default ForkJoinPool executor")
+      (server/shutdown server)))
+  (testing "exceptional completion"
+    (let [input-ch (async/chan 3)
+          output-ch (async/chan 3)
+          server (server/chan-server {:output-ch output-ch
+                                      :input-ch input-ch})
+          _ (server/start server nil)
+          thread-p (-> (server/send-request server "req" {:body "foo"})
+                       (p/catch (fn [_] (Thread/currentThread))))
+          client-rcvd-msg (h/assert-take output-ch)
+          _ (async/put! input-ch
+                        (-> (lsp.responses/response (:id client-rcvd-msg))
+                            (lsp.responses/error {:code 1234
+                                                  :message "Something bad"
+                                                  :data {:body "foo"}})))
+          thread (deref thread-p 100 nil)]
+      (is (not (core-async-dispatch-thread? thread)))
+      (is (instance? java.util.concurrent.ForkJoinWorkerThread thread)
+          "completes on default ForkJoinPool executor")
+      (server/shutdown server)))
+  (testing "completion with :current-thread executor for legacy behavior"
+    (let [input-ch (async/chan 3)
+          output-ch (async/chan 3)
+          server (server/chan-server {:output-ch output-ch
+                                      :input-ch input-ch
+                                      :response-executor :current-thread})
+          _ (server/start server nil)
+          thread-p (-> (server/send-request server "req" {:body "foo"})
+                       (p/then (fn [_] (Thread/currentThread))))
+          client-rcvd-msg (h/assert-take output-ch)
+          _ (async/put! input-ch (lsp.responses/response (:id client-rcvd-msg) {:result "good"}))
+          thread (deref thread-p 100 nil)]
+      (is (core-async-dispatch-thread? thread) "completes on core.async dispatch thread")
+      (server/shutdown server))))
+
 (def fixed-clock
   (-> (java.time.LocalDateTime/of 2022 03 05 13 35 23 0)
       (.toInstant java.time.ZoneOffset/UTC)