make sure body coercion happens off the netty thread, and don't write…

… content-length for responses to HEAD requests, fixes #267 and #268

src/aleph/http/client_middleware.clj

@@ -9,6 +9,7 @@
     [clojure.walk :refer [prewalk]]
     [manifold.deferred :as d]
     [manifold.stream :as s]
+    [manifold.executor :as ex]
     [byte-streams :as bs]
     [clojure.edn :as edn])
   (:import
@@ -624,14 +625,16 @@
                     wrap-exceptions
                     wrap-request-timing)]
     (fn [req]
-      (if (:aleph.http.client/close req)
-        (client req)
-
-        (let [req' (reduce #(%2 %1) req default-middleware)]
-          (d/chain' (client' req')
-
-            ;; coerce the response body
-            (fn [{:keys [body] :as rsp}]
-              (if body
-                (coerce-response-body req' rsp)
-                rsp))))))))
+      (let [executor (ex/executor)]
+        (if (:aleph.http.client/close req)
+          (client req)
+
+          (let [req' (reduce #(%2 %1) req default-middleware)]
+            (d/chain' (client' req')
+
+              ;; coerce the response body
+              (fn [{:keys [body] :as rsp}]
+                (if body
+                  (d/future-with (or executor (ex/wait-pool))
+                    (coerce-response-body req' rsp))
+                  rsp)))))))))

src/aleph/http/core.clj

@@ -324,15 +324,19 @@
       (netty/write-and-flush ch empty-last-content))))
 
 (defn send-contiguous-body [ch ^HttpMessage msg body]
-  (let [body (if body
-               (DefaultLastHttpContent. (netty/to-byte-buf ch body))
-               empty-last-content)
+  (let [omitted? (identical? :aleph/omitted body)
+        body (if (or (nil? body) omitted?)
+               empty-last-content
+               (DefaultLastHttpContent. (netty/to-byte-buf ch body)))
         length (-> ^HttpContent body .content .readableBytes)]
-    (if (instance? HttpResponse msg)
-      (let [code (-> ^HttpResponse msg .getStatus .code)]
-        (when-not (or (<= 100 code 199) (= 204 code))
-          (try-set-content-length! msg length)))
-      (try-set-content-length! msg length))
+
+    (when-not omitted?
+      (if (instance? HttpResponse msg)
+        (let [code (-> ^HttpResponse msg .getStatus .code)]
+          (when-not (or (<= 100 code 199) (= 204 code))
+            (try-set-content-length! msg length)))
+        (try-set-content-length! msg length)))
+
     (netty/write ch msg)
     (netty/write-and-flush ch body)))
 
@@ -357,6 +361,7 @@
 
               (or
                 (nil? body)
+                (identical? :aleph/omitted body)
                 (instance? String body)
                 (instance? ary-class body)
                 (instance? ByteBuffer body)

src/aleph/http/server.clj

@@ -33,7 +33,7 @@
      HttpContent HttpHeaders
      HttpRequest HttpResponse
      HttpResponseStatus DefaultHttpHeaders
-     HttpServerCodec HttpVersion
+     HttpServerCodec HttpVersion HttpMethod
      LastHttpContent]
     [io.netty.handler.codec.http.websocketx
      WebSocketServerHandshakerFactory
@@ -142,11 +142,12 @@
    handler
    rejected-handler
    executor
-   req
+   ^HttpRequest req
    previous-response
    body
    keep-alive?]
   (let [^NettyRequest req' (http/netty-request->ring-request req ssl? (.channel ctx) body)
+        head? (identical? HttpMethod/HEAD (.method req))
         rsp (if executor
 
               ;; handle request on a separate thread
@@ -182,7 +183,9 @@
                 (when (not (-> req' ^AtomicBoolean (.websocket?) .get))
                   (send-response ctx keep-alive? ssl?
                     (if (map? rsp)
-                      rsp
+                      (if head?
+                        (assoc rsp :body :aleph/omitted)
+                        rsp)
                       (invalid-value-response req rsp))))))))))))
 
 (defn ring-handler

test/aleph/http_test.clj

@@ -331,11 +331,11 @@
 
 ;;;
 
-(deftest ^:stress handle-large-responses []
+(deftest test-large-responses
   (with-handler basic-handler
     (let [pool (http/connection-pool {:connection-options {:response-buffer-size 16}})]
-      (dotimes [i 1e6]
-        (when (zero? (rem i 1e2))
+      (dotimes [i 1 #_1e6]
+        #_(when (zero? (rem i 1e2))
           (prn i))
         (-> @(http/get (str "http://localhost:" port "/big")
                {:as :byte-array})