                   (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)]

  ([state err current-state]
     (handle-err state err current-state false))
  ([state err current-state locked?]
     (when (and current-state
                (not (.aborting? current-state)))
       (let [upstream  (.upstream current-state)]
         (swap-then!
          state
          #(assoc % :aborting? true)
          (fn [^State current-state]
            (close-channel current-state)
            (when upstream
              (try
                (if locked?

       (close-channel current-state)

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

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

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

(defn mk-upstream-handler
  [^ChannelGroup channel-group app opts]
  (let [state    (atom (mk-initial-netty-state))
        app      (handler app opts)
        upstream (app (mk-netty-downstream-fn state))]

    ;; The actual Netty upstream handler.
    (reify ChannelUpstreamHandler
      (^void handleUpstream [_ ^ChannelHandlerContext ctx ^ChannelEvent evt]
        (let [current-state ^State @state]
          (cond
           ;; Handle message events
           (message-event? evt)
           (let [[evt val] (decode (.getMessage ^MessageEvent evt))]
             (upstream evt val))

           (interest-changed-event? evt)
           (if (.isWritable (.getChannel evt))
             (upstream :resume nil)
             (upstream :pause nil))

           ;; 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
             (swap! state #(assoc % :ch ch)))

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

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

           (exception-event? evt)
           (upstream :abort (.getCause evt))))))))
