        (.writeBytes (.getBodyBuffer message) msg)
        (.send producer message))))
 
-(defmethod extensions/consume-message HornetQClusteredConnection
+(defmethod extensions/consume-message HornetQConnection
   [queue consumer]
   (.receive consumer))
 
-(defmethod extensions/read-message HornetQClusteredConnection
+(defmethod extensions/read-message HornetQConnection
   [queue message]
   (fressian/read (.toByteBuffer (.getBodyBuffer message))))
 
-(defmethod extensions/ack-message HornetQClusteredConnection
+(defmethod extensions/ack-message HornetQConnection
   [queue message]
   (.acknowledge message))
 
-(defmethod extensions/commit-tx HornetQClusteredConnection
+(defmethod extensions/commit-tx HornetQConnection
   [queue session]
   (.commit session))
 
-(defmethod extensions/close-resource HornetQClusteredConnection
+(defmethod extensions/close-resource HornetQConnection
   [queue resource]
   (.close resource))
 

    component))

(defn hornetq [opts]
  (map->HornetQClusteredConnection {:opts opts}))

(defmethod extensions/optimize-concurrently HornetQClusteredConnection
  [queue event]
  (if (= (:onyx/consumption (:onyx.core/task-map event)) :concurrent)
    (do (.close (:session-factory queue))

    (catch Exception e
      (info e))))

(defmethod extensions/n-messages-remaining HornetQConnection
  [queue session queue-name]
  (let [query (.queueQuery session (SimpleString. queue-name))]
    (.getMessageCount query)))

(defmethod extensions/n-consumers HornetQConnection
  [queue queue-name]
  (let [session (.createSession (:session-factory queue))
        requestor (ClientRequestor. session "jms.queue.hornetq.management")
