Merge branch 'master' into fix-cleanup-2

Author: kachayev

project.clj

@@ -1,4 +1,4 @@
-(def netty-version "4.1.27.Final")
+(def netty-version "4.1.28.Final")
 
 (def netty-modules
   '[transport

src/aleph/http.clj

@@ -117,6 +117,7 @@
    | `name-resolver` | specify the mechanism to resolve the address of the unresolved named address. When not set or equals to `:default`, JDK's built-in domain name lookup mechanism is used (blocking). Set to`:noop` not to resolve addresses or pass an instance of `io.netty.resolver.AddressResolverGroup` you need. Note, that if the appropriate connection-pool is created with dns-options shared DNS resolver would be used
    | `proxy-options` | a map to specify proxy settings. HTTP, SOCKS4 and SOCKS5 proxies are supported. Note, that when using proxy `connections-per-host` configuration is still applied to the target host disregarding tunneling settings. If you need to limit number of connections to the proxy itself use `total-connections` setting.
    | `response-executor` | optional `java.util.concurrent.Executor` that will execute response callbacks
+   | `log-activity` | when set, logs all events on each channel (connection) with a log level given. Accepts either one of `:trace`, `:debug`, `:info`, `:warn`, `:error` or an instance of `io.netty.handler.logging.LogLevel`. Note, that this setting *does not* enforce any changes to the logging configuration (default configuration is `INFO`, so you won't see any `DEBUG` or `TRACE` level messages, unless configured explicitly)
 
    Supported `proxy-options` are
 
@@ -235,7 +236,8 @@
      | `pool-timeout` | timeout in milliseconds for the pool to generate a connection
      | `connection-timeout` | timeout in milliseconds for the connection to become established
      | `request-timeout` | timeout in milliseconds for the arrival of a response over the established connection
-     | `read-timeout` | timeout in milliseconds for the response to be completed"
+     | `read-timeout` | timeout in milliseconds for the response to be completed
+     | `follow-redirects?` | whether to follow redirects, defaults to `true`; see `aleph.http.client-middleware/handle-redirects`"
     [{:keys [pool
              middleware
              pool-timeout
@@ -247,8 +249,7 @@
       :or {pool default-connection-pool
            response-executor default-response-executor
            middleware identity
-           connection-timeout 6e4 ;; 60 seconds
-           follow-redirects? true}
+           connection-timeout 6e4} ;; 60 seconds
       :as req}]
 
     (executor/with-executor response-executor

src/aleph/http/client.clj

@@ -24,13 +24,17 @@
      HttpRequest
      HttpResponse
      HttpContent
+     HttpUtil
+     HttpHeaderNames
      LastHttpContent
      FullHttpResponse
      HttpObjectAggregator]
     [io.netty.channel
      Channel
      ChannelHandler ChannelHandlerContext
      ChannelPipeline]
+    [io.netty.handler.stream ChunkedWriteHandler]
+    [io.netty.handler.codec.http FullHttpRequest]
     [io.netty.handler.codec.http.websocketx
      CloseWebSocketFrame
      PingWebSocketFrame
@@ -51,6 +55,9 @@
      HttpProxyHandler
      Socks4ProxyHandler
      Socks5ProxyHandler]
+    [io.netty.handler.logging
+     LoggingHandler
+     LogLevel]
     [java.util.concurrent.atomic
      AtomicInteger]
     [aleph.utils
@@ -171,7 +178,7 @@
 
           (instance? HttpResponse msg)
           (let [rsp msg]
-            (if (HttpHeaders/isTransferEncodingChunked rsp)
+            (if (HttpUtil/isTransferEncodingChunked rsp)
               (let [s (netty/buffered-source (netty/channel ctx) #(alength ^bytes %) buffer-capacity)
                     c (d/deferred)]
                 (reset! stream s)
@@ -262,7 +269,7 @@
 ;;  * `curl` uses separate option `--proxytunnel` flag to switch tunneling on
 ;;  * `curl` uses CONNECT when sending request to HTTPS destination through HTTP proxy
 ;;
-;; Explicitily setting `tunnel?` to false when it's expected to use CONNECT
+;; Explicitly setting `tunnel?` to false when it's expected to use CONNECT
 ;; throws `IllegalArgumentException` to reduce the confusion
 (defn http-proxy-handler
   [^InetSocketAddress address
@@ -343,6 +350,21 @@
         (.remove (.pipeline ctx) this))
       (.fireUserEventTriggered ^ChannelHandlerContext ctx evt))))
 
+(defn coerce-log-level [level]
+  (if (instance? LogLevel level)
+    level
+    (let [netty-level (case level
+                        :trace LogLevel/TRACE
+                        :debug LogLevel/DEBUG
+                        :info LogLevel/INFO
+                        :warn LogLevel/WARN
+                        :error LogLevel/ERROR
+                        nil)]
+      (when (nil? netty-level)
+        (throw (IllegalArgumentException.
+                (str "unknown log level given: " level))))
+      netty-level)))
+
 (defn pipeline-builder
   [response-stream
    {:keys
@@ -354,7 +376,8 @@
      raw-stream?
      proxy-options
      ssl?
-     idle-timeout]
+     idle-timeout
+     log-activity]
     :or
     {pipeline-transform identity
      response-buffer-size 65536
@@ -365,7 +388,11 @@
   (fn [^ChannelPipeline pipeline]
     (let [handler (if raw-stream?
                     (raw-client-handler response-stream response-buffer-size)
-                    (client-handler response-stream response-buffer-size))]
+                    (client-handler response-stream response-buffer-size))
+          logger (when (some? log-activity)
+                   (LoggingHandler.
+                    "aleph-client"
+                    ^LogLevel (coerce-log-level log-activity)))]
       (doto pipeline
         (.addLast "http-client"
           (HttpClientCodec.
@@ -374,6 +401,7 @@
             max-chunk-size
             false
             false))
+        (.addLast "streamer" ^ChannelHandler (ChunkedWriteHandler.))
         (.addLast "handler" ^ChannelHandler handler)
         (http/attach-idle-handlers idle-timeout))
       (when (some? proxy-options)
@@ -388,6 +416,8 @@
               "pending-proxy-connection"
               ^ChannelHandler
               (pending-proxy-connection-handler response-stream)))))
+      (when (some? logger)
+        (.addFirst pipeline "activity-logger" logger))
       (pipeline-transform pipeline))))
 
 (defn close-connection [f]
@@ -448,20 +478,43 @@
                                           (assoc req :uri (:request-url req))
                                           req))]
                 (when-not (.get (.headers req') "Host")
-                  (HttpHeaders/setHost req' (str host (when explicit-port? (str ":" port)))))
+                  (.set (.headers req') HttpHeaderNames/HOST (str host (when explicit-port? (str ":" port)))))
                 (when-not (.get (.headers req') "Connection")
-                  (HttpHeaders/setKeepAlive req' keep-alive?))
+                  (HttpUtil/setKeepAlive req' keep-alive?))
                 (when (and (non-tunnel-proxy? proxy-options')
                         (get proxy-options :keep-alive? true)
                         (not (.get (.headers req') "Proxy-Connection")))
                   (.set (.headers req') "Proxy-Connection" "Keep-Alive"))
 
-                (let [body (if-let [parts (get req :multipart)]
-                             (let [boundary (multipart/boundary)
-                                   content-type (str "multipart/form-data; boundary=" boundary)]
-                               (HttpHeaders/setHeader req' "Content-Type" content-type)
-                               (multipart/encode-body boundary parts))
-                             (get req :body))]
+                (let [body (:body req)
+                      parts (:multipart req)
+                      multipart? (some? parts)
+                      [req' body] (cond
+                                    ;; RFC #7231 4.3.8. TRACE
+                                    ;; A client MUST NOT send a message body...
+                                    (= :trace (:request-method req))
+                                    (do
+                                      (when (or (some? body) multipart?)
+                                        (log/warn "TRACE request body was omitted"))
+                                      [req' nil])
+
+                                    (not multipart?)
+                                    [req' body]
+
+                                    :else
+                                    (multipart/encode-request req' parts))]
+
+                  (when-let [save-message (get req :aleph/save-request-message)]
+                    ;; debug purpose only
+                    ;; note, that req' is effectively mutable, so
+                    ;; it will "capture" all changes made during "send-message"
+                    ;; execution
+                    (reset! save-message req'))
+
+                  (when-let [save-body (get req :aleph/save-request-body)]
+                    ;; might be different in case we use :multipart
+                    (reset! save-body body))
+
                   (netty/safe-execute ch
                     (http/send-message ch true ssl? req' body))))
 
@@ -584,14 +637,14 @@
                        #(when (.isOpen ch)
                          (d/chain'
                            (netty/wrap-future (.close handshaker ch (CloseWebSocketFrame.)))
-                           (fn [_] (netty/close ch))))))))
+                           (fn [_] (netty/close ctx))))))))
 
                (instance? FullHttpResponse msg)
                (let [rsp ^FullHttpResponse msg]
                  (throw
                    (IllegalStateException.
                      (str "unexpected HTTP response, status: "
-                       (.getStatus rsp)
+                       (.status rsp)
                        ", body: '"
                        (bs/to-string (.content rsp))
                        "'"))))

src/aleph/http/client_middleware.clj

@@ -4,7 +4,6 @@
   (:refer-clojure :exclude [update])
   (:require
     [potemkin :as p]
-    [clojure.stacktrace :refer [root-cause]]
     [clojure.string :as str]
     [clojure.walk :refer [prewalk]]
     [manifold.deferred :as d]
@@ -786,7 +785,7 @@
     (write-cookie-header cookies cookie-spec req)))
 
 (defn wrap-cookies
-  "Middleware that set 'Cookie' header based on the contentn of cookies passed
+  "Middleware that set 'Cookie' header based on the content of cookies passed
    with the request or from cookies storage (when provided). Source for 'Cookie'
    header content by priorities:
 
@@ -906,6 +905,34 @@
 (defmethod coerce-response-body :default [_ resp]
   resp)
 
+(defn wrap-request-debug [req]
+  (cond-> req
+    (opt req :save-request)
+    (assoc :aleph/save-request-message (atom nil))
+
+    (opt req :debug-body)
+    (assoc :aleph/save-request-body (atom nil))))
+
+(defn handle-response-debug [req rsp]
+  (let [saved-message (get req :aleph/save-request-message)
+        saved-body (get req :aleph/save-request-body)
+        req' (dissoc req
+                     :aleph/save-request-body
+                     :aleph/save-request-message
+                     :save-request
+                     :save-request?
+                     :debug-body
+                     :debug-body?)]
+    (cond-> rsp
+      (some? saved-message)
+      (assoc :aleph/netty-request @saved-message)
+
+      (some? saved-body)
+      (assoc :aleph/request-body @saved-body)
+
+      (opt req :save-request)
+      (assoc :aleph/request req'))))
+
 (def default-middleware
   [wrap-method
    wrap-url
@@ -918,7 +945,8 @@
    wrap-accept
    wrap-accept-encoding
    wrap-content-type
-   wrap-cookies])
+   wrap-cookies
+   wrap-request-debug])
 
 (defn wrap-request
   "Returns a batteries-included HTTP request function corresponding to the given
@@ -938,7 +966,8 @@
 
               ;; 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)))))))))
+                (let [rsp' (handle-response-debug req' rsp)]
+                  (if (nil? body)
+                    rsp'
+                    (d/future-with (or executor (ex/wait-pool))
+                      (coerce-response-body req' rsp'))))))))))))

src/aleph/http/core.clj

@@ -23,6 +23,7 @@
     [io.netty.handler.codec.http
      DefaultHttpRequest DefaultLastHttpContent
      DefaultHttpResponse DefaultFullHttpRequest
+     FullHttpRequest
      HttpHeaders HttpUtil HttpContent
      HttpMethod HttpRequest HttpMessage
      HttpResponse HttpResponseStatus
@@ -34,7 +35,7 @@
      IdleStateEvent
      IdleStateHandler]
     [io.netty.handler.stream
-     ChunkedFile ChunkedWriteHandler]
+     ChunkedInput ChunkedFile ChunkedWriteHandler]
     [java.io
      File
      RandomAccessFile
@@ -246,6 +247,9 @@
       x
       (str x))))
 
+(defn chunked-writer-enabled? [^Channel ch]
+  (some? (-> ch netty/channel .pipeline (.get ChunkedWriteHandler))))
+
 (defn send-streaming-body [ch ^HttpMessage msg body]
 
   (HttpUtil/setTransferEncodingChunked msg (boolean (not (has-content-length? msg))))
@@ -327,6 +331,10 @@
     (netty/write ch msg)
     (netty/write-and-flush ch ci)))
 
+(defn send-chunked-body [ch ^HttpMessage msg ^ChunkedInput body]
+  (netty/write ch msg)
+  (netty/write-and-flush ch body))
+
 (defn send-file-region [ch ^HttpMessage msg ^File file]
   (let [raf (RandomAccessFile. file "r")
         len (.length raf)
@@ -345,7 +353,7 @@
         (bs/to-byte-buffers {:chunk-size 1e6})
         s/->source))
 
-    (-> ch netty/channel .pipeline (.get ChunkedWriteHandler))
+    (chunked-writer-enabled? ch)
     (send-chunked-file ch msg file)
 
     :else
@@ -395,6 +403,9 @@
                 (instance? ByteBuf body))
               (send-contiguous-body ch msg body)
 
+              (instance? ChunkedInput body)
+              (send-chunked-body ch msg body)
+
               (instance? File body)
               (send-file-body ch ssl? msg body)
 

src/aleph/http/encoding.clj

@@ -1,8 +1,7 @@
 (ns aleph.http.encoding
   (:require
     [byte-streams :as bs]
-    [primitive-math :as p]
-    [potemkin :refer [doary]])
+    [primitive-math :as p])
   (:import
     [io.netty.buffer
      ByteBuf