Merge pull request #592 from arnaudgeiser/542

client: ensure response in case of Exception

Author: KingMob

src/aleph/http/client.clj

@@ -554,12 +554,16 @@
                             ;; might be different in case we use :multipart
                             (reset! save-body body))
 
-                          (netty/safe-execute ch
-                            (http/send-message ch true ssl? req' body))))
+                          (-> (netty/safe-execute ch
+                                (http/send-message ch true ssl? req' body))
+                              (d/catch' (fn [e]
+                                          (s/put! responses (d/error-deferred e))
+                                          (netty/close ch))))))
 
                       ;; this will usually happen because of a malformed request
                       (catch Throwable e
-                        (s/put! responses (d/error-deferred e)))))
+                        (s/put! responses (d/error-deferred e))
+                        (netty/close ch))))
                   requests)
 
                 (s/on-closed responses

src/aleph/http/core.clj

@@ -509,6 +509,12 @@
             (d/catch' (fn [_]))))]
 
   (defn send-message
+    "Write an HttpMessage and body to a Netty channel.
+
+     Accepts Strings, ByteBuffers, ByteBufs, byte[], ChunkedInputs,
+     Files, Paths, HttpFiles, seqs and streams for bodies.
+
+     Seqs and streams must be, or be coercible to, a stream of ByteBufs."
     [ch keep-alive? ssl? ^HttpMessage msg body]
 
     (let [f (cond
@@ -539,7 +545,8 @@
                 (try
                   (send-streaming-body ch msg body)
                   (catch Throwable e
-                    (log/error e "error sending body of type " class-name)))))]
+                    (log/error e "error sending body of type " class-name)
+                    (throw e)))))]
 
       (when-not keep-alive?
         (handle-cleanup ch f))

src/aleph/netty.clj

@@ -272,7 +272,12 @@
     x
     (.channel ^ChannelHandlerContext x)))
 
-(defmacro safe-execute [ch & body]
+(defmacro safe-execute
+  "Executes the body on the event-loop (an executor service) associated with the Netty channel.
+
+   Executes immediately if current thread is in the event loop. Otherwise, returns a deferred
+   that will hold the result once done."
+  [ch & body]
   `(let [f# (fn [] ~@body)
          event-loop# (-> ~ch aleph.netty/channel .eventLoop)]
      (if (.inEventLoop event-loop#)

test/aleph/http_test.clj

@@ -273,6 +273,13 @@
                                 {:body (io/file "test/file.txt")
                                  :pool client-pool}))))))))
 
+(deftest test-invalid-body
+  (let [client-url (str "http://localhost:" port)]
+    (with-handler identity
+      (is (thrown? IllegalArgumentException
+                   @(http-put client-url
+                              {:body 123}))))))
+
 (def characters
   (let [charset (conj (mapv char (range 32 127)) \newline)]
     (repeatedly #(rand-nth charset))))