                (.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)]
      (assoc component :session-factory session-factory)))

  (stop [component]
    (taoensso.timbre/info "Stopping HornetQ")
    component))

(defn hornetq [addr]

  (let [tc (TransportConfiguration. (.getName NettyConnectorFactory))
        locator (HornetQClient/createServerLocatorWithoutHA (into-array [tc]))
        session-factory (.createSessionFactory locator)
        session (.createTransactedSession session-factory)
        queue (:hornetq/queue-name task)
        consumer (.createConsumer session queue "" 0 64000 false)]
    (.start session)
