<commit_before>;; Copyright 2008-2012 Red Hat, Inc, and individual contributors.
;; 
;; This is free software; you can redistribute it and/or modify it
;; under the terms of the GNU Lesser General Public License as
;; published by the Free Software Foundation; either version 2.1 of
;; the License, or (at your option) any later version.
;; 
;; This software is distributed in the hope that it will be useful,
;; but WITHOUT ANY WARRANTY; without even the implied warranty of
;; MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
;; Lesser General Public License for more details.
;; 
;; You should have received a copy of the GNU Lesser General Public
;; License along with this software; if not, write to the Free
;; Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA
;; 02110-1301 USA, or see the FSF site: http://www.fsf.org.

(ns immutant.messaging.core
  (:use [immutant.utilities :only (at-exit)])
  (:import (javax.jms DeliveryMode Destination Queue Session Topic))
  (:require [immutant.registry          :as lookup]
            [immutant.messaging.hornetq :as hornetq]
            [immutant.xa.transaction    :as tx]
            [clojure.tools.logging      :as log]))

;;; The name of the JBoss connection factory
(def factory-name "jboss.naming.context.java.ConnectionFactory")
;;; Default subscriber name, which can't be null, apparently
(def default-subscriber-name "default")


;;; Thread-local connection set
(def ^{:private true, :dynamic true} *connections* nil)
;;; Thread-local current connection
(def ^{:private true, :dynamic true} ^javax.jms.XAConnection *connection* nil)
;;; Thread-local map of transactions to sessions
(def ^{:private true, :dynamic true} *sessions* nil)

(defn ^{:private true} remote-connection? [opts]
  (:host opts))

