Fix for memory leak when using InputStreams to obtain body

The original ch close handler holds a reference to body' even if it didn't
need it. And since channels are extremely long-lived, many non-stream
resources could be inadvertently held onto.

Fixes #523.

Author: KingMob

src/aleph/http/core.clj

@@ -308,37 +308,39 @@
 
     (let [src (if (or (sequential? body') (s/stream? body'))
                 (->> body'
-                  s/->source
-                  (s/map (fn [x]
-                           (try
-                             (netty/to-byte-buf x)
-                             (catch Throwable e
-                               (log/error e "error converting " (.getName (class x)) " to ByteBuf")
-                               (netty/close ch))))))
+                     s/->source
+                     (s/map (fn [x]
+                              (try
+                                (netty/to-byte-buf x)
+                                (catch Throwable e
+                                  (log/error e "error converting " (.getName (class x)) " to ByteBuf")
+                                  (netty/close ch))))))
                 (netty/to-byte-buf-stream body' 8192))
 
-          sink (netty/sink ch false #(DefaultHttpContent. %))]
+          sink (netty/sink ch false #(DefaultHttpContent. %))
+
+          ;; mustn't close over body' if NOT a stream, can hold on to data too long when conns are keep-alive
+          ch-close-handler (if (s/stream? body')
+                             #(s/close! body')
+                             #(s/close! src))]
 
       (s/connect src sink)
 
       (-> ch
-        netty/channel
-        .closeFuture
-        netty/wrap-future
-        (d/chain' (fn [_] (if (s/stream? body')
-                            (s/close! body')
-                            (s/close! src)))))
+          netty/channel
+          .closeFuture
+          netty/wrap-future
+          (d/chain' (fn [_] (ch-close-handler))))
 
       (let [d (d/deferred)]
         (s/on-closed sink
-          (fn []
-
-            (when (instance? Closeable body)
-              (.close ^Closeable body))
+                     (fn []
+                       (when (instance? Closeable body)
+                         (.close ^Closeable body))
 
-            (.execute (-> ch aleph.netty/channel .eventLoop)
-              #(d/success! d
-                 (netty/write-and-flush ch empty-last-content)))))
+                       (.execute (-> ch aleph.netty/channel .eventLoop)
+                                 #(d/success! d
+                                              (netty/write-and-flush ch empty-last-content)))))
         d))
 
     (netty/write-and-flush ch empty-last-content)))
@@ -496,12 +498,12 @@
       handle-cleanup
       (fn [ch f]
         (-> f
-          (d/chain'
-            (fn [^ChannelFuture f]
-              (if f
-                (.addListener f ChannelFutureListener/CLOSE)
-                (netty/close ch))))
-          (d/catch' (fn [_]))))]
+            (d/chain'
+              (fn [^ChannelFuture f]
+                (if f
+                  (.addListener f ChannelFutureListener/CLOSE)
+                  (netty/close ch))))
+            (d/catch' (fn [_]))))]
 
   (defn send-message
     [ch keep-alive? ssl? ^HttpMessage msg body]

src/aleph/http/server.clj

@@ -185,28 +185,28 @@
                   (error-response e))))]
 
     (-> previous-response
-      (d/chain'
-        netty/wrap-future
-        (fn [_]
-          (netty/release req)
-          (-> rsp
-            (d/catch' error-response)
-            (d/chain'
-              (fn [rsp]
-                (when (not (-> req' ^AtomicBoolean (.websocket?) .get))
-                  (send-response ctx keep-alive? ssl?
-                    (cond
-
-                      (map? rsp)
-                      (if head?
-                        (assoc rsp :body :aleph/omitted)
-                        rsp)
-
-                      (nil? rsp)
-                      {:status 204}
-
-                      :else
-                      (invalid-value-response req rsp))))))))))))
+        (d/chain'
+          netty/wrap-future
+          (fn [_]
+            (netty/release req)
+            (-> rsp
+                (d/catch' error-response)
+                (d/chain'
+                  (fn [rsp]
+                    (when (not (-> req' ^AtomicBoolean (.websocket?) .get))
+                      (send-response ctx keep-alive? ssl?
+                                     (cond
+
+                                       (map? rsp)
+                                       (if head?
+                                         (assoc rsp :body :aleph/omitted)
+                                         rsp)
+
+                                       (nil? rsp)
+                                       {:status 204}
+
+                                       :else
+                                       (invalid-value-response req rsp))))))))))))
 
 (defn exception-handler [ctx ex]
   (when-not (instance? IOException ex)