 
 (defn- stream-or-finalize-response
   [state ^HttpChunk msg args]
-  (let [downstream-fn (.downstream args)
-        upstream-fn   (.upstream args)]
+  (let [upstream-fn   (.upstream args)
+        downstream-fn (.downstream args)]
     (if (.isLast msg)
       (do (swap-then!
            state
            (fn [current-state]
              (if (= response-pending (.next-up-fn current-state))
                (assoc current-state
-                 :next-up-fn request-complete
-                 :next-dn-fn nil)
+                 :next-dn-fn request-complete
+                 :next-up-fn nil)
                (assoc current-state
-                 :next-dn-fn nil)))
+                 :next-up-fn nil)))
            finalize-request)
-          (downstream-fn upstream-fn :done nil))
-      (downstream-fn upstream-fn :body (.getContent msg)))))
+          (upstream-fn downstream-fn :done nil))
+      (upstream-fn downstream-fn :body (.getContent msg)))))
 
 (defn- initial-response
   [state ^HttpResponse msg current-state]

  ;;       all downstream function calls happen on the same
  ;;       thread.
  (when (.chunked? current-state)
    ((.downstream current-state) (.upstream current-state) :connected nil)))

(defn- handle-ch-interest-change
  [state current-state writable?]
  (let [downstream-fn (.downstream current-state)
        upstream-fn   (.upstream current-state)
        ch            (.ch current-state)]
    (when (and downstream-fn (not= (.isWritable ch) @writable?))
      (swap-then!
       writable? not
       #(try (downstream-fn upstream-fn (if % :resume :pause) nil)
             (catch Exception err
               (throw (Exception. "Not implemented yet"))))))))


           (.setReadable (.ch current-state) true)

           :else
           ((.next-dn-fn current-state) state evt val current-state)))))))

(defn- mk-initial-state
  [pool [hdrs body :as  req] upstream-fn]
  (let [state       (atom nil)
        downstream-fn (mk-downstream-fn state)]
    (swap!
     state
     (fn [_] (State. pool               ;; The connection pool
