     (catch Exception e
       (info e))))
 
-(defmethod extensions/n-messages-remaining HornetQClusteredConnection
+(defmethod extensions/n-messages-remaining HornetQConnection
   [queue session queue-name]
   (let [query (.queueQuery session (SimpleString. queue-name))]
     (.getMessageCount query)))
 
-(defmethod extensions/n-consumers HornetQClusteredConnection
+(defmethod extensions/n-consumers HornetQConnection
   [queue queue-name]
   (let [session (.createSession (:session-factory queue))
         requestor (ClientRequestor. session "jms.queue.hornetq.management")

    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))

    (extensions/commit-tx queue session)
    (extensions/close-resource queue session)))

(defmethod extensions/produce-message HornetQConnection
  ([queue producer session msg]
     (let [message (.createMessage session true)]
       (.writeBytes (.getBodyBuffer message) msg)
