 
 (declare close-listener)
 
-(defrecord Listener [connection handler state]
+(defrecord Listener [connection handler state future]
   java.io.Closeable
   (close [this] (close-listener this)))
 

         state-atom#   (atom ~initial-state)
         {:as conn# in# :in} (conns/make-new-connection
                              (assoc (conns/conn-spec ~conn-spec)
                                :listener? true))]

     (future-call ; Thread to long-poll for messages
      (bound-fn []
        (while true ; Closes when conn closes
          (let [reply# (protocol/get-unparsed-reply in# {})]
            (try
              (@handler-atom# reply# @state-atom#)
              (catch Throwable t#
                (timbre/error t# "Listener handler exception")))))))

     (protocol/with-context conn# ~@body
       (protocol/execute-requests (not :get-replies) nil))
     (Listener. conn# handler-atom# state-atom#)))

(defmacro with-open-listener
  "Evaluates body within the context of given listener's preexisting persistent

  `(protocol/with-context (:connection ~listener) ~@body
     (protocol/execute-requests (not :get-replies) nil)))

(defn close-listener [listener]
  (conns/close-conn (:connection listener))
  (future-cancel (:future listener)))

(defmacro with-new-pubsub-listener
  "A wrapper for `with-new-listener`.
