           response (handler (nippy/thaw request))]
       (send! protocol socket [q/MSG-REP address (nippy/freeze response)])
       (send! protocol socket [q/MSG-READY]))))
- 
+
 (defn- handle-incoming
   [consumer]
   (fn []

    ((@consumer :initialize))
    (.start (Thread. #(.start zloop)))
    consumer))
  
(defn start
  [method handler]
  (with-heartbeat (create-consumer method handler)))

        pong (fn [_] (do (reset! reconnecter (create-reconnecter consumer))
                        (swap! consumer assoc :liveness heartbeat-liveness)))
        ping #(do (send! (@consumer :protocol) (@consumer :socket) [q/MSG-PING]))
        heartbeat (fn []
                    (when (contains? @consumer :socket)
                      (let [next-liveness (dec (get @consumer :liveness (inc heartbeat-liveness)))]
                        (swap! consumer assoc :liveness next-liveness)
