Merge pull request #484 from alexander-yakushev/server-aggregator

Add support for HttpObjectAggregator in server pipeline

Author: arnaudgeiser

src/aleph/http/server.clj

@@ -38,6 +38,7 @@
      TooLongFrameException]
     [io.netty.handler.codec.http
      DefaultFullHttpResponse
+     FullHttpRequest
      HttpContent HttpHeaders HttpUtil
      HttpContentCompressor
      HttpRequest HttpResponse
@@ -278,6 +279,16 @@
               (handle-request ctx req s))
             (reset! request req)))
 
+        process-full-request
+        (fn [ctx ^FullHttpRequest req]
+          ;; HttpObjectAggregator disables chunked encoding, no need to check for it.
+          (let [content (.content req)
+                body (when (pos? (.readableBytes content))
+                       (netty/buf->array content))]
+            ;; Don't release content as it will happen automatically once whole
+            ;; request is released.
+            (handle-request ctx req body)))
+
         process-last-content
         (fn [ctx ^HttpContent msg]
           (let [content (.content msg)]
@@ -356,6 +367,12 @@
       ([_ ctx msg]
         (cond
 
+          ;; Happens when io.netty.handler.codec.http.HttpObjectAggregator is part of the pipeline.
+          (instance? FullHttpRequest msg)
+          (if (invalid-request? msg)
+            (reject-invalid-request ctx msg)
+            (process-full-request ctx msg))
+
           (instance? HttpRequest msg)
           (if (invalid-request? msg)
             (reject-invalid-request ctx msg)
@@ -404,6 +421,21 @@
       ([_ ctx msg]
         (cond
 
+          ;; Happens when io.netty.handler.codec.http.HttpObjectAggregator is part of the pipeline.
+          (instance? FullHttpRequest msg)
+          (if (invalid-request? msg)
+            (reject-invalid-request ctx msg)
+            (let [^FullHttpRequest req msg
+                  content (.content req)
+                  ch (netty/channel ctx)
+                  s (netty/source ch)]
+              (when-not (zero? (.readableBytes content))
+                ;; Retain the content of FullHttpRequest one extra time to
+                ;; compensate for it being released together with the request.
+                (netty/put! ch s (netty/acquire content)))
+              (s/close! s)
+              (handle-request ctx req s)))
+
           (instance? HttpRequest msg)
           (if (invalid-request? msg)
             (reject-invalid-request ctx msg)

test/aleph/http_test.clj

@@ -16,7 +16,7 @@
   (:import
    (aleph.utils ConnectionTimeoutException RequestTimeoutException)
    (io.netty.channel ChannelHandlerContext ChannelOutboundHandlerAdapter ChannelPipeline ChannelPromise)
-   (io.netty.handler.codec.http HttpMessage)
+   (io.netty.handler.codec.http HttpMessage HttpObjectAggregator)
    (java.io File)
    (java.util.concurrent TimeoutException)
    (java.util.zip GZIPInputStream ZipException)
@@ -601,3 +601,41 @@
                                       (.write ctx msg p)))))})
       (let [resp @(http-get (str "http://localhost:" port "/string"))]
         (is (= test-header-val (get (:headers resp) test-header-name)))))))
+
+(defn add-http-object-aggregator [^ChannelPipeline pipeline]
+  (let [max-content-length 5]
+    (.addBefore pipeline
+                "request-handler"
+                "http-object-aggregator"
+                (HttpObjectAggregator. max-content-length))))
+
+(deftest test-http-object-aggregator-support
+  (with-server (http/start-server
+                basic-handler
+                {:port port
+                 :pipeline-transform add-http-object-aggregator})
+    (let [rsp @(http-put (str "http://localhost:" port "/echo")
+                         {:body "hello"})]
+      (is (= "hello" (bs/to-string (:body rsp))))
+      (is (= 200 (:status rsp))))
+
+    (let [rsp @(http-put (str "http://localhost:" port "/echo")
+                         {:body "hello, world!"})]
+      (is (= 413 (:status rsp)))
+      (is (empty? (bs/to-string (:body rsp)))))))
+
+(deftest test-http-object-aggregator-raw-stream-support
+  (with-server (http/start-server
+                basic-handler
+                {:port port
+                 :raw-stream? true
+                 :pipeline-transform add-http-object-aggregator})
+    (let [rsp @(http-put (str "http://localhost:" port "/echo")
+                         {:body "hello"})]
+      (is (= "hello" (bs/to-string (:body rsp))))
+      (is (= 200 (:status rsp))))
+
+    (let [rsp @(http-put (str "http://localhost:" port "/echo")
+                         {:body "hello, world!"})]
+      (is (= 413 (:status rsp)))
+      (is (empty? (bs/to-string (:body rsp)))))))