 (defn- downstream-fn
   [state channel keepalive?]
   ;; The state of the response needs to be tracked
-  (let [next-fn (atom #(initialize-resp %1 %2
-                                        state channel
-                                        keepalive? false nil))]
+  (let [next-fn (atom #(initialize-resp
+                        %1 %2
+                        state channel
+                        keepalive? false nil))]
 
     ;; The upstream application will call this function to
     ;; send the response back to the client

                     (.getContent msg)
                     nil)])
        (upstream :done nil)
        (partial transition-from-req-done upstream)))))

(defn- netty-bridge
  [app opts]

  "State transition from an HTTP request has finished being processed.
   If the response has already been sent, then the next state is to
   listen for new requests."
  [state]
  (swap! state
         (fn [[_ [app opts upstream] responded?]]
           (if responded?
             [incoming-request [app opts]]
             [waiting-for-response [app opts upstream]]))))

(defn- transition-to-streaming-body
  [state]
  (swap! state
         (fn [[_ args responded?]]
           [stream-request-body args responded?])))

(defn- register-upstream-handler
  [state upstream]
  (swap! state
         (fn [[f [app opts] responded?]]
           [f [app opts upstream] responded?])))

(defn- stream-request-body
  [state _ ^HttpChunk chunk [app opts upstream]]
  [[_ _ upstream] _  _ ^HttpChunk chunk]
  (if (.isLast chunk)
    (do
      (upstream :done nil)
      (transition-from-req-done state))
    (upstream :body (.getContent chunk))))

(defn- incoming-request
  [state ch ^HttpMessage msg [app opts]]
  ;; [[app opts] state channel ^HttpMessage msg]
  (let [upstream (app (downstream-fn state ch (HttpHeaders/isKeepAlive msg)))
        headers  (headers-from-netty-req msg)]

    ;; Add the upstream handler to the state
    (register-upstream-handler state upstream)

    ;; Send the HTTP headers upstream
    (if (.isChunked msg)
      (do
        (upstream :request [headers nil])
        (transition-to-streaming-body state))
      (do
        (upstream :request
                  [headers
