       (reset! event-lock true))))
 
 (defn- handle-interest-ops
-  [state upstream]
+  [state evt upstream]
   (let [current-state @state
-        writable?     (.writable? current-state)
-        ch            (.ch current-state)]
-    (when (not= writable? (.isWritable ch))
-      (swap! state #(assoc % :writable? (not writable?)))
-      (upstream (if writable? :pause :resume) nil))))
+        was-writable? (.writable? current-state)
+        now-writable? (= :resume evt)]
+
+    (when (not= was-writable? now-writable?)
+      (swap! state #(assoc % :writable? now-writable?))
+      (upstream evt nil))))
 
 ;; Handles sending messages upstream in a sane and thread-safe way.
 ;;

                  (send-upstream state :abort err current-state))
                (catch Exception _)))))))))

(defn- mk-downstream-fn
  [state]
  (fn [evt val]
    (let [current-state ^State @state]
      (when-not (.upstream current-state)
        (throw (Exception. "Not callbable yet")))

      (cond
       (= :message evt)
       (let [ch (.ch current-state)]

        (loop []
          (when-let [[evt val] (poll-queue current-state)]
            (cond
             (#{:pause :resume} evt)
             (handle-interest-ops state evt upstream)

             (= :schedule evt)
             (val)
