         (.send producer message)
         (.close producer))
       (.commit session)
-      (.close session))))
-
-(defmethod l-ext/start-lifecycle? :hornetq/read-segments
-  [_ {:keys [hornetq/session onyx.core/ingress-queues onyx.core/task-map]}]
-  {:onyx.core/start-lifecycle?
-   (if (= (:task/consumption task-map) :sequential)
-     (let [query (.queueQuery session (SimpleString. (:hornetq/queue-name task-map)))]
-       (zero? (.getConsumerCount query)))
-     true)})
-
-(defmethod l-ext/inject-lifecycle-resources :hornetq/read-segments
-  [_ {:keys [onyx.core/task-map] :as pipeline-data}]
-  (let [config {"host" (:hornetq/host task-map) "port" (:hornetq/port task-map)}
-        tc (TransportConfiguration. (.getName NettyConnectorFactory) config)
-        locator (HornetQClient/createServerLocatorWithoutHA (into-array [tc]))]
-    (let [session-factory (.createSessionFactory locator)
-          session (.createTransactedSession session-factory)]
-      (merge pipeline-data
-             {:onyx.core/holding-pen (atom {})
-              :hornetq/locator locator
-              :hornetq/session-factory session-factory
-              :hornetq/session session
-              :hornetq/consumer (.createConsumer session (:hornetq/queue-name task-map))}))))
-
-(defmethod l-ext/close-temporal-resources :hornetq/read-segments
-  [_ pipeline-data]
-  (.commit (:hornetq/session pipeline-data))
-  pipeline-data)
-
-(defmethod l-ext/close-lifecycle-resources :hornetq/read-segments
-  [_ pipeline-data]
-  (.close (:hornetq/consumer pipeline-data))
-  (.close (:hornetq/session pipeline-data))
-  (.close (:hornetq/session-factory pipeline-data))
-  (.close (:hornetq/locator pipeline-data))
-  pipeline-data)
-
-(defmethod p-ext/read-batch [:input :hornetq]
-  [event] (read-batch-shim event))
-
-(defmethod p-ext/decompress-batch [:input :hornetq]
-  [event] (decompress-batch-shim event))
-
-(defmethod p-ext/apply-fn [:input :hornetq]
-  [event] (apply-fn-in-shim event))
-
-(defmethod p-ext/ack-message [:input :hornetq]
-  [event]
-  (let [state @(:onyx.core/holding-pen event)
-        message (get state (:onyx.core/message-id event))]
-    (.acknowledge message)
-    (swap! (:onyx.core/holding-pen event) dissoc (:onyx.core/message-id event))))
-
-(defmethod p-ext/replay-message [:input :hornetq]
-  [event]
-  (println "Releasing HornetQ message"))
-
-(defmethod p-ext/apply-fn [:output :hornetq]
-  [event] (apply-fn-out-shim event))
-
-(defmethod p-ext/compress-batch [:output :hornetq]
-  [event] (compress-batch-shim event))
-
-(defmethod p-ext/write-batch [:output :hornetq]
-  [event] (write-batch-shim event))
-
-(defmethod l-ext/inject-lifecycle-resources :hornetq/write-segments
-  [_ {:keys [onyx.core/task-map] :as pipeline}]
-  (let [config {"host" (:hornetq/host task-map) "port" (:hornetq/port task-map)}
-        tc (TransportConfiguration. (.getName NettyConnectorFactory) config)
-        locator (HornetQClient/createServerLocatorWithoutHA (into-array [tc]))]
-
-    (when (= (:onyx/consumption task-map) :sequential)
-      (.setConsumerWindowSize locator 0))
-
-    (let [session-factory (.createSessionFactory locator)]
-      {:hornetq/locator locator
-       :hornetq/session-factory session-factory})))
-
-(defmethod l-ext/close-temporal-resources :hornetq/write-segments
-  [_ pipeline-data]
-  (.close (:hornetq/producer pipeline-data))
-  (.close (:hornetq/session pipeline-data))
+      (.close session)))
   {})
 
-(defmethod l-ext/close-lifecycle-resources :hornetq/write-segments
-  [_ pipeline-data]
-  (.close (:hornetq/session-factory pipeline-data))
-  (.close (:hornetq/locator pipeline-data))
-  {})
-
-(defmethod p-ext/seal-resource [:output :hornetq]
-  [pipeline-data]
-  (seal-resource-shim pipeline-data)
-  {})
-

               (conj rets segment)
               (recur f n (conj rets segment)))))))))

(defn read-batch [consumer task]
  (let [queue-name (:hornetq/queue-name task)
        timeout (or (:onyx/batch-timeout task) 1000)]
    (let [f (fn [] {:message (.receive consumer timeout)})
          rets (filter (comp not nil? :message)
                       (take-segments f (:onyx/batch-size task)))]
      {:onyx.core/batch (or rets [])})))

(defn decompress-segment [segment]
  (fressian/read (.toByteBuffer (.getBodyBuffer segment))))

(defn compress-segment [segment]
  (.array (fressian/write segment)))

(defn write-batch [session-factory task compressed]
  (let [session (.createTransactedSession session-factory)
        queue (:hornetq/queue-name task)
        producer (.createProducer session queue)]
    (.start session)
    (doseq [x compressed]

     :hornetq/producer producer
     :onyx.core/written? true}))

(defmethod l-ext/inject-lifecycle-resources :hornetq/write-segments
  [_ {:keys [onyx.core/task-map] :as pipeline}]
  (let [config {"host" (:hornetq/host task-map) "port" (:hornetq/port task-map)}
        tc (TransportConfiguration. (.getName NettyConnectorFactory) config)
        locator (HornetQClient/createServerLocatorWithoutHA (into-array [tc]))
        session-factory (.createSessionFactory locator)]
    {:hornetq/locator locator
     :hornetq/session-factory session-factory}))

(defmethod l-ext/close-temporal-resources :hornetq/write-segments
  [_ event]
  (.close (:hornetq/producer event))
  (.close (:hornetq/session event))
  {})

(defmethod l-ext/close-lifecycle-resources :hornetq/write-segments
  [_ event]
  (.close (:hornetq/session-factory event))
  (.close (:hornetq/locator event))
  {})

(defmethod p-ext/seal-resource [:output :hornetq]
  [{:keys [onyx.core/task-map] :as event}]
  (let [queue-name (:hornetq/queue-name task-map)]
    (let [session (.createTransactedSession (:hornetq/session-factory event))]
      (let [producer (.createProducer session queue-name)
