                   (send-upstream state :abort err current-state))
                 (catch Exception _)))))))))
 
-(defn- mk-downstream-fn
+(defn- mk-downstream
+  [next-dn state]
+  (fn [evt val]
+    (let [current-state @state]
+      (when-not (.upstream current-state)
+        (throw (Exception. "Not callbable yet")))
+
+      (cond
+       (= :schedule evt)
+       (send-upstream state :schedule val current-state)
+
+       (= :abort evt)
+       (handle-err state val current-state)
+
+       :else
+       (next-dn evt val)))))
+
+(defn handler
+  [app opts]
+  (fn [dn]
+    (let [state    (atom (mk-initial-state dn))
+          upstream (app (mk-downstream dn state))]
+      ;; Save off the upstream
+      (swap! state #(assoc % :upstream upstream))
+      ;; And now the upstream
+      (fn [evt val]
+        (try
+          (let [current-state @state]
+            (if (= :abort evt)
+              (when-not (.aborting? current-state)
+                (swap-then!
+                 state
+                 #(assoc % :aborting? true)
+                 #(send-upstream state evt val %)))
+              (send-upstream state evt val current-state)))
+          (catch Exception err
+            (handle-err state err @state)))))))
+
+(defn- mk-netty-downstream-fn
   [state]
   (fn [evt val]
-    (let [current-state ^State @state]
-      (when-not (.upstream current-state)
-        (throw (Exception. "Not callbable yet")))
-
+    (let [current-state @state]
       (cond
        (= :message evt)
        (let [ch (.ch current-state)]

       (close-channel current-state)

       (= :pause evt)
       (when (.upstream current-state)
         (.setReadable (.ch current-state) false))

       (= :resume evt)
       (when (.upstream current-state)
         (.setReadable (.ch current-state) true))

       (= :schedule evt)
       (send-upstream state evt val current-state)

       (= :abort evt)
       (handle-err state val current-state)

       :else
       (throw (Exception. (str "Unexpected event: " evt)))))))

(defn mk-upstream-handler
  [^ChannelGroup channel-group app opts]
  (let [state (atom nil)]
    (reify ChannelUpstreamHandler
      (^void handleUpstream [_ ^ChannelHandlerContext ctx ^ChannelEvent evt]
        (let [current-state ^State @state]
          (try
            (cond
             ;; Handle message events
             (message-event? evt)
             (let [[evt val] (decode (.getMessage ^MessageEvent evt))]
               (send-upstream state evt val current-state))

             (interest-changed-event? evt)
             (send-upstream state :interest-ops nil current-state)

             ;; The bind function is invoked on channel open, ideally
             ;; this would always get invoked. However, there is a
             ;; possibility that opening the socket channel
             ;; fails. That case isn't handled right now.
             (channel-open-event? evt)
             (let [ch (.getChannel evt)]
               ;; First, track the channel in the channel group
               (.add channel-group ch)
               ;; Now, initialize the state with the channel
               (reset! state (initial-state ch))
               (let [upstream (app (mk-downstream-fn state))]
                 (swap! state #(assoc % :upstream upstream))))

             (channel-connected-event? evt)
             (let [ch-info (-> current-state .ch channel-info)]
               (send-upstream state :open ch-info current-state))

             (channel-disconnected-event? evt)
             (send-upstream state :close nil current-state)

             (exception-event? evt)
             (handle-err state (.getCause evt) current-state))
            (catch Exception err
              (handle-err state err @state))))))))

;; ==== Handlers

(defrecord State
    [upstream
     downstream
     aborting?
     writable?
     event-lock
     state-queue
     message-queue])

(defrecord NettyState
    [ch
     last-write])

(defn- mk-initial-state
  [dn]
  (State.
   nil             ;; upstream
   dn              ;; downstream
   false           ;; aborting?
   true            ;; writable?
   (atom true)     ;; event-lock
   (LinkedList.)   ;; state-queue
   (LinkedList.))) ;; message-queue

(defn- mk-initial-netty-state
  []
  (NettyState. nil nil))

(declare handle-err)

