    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)))
+   shutdown-fn - an optional function to run at system shutdown time. May be nil.
+   mode - can be :compute (for a green thread) or :blocking (for a real thread)"
+  ([ec :- EventChannel
+    listener-fn :- (=> s/Any Event)
+    pred :- (=> s/Bool Event)
+    shutdown-fn :- (s/maybe (=> s/Any))]
+   (register-listener ec listener-fn pred shutdown-fn :compute))
+  ([{m :mult :as ec} :- EventChannel
+    listener-fn :- (=> s/Any Event)
+    pred :- (=> s/Bool Event)
+    shutdown-fn :- (s/maybe (=> s/Any))
+    mode :- (s/enum :compute :blocking)]
+   (let [events (chan)
+         end-chan (chan)]
+     (tap m events)
+     (case mode
+       :blocking
+       (listen :mode :blocking
+               :pred pred
+               :listener-fn listener-fn
+               :events events
+               :end-chan end-chan)
+
+       :compute
+       (listen :mode :compute
+               :pred pred
+               :listener-fn listener-fn
+               :events events
+               :end-chan end-chan))
+     (.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! chan tap]]
            [schema.core :as s :refer [=>]])
  (:import [clojure.core.async.impl.protocols Channel]
           [clojure.core.async.impl.buffers FixedBuffer]

          (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.