(defn ^{:private true} connection-key [opts]
  (map #(% opts) [:host :port :username :password]))

;; ignore reflection here since it only occurs once at compile time
(let [local-connection-factory
      (if-let [reference-factory (lookup/fetch factory-name)]
        (let [reference (.getReference reference-factory)]
          (try
            (.getInstance reference)
            (finally (at-exit #(.release reference)))))
        (do
          (log/warn "Unable to obtain JMS Connection Factory - assuming we are outside the container")
          (hornetq/connection-factory)))]
  (defn connection-factory
    [opts]
    (if (remote-connection? opts)
      (hornetq/connection-factory opts)
      local-connection-factory)))

(defn queue-name? [^String name]
  (not (nil? (re-find #"^.?queue" name))))

(defn queue? [queue]
  (or (isa? (class queue) Queue) (queue-name? queue)))

(defn topic-name? [^String name]
  (not (nil? (re-find #"^.?topic" name))))

(defn topic? [topic]
  (or (isa? (class topic) Topic) (topic-name? topic)))

(defn wash-publish-options
  "Wash publish options relative to default values from a producer"
  [opts ^javax.jms.MessageProducer producer]
  {:delivery (if (contains? opts :persistent)
               (if (:persistent opts)
                 DeliveryMode/PERSISTENT
                 DeliveryMode/NON_PERSISTENT)
               (.getDeliveryMode producer))
   :priority (let [priorities {:low 1, :normal 4, :high 7, :critical 9}]
               (or (priorities (:priority opts))
                   (:priority opts)
                   (.getPriority producer)))
   :ttl (or (:ttl opts) (.getTimeToLive producer))})

(defn set-attributes!
  "Sets attributes on a JMS message. Returns message."
  [^javax.jms.Message message attributes]
  (doseq [[attr v] attributes]
    (condp = attr
      :correlation-id (.setJMSCorrelationID message v)
      :reply-to       (.setJMSReplyTo message v)
      :type           (.setJMSType message v)
      nil))
  message)

(defn set-properties!
  "Set user-defined properties on a JMS message. Returns message"
  [^javax.jms.Message message properties]
  (doseq [[k,v] properties]
    (let [key (name k)]
      (cond
       (integer? v) (.setLongProperty message key (long v))
       (float? v) (.setDoubleProperty message key (double v))
       (instance? Boolean v) (.setBooleanProperty message key v)
       :else (.setStringProperty message key (str v)))))
  message)

(defn get-properties
  "Extract properties from message into a map, turning the JMS
   property names into keywords unless the :keywords option is false"
  [message & {:keys [keywords] :or {keywords true}}]
  (into {} (for [k (enumeration-seq (.getPropertyNames message))]
             [(if keywords (keyword k) k) (.getObjectProperty message k)])))

(defn create-destination [^Session session name-or-dest]
  (cond
   (isa? (class name-or-dest) Destination) name-or-dest
   (queue-name? name-or-dest) (.createQueue session name-or-dest)
   (topic-name? name-or-dest) (.createTopic session name-or-dest)
   :else (throw (Exception. "Illegal destination name"))))

(defn stop-destination [name]
  (if-let [manager
           (lookup/fetch "jboss.messaging.default.jms.manager")]
    (try
      (if (queue-name? name)
        (.destroyQueue manager name)
        (.destroyTopic manager name))
      (log/info "Stopped" name)
      (catch Throwable e
        (log/warn e)))))

(defn start-queue [name & {:keys [durable selector] :or {durable false selector ""}}]
  (if-let [manager (lookup/fetch "jboss.messaging.default.jms.manager")]
    (do (.createQueue manager false name selector durable (into-array String []))
        (at-exit #(stop-destination name)))
    (throw (Exception. (str "Unable to start queue, " name)))))

(defn start-topic [name & opts]
  (if-let [manager (lookup/fetch "jboss.messaging.default.jms.manager")]
    (do (.createTopic manager false name (into-array String []))
        (at-exit #(stop-destination name)))<commit_msg>Durable should be the default, but it has no effect [IMMUTANT-118]<commit_after>;; Copyright 2008-2012 Red Hat, Inc, and individual contributors.
;; 
;; This is free software; you can redistribute it and/or modify it
;; under the terms of the GNU Lesser General Public License as
;; published by the Free Software Foundation; either version 2.1 of
;; the License, or (at your option) any later version.
;; 
;; This software is distributed in the hope that it will be useful,
;; but WITHOUT ANY WARRANTY; without even the implied warranty of
;; MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
;; Lesser General Public License for more details.
;; 
;; You should have received a copy of the GNU Lesser General Public
;; License along with this software; if not, write to the Free
;; Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA
;; 02110-1301 USA, or see the FSF site: http://www.fsf.org.

(ns immutant.messaging.core
  (:use [immutant.utilities :only (at-exit)])
  (:import (javax.jms DeliveryMode Destination Queue Session Topic))
  (:require [immutant.registry          :as lookup]
            [immutant.messaging.hornetq :as hornetq]
            [immutant.xa.transaction    :as tx]
            [clojure.tools.logging      :as log]))

;;; The name of the JBoss connection factory
(def factory-name "jboss.naming.context.java.ConnectionFactory")
;;; Default subscriber name, which can't be null, apparently
(def default-subscriber-name "default")


;;; Thread-local connection set
(def ^{:private true, :dynamic true} *connections* nil)
;;; Thread-local current connection
(def ^{:private true, :dynamic true} ^javax.jms.XAConnection *connection* nil)
;;; Thread-local map of transactions to sessions
(def ^{:private true, :dynamic true} *sessions* nil)

(defn ^{:private true} remote-connection? [opts]
  (:host opts))

(defn ^{:private true} connection-key [opts]
  (map #(% opts) [:host :port :username :password]))

;; ignore reflection here since it only occurs once at compile time
(let [local-connection-factory
      (if-let [reference-factory (lookup/fetch factory-name)]
        (let [reference (.getReference reference-factory)]
          (try
            (.getInstance reference)
            (finally (at-exit #(.release reference)))))
        (do
          (log/warn "Unable to obtain JMS Connection Factory - assuming we are outside the container")
          (hornetq/connection-factory)))]
  (defn connection-factory
    [opts]
    (if (remote-connection? opts)
      (hornetq/connection-factory opts)
      local-connection-factory)))

(defn queue-name? [^String name]
  (not (nil? (re-find #"^.?queue" name))))

(defn queue? [queue]
  (or (isa? (class queue) Queue) (queue-name? queue)))

(defn topic-name? [^String name]
  (not (nil? (re-find #"^.?topic" name))))

(defn topic? [topic]
  (or (isa? (class topic) Topic) (topic-name? topic)))

(defn wash-publish-options
  "Wash publish options relative to default values from a producer"
  [opts ^javax.jms.MessageProducer producer]
  {:delivery (if (contains? opts :persistent)
               (if (:persistent opts)
                 DeliveryMode/PERSISTENT
                 DeliveryMode/NON_PERSISTENT)
               (.getDeliveryMode producer))
   :priority (let [priorities {:low 1, :normal 4, :high 7, :critical 9}]
               (or (priorities (:priority opts))
                   (:priority opts)
                   (.getPriority producer)))
   :ttl (or (:ttl opts) (.getTimeToLive producer))})

(defn set-attributes!
  "Sets attributes on a JMS message. Returns message."
  [^javax.jms.Message message attributes]
  (doseq [[attr v] attributes]
    (condp = attr
      :correlation-id (.setJMSCorrelationID message v)
      :reply-to       (.setJMSReplyTo message v)
      :type           (.setJMSType message v)
      nil))
  message)

(defn set-properties!
  "Set user-defined properties on a JMS message. Returns message"
  [^javax.jms.Message message properties]
  (doseq [[k,v] properties]
    (let [key (name k)]
      (cond
       (integer? v) (.setLongProperty message key (long v))
       (float? v) (.setDoubleProperty message key (double v))
       (instance? Boolean v) (.setBooleanProperty message key v)
       :else (.setStringProperty message key (str v)))))
  message)

(defn get-properties
  "Extract properties from message into a map, turning the JMS
   property names into keywords unless the :keywords option is false"
  [message & {:keys [keywords] :or {keywords true}}]
  (into {} (for [k (enumeration-seq (.getPropertyNames message))]
             [(if keywords (keyword k) k) (.getObjectProperty message k)])))

(defn create-destination [^Session session name-or-dest]
  (cond
   (isa? (class name-or-dest) Destination) name-or-dest
   (queue-name? name-or-dest) (.createQueue session name-or-dest)
   (topic-name? name-or-dest) (.createTopic session name-or-dest)
   :else (throw (Exception. "Illegal destination name"))))

(defn stop-destination [name]
  (if-let [manager
           (lookup/fetch "jboss.messaging.default.jms.manager")]
    (try
      (if (queue-name? name)
        (.destroyQueue manager name)
        (.destroyTopic manager name))
      (log/info "Stopped" name)
      (catch Throwable e
        (log/warn e)))))

(defn start-queue [name & {:keys [durable selector] :or {durable true selector ""}}]
  (if-let [manager (lookup/fetch "jboss.messaging.default.jms.manager")]
    (do (.createQueue manager false name selector durable (into-array String []))
        (at-exit #(stop-destination name)))
    (throw (Exception. (str "Unable to start queue, " name)))))

(defn start-topic [name & opts]
  (if-let [manager (lookup/fetch "jboss.messaging.default.jms.manager")]
    (do (.createTopic manager false name (into-array String []))
        (at-exit #(stop-destination name)))
{'fraction_non_alphanumeric': 0.12749483826565727, 'fraction_numerical': 0.004559532002752925, 'mean_word_length': 3.344170403587444, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7659051', 'n_tokens_mistral': 3581, 'n_tokens_neox': 3408, 'n_words': 1209}