Wire up response-executor option in connection-pool (fixes #773)

- Add PoolWithExecutor record to associate response-executor with pool
- Update connection-pool to return wrapped pool when response-executor is provided
- Update request function to extract executor from pool wrapper
- Add test to verify executor affinity correctly

Author: gabrielfranca95

project.clj

@@ -7,7 +7,7 @@
   :description "A framework for asynchronous communication"
   :url "https://github.com/clj-commons/aleph"
   :license {:name "MIT License"}
-  :dependencies [[org.clojure/clojure "1.12.2"]
+  :dependencies [[org.clojure/clojure "1.12.0"]
                  [org.clojure/tools.logging "1.3.0" :exclusions [org.clojure/clojure]]
                  [manifold "0.4.3" :exclusions [org.clojure/tools.logging]]
                  [org.clj-commons/byte-streams "0.3.4"]

src/aleph/http.clj

@@ -23,14 +23,14 @@
       ReadTimeoutException
       RequestCancellationException
       RequestTimeoutException)
-    (io.aleph.dirigiste Pools)
+    (io.aleph.dirigiste IPool Pools)
     (io.netty.channel ConnectTimeoutException)
     (io.netty.handler.codec Headers)
     (io.netty.handler.codec.http HttpHeaders)
     (java.net
       InetSocketAddress
       URI)
-    (java.util.concurrent TimeoutException)))
+    (java.util.concurrent Executor TimeoutException)))
 
 (defn start-server
   "Starts an HTTP server using the provided Ring `handler`.  Returns a server
@@ -133,6 +133,29 @@
 (def ^:no-doc default-response-executor
   (flow/utilization-executor 0.9 256 {:onto? false}))
 
+;; Wrapper record to associate a response-executor with a connection pool.
+;; This enables the `request` function to use the pool's executor for response deferreds.
+(defrecord PoolWithExecutor [^IPool pool ^Executor response-executor]
+  java.io.Closeable
+  (close [_] (.shutdown pool))
+
+  IPool
+  (shutdown [_] (.shutdown pool)))
+
+(defn- get-pool
+  "Returns the underlying IPool from a pool or PoolWithExecutor."
+  ^IPool [pool]
+  (if (instance? PoolWithExecutor pool)
+    (:pool pool)
+    pool))
+
+(defn- get-response-executor
+  "Returns the response-executor from a PoolWithExecutor, or the default executor."
+  [pool]
+  (if (instance? PoolWithExecutor pool)
+    (:response-executor pool)
+    default-response-executor))
+
 (defn connection-pool
   "Returns a connection pool which can be used as an argument in `request`.
 
@@ -251,19 +274,20 @@
 
                               true
                               (update :ssl-context #(client/ssl-context % http-versions insecure?)))
+        response-executor' (:response-executor connection-options)
         p (promise)
         create-pool-fn (or pool-builder-fn
                            flow/instrumented-pool)
         create-pool-ctrl-fn (or pool-controller-builder-fn
                                 #(Pools/utilizationController target-utilization connections-per-host total-connections))
-        pool (create-pool-fn
+        raw-pool (create-pool-fn
                {:generate       (fn [host]
                                   (let [c (promise)
                                         conn (create-connection
                                                host
                                                conn-options'
                                                middleware
-                                               #(flow/dispose @p host [@c]))]
+                                               #(flow/dispose (get-pool @p) host [@c]))]
                                     (deliver c conn)
                                     [conn]))
                 :destroy        (fn [_ c]
@@ -273,7 +297,10 @@
                 :control-period control-period
                 :max-queue-size max-queue-size
                 :controller     (create-pool-ctrl-fn)
-                :stats-callback stats-callback})]
+                :stats-callback stats-callback})
+        pool (if response-executor'
+               (->PoolWithExecutor raw-pool response-executor')
+               raw-pool)]
     @(deliver p pool)))
 
 (def ^:no-doc default-connection-pool
@@ -384,20 +411,21 @@
              request-timeout
              read-timeout]
       :or   {pool               default-connection-pool
-             response-executor  default-response-executor
              middleware         identity
              connection-timeout aleph.netty/default-connect-timeout}
       :as   req}]
-    (let [dispose-conn! (atom (fn []))
-          result (d/deferred response-executor)
-          response (executor/with-executor response-executor
+    (let [response-executor' (or response-executor (get-response-executor pool))
+          raw-pool (get-pool pool)
+          dispose-conn! (atom (fn []))
+          result (d/deferred response-executor')
+          response (executor/with-executor response-executor'
                      ((middleware
                        (fn [req]
                          (let [k (client/req->domain req)
                                start (System/currentTimeMillis)]
 
                            ;; acquire a connection
-                           (-> (flow/acquire pool k)
+                           (-> (flow/acquire raw-pool k)
                                (maybe-timeout! pool-timeout)
 
                                ;; pool timeout triggered
@@ -410,7 +438,7 @@
                                   ;; NOTE: All error handlers below delegate disposal of the
                                   ;; connection to the error handler on `result` which uses this
                                   ;; function.
-                                  (reset! dispose-conn! (fn [] (flow/dispose pool k conn)))
+                                  (reset! dispose-conn! (fn [] (flow/dispose raw-pool k conn)))
 
                                   (if (realized? result)
                                     ;; to account for race condition between setting `dispose-conn!`
@@ -478,8 +506,8 @@
                                                                      (<= 400 (:status rsp)))
                                                                (do
                                                                  (log/trace "Connection finished. Disposing...")
-                                                                 (flow/dispose pool k conn))
-                                                               (flow/release pool k conn)))))
+                                                                 (flow/dispose raw-pool k conn))
+                                                               (flow/release raw-pool k conn)))))
                                                       (-> rsp
                                                           (dissoc :aleph/destroy-conn?)
                                                           (assoc :connection-time (- end start)))))))))

test/aleph/http_test.clj

@@ -757,6 +757,22 @@
          (let [rsp (http-get "/" {:connection-pool pool})]
            (is (= http/default-response-executor (.executor rsp))))))))
 
+(deftest test-connection-pool-response-executor
+  (let [pool-executor (java.util.concurrent.Executors/newFixedThreadPool 1)
+        pool (http/connection-pool
+               {:connection-options
+                {:response-executor pool-executor
+                 :insecure? true}})]
+    (try
+      (with-both-handlers basic-handler
+        (let [rsp (http-get "/string" {:pool pool})]
+          ;; Verify that the response deferred uses the custom pool executor
+          (is (= pool-executor (.executor rsp))
+              "Response deferred should use the connection pool's response-executor")))
+      (finally
+        (.shutdown pool-executor)
+        (.shutdown pool)))))
+
 (deftest test-trace-request-omitted-body
   (with-handler echo-handler
     (is (= "" (-> @(http-trace "/" {:body "REQUEST"})