                       (.addPoller (@consumer :zloop)
                                   (.getItem poller 0)
                                   (q/zloop-handler (fn [] ((handle-incoming consumer)) (int 0)))
-                                  (Object.))
+                                  {})
                       (send! protocol socket [q/MSG-READY]))]
     (log/info "[consumer] started consumer for" method)
     (swap! consumer assoc
            :handlers {q/MSG-REQ (handle-request consumer handler)}
            :initialize initialize)
     ((@consumer :initialize))
-    (send (@consumer :consumer) (fn [_] (.start zloop)))
+    ; mysterious hack required to start the zloop
+    (.addTimer zloop 1000 0 (q/zloop-handler #(do (Thread/sleep 1) 0)) {})
+
+    (.start (Thread. #(.start zloop)))
     consumer))
 
 (defn start

        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)
                        (if (> next-liveness 0)
                          (ping)
                          (@reconnecter)))))]
    (swap! consumer assoc :handlers (merge (@consumer :handlers) {q/MSG-PONG pong}))
    (.addTimer (@consumer :zloop) heartbeat-interval 0
               (q/zloop-handler #(do (heartbeat) 0)) (Object.))))

(defn- handle-request
  [consumer handler]


(defprotocol Protocol
  (send! [this socket messages]))

(defrecord MessageProtocol [method]
  Protocol
