           (recur timeout)
           (count chan-buf))))))
 
+(defmacro ^:private listen
+  "Helper for register-listener below that handles the mode
+   selection (real/green thread)"
+  [& {:keys [mode pred listener-fn events end-chan]}]
+  (let [loop-fn (case mode :blocking 'thread :compute 'go-loop)
+        alt-fn (case mode :blocking 'alt!! :compute 'alt!)]
+    `(~loop-fn []
+      (if-let [event# (~alt-fn [~events ~end-chan] ([v#] v#))]
+        (do
+          (when (~pred event#)
+            (~listener-fn event#))
+          (recur))
+        (do (close! ~end-chan)
+            (close! ~events))))))
 
 (s/defn register-listener :- Channel
   "Creates a GO loop to direct events to a function.

   ec - An event channel, created with new-event-channel.
   f - user provided function that will be called when an event arrives on the event channel.
   pred - a predicate to test events. The user function will only be run if this predicate passes.
   shutdown-fn - an optional function to run at system shutdown time. May be nil."
  [{m :mult :as ec} :- EventChannel
   listener-fn :- (=> s/Any Event)
   pred :- (=> s/Bool Event)
   shutdown-fn :- (s/maybe (=> s/Any))]
  (let [events (chan)]
    (tap m events)
    (let [end-chan (chan)]
      (go-loop []
        (if-let [event (alt! [events end-chan] ([v] v))]
          (do
            (when (pred event)
              (listener-fn event))
            (recur))
          (do
            (close! end-chan)
            (close! events))))
      (.addShutdownHook (Runtime/getRuntime)
                        (Thread. #(do (a/close! end-chan)
                                      (when (fn? shutdown-fn)
                                        (shutdown-fn)))))
      end-chan)))

(s/defn drain :- [s/Any]
  "Extract elements from a channel into a lazy-seq.

  (:require [ctia.lib.time :as time]
            [ctia.schemas.common :as c]
            [ctia.schemas.verdict :as v]
            [clojure.core.async :as a :refer [go-loop alt! alt!! chan tap thread]]
            [schema.core :as s :refer [=>]])
  (:import [clojure.core.async.impl.protocols Channel]
           [clojure.core.async.impl.buffers FixedBuffer]
