                (.acknowledge m)
                m)
           rets (doall (take-while (comp not nil?) (repeatedly (:hornetq/batch-size task) f)))]
+      (.commit session)
       (.close consumer)
       (.close session)
+      (.close session-factory)
+      (.close locator)
       rets)))
 
 (defn decompress-segment [segment]

  (let [tc (TransportConfiguration. (.getName NettyConnectorFactory))
        locator (HornetQClient/createServerLocatorWithoutHA (into-array [tc]))
        session-factory (.createSessionFactory locator)
        session (.createSession session-factory)
        queue (:hornetq/queue-name task)
        producer (.createProducer session queue)]
    (.start session)

    (let [tc (TransportConfiguration. (.getName NettyConnectorFactory))
          locator (HornetQClient/createServerLocatorWithoutHA (into-array [tc]))
          session-factory (.createSessionFactory locator)]
      (assoc component
        :locator locator
        :session-factory session-factory)))

  (stop [{:keys [locator session-factory] :as component}]
    (taoensso.timbre/info "Stopping HornetQ")

    (.close session-factory)
    (.close locator)
    
    component))

(defn hornetq [addr]
