<commit_before>(ns apio.brokers.rabbitmq
  (:require [langohr.core :as rmq]
            [langohr.channel :as lch]
            [langohr.queue :as lq]
            [langohr.consumers :as lc]
            [langohr.basic :as lb]
            [apio.core :as core]
            [apio.util :as util])
  (:gen-class))

(def ^{:const true} default-exchange-name "")
(def ^{:const true} default-queue-name "niwi.be.queue.test")
(def ^{:dynamic true} *connection*)
(def ^{:dynamic true} *channel*)

(defn config->rabbitmq-settings
  "Read current configuration and build
  rabbitmq connection settings."
  []
  (let [conf      (core/current-config)
        conf      (-> conf :broker)
        settings  (atom {})]

    (if (not (nil? (:vhost conf)))
      (swap! settings assoc :vhost (:vhost conf))
      (swap! settings assoc :vhost (:vhost rmq/*default-config*)))

    (if (not (nil? (:host conf)))
      (swap! settings assoc :host (:host conf))
      (swap! settings assoc :host (:host rmq/*default-config*)))

    (if (not (nil? (:port conf)))
      (swap! settings assoc :port (util/parse-int (:port conf)))
      (swap! settings assoc :port (:port rmq/*default-config*)))

    (if (not (nil? (:username conf)))
      (swap! settings assoc :username (:username conf))
      (swap! settings assoc :username (:username rmq/*default-config*)))

    (if (not (nil? (:password conf)))
      (swap! settings assoc :password (:password conf))
      (swap! settings assoc :password (:password rmq/*default-config*)))
    @settings))

(defn- wrap-message-handler
  "Private method that create lexycal scope closure
  wrapping a message handler for convert a bytes payload
  into utf-8 encoded string."
  [handler]
  (let [wrapper (fn [ch, metadata, ^bytes payload]
                  (handler (String. payload "UTF-8") metadata))]
    wrapper))

(defn initialize-connection
  []
  (let [conf (config->rabbitmq-settings)]
    (try
      (rmq/connect conf)
      (catch Exception e
        (do
          (.printStackTrace e)
          (System/exit 1))))))

(defn initialize-channel [] (lch/open *connection*))

(defn attach-message-handler
  [handler]
  (let [wrapped-handler (wrap-message-handler handler)]
    (lq/declare *channel* default-queue-name :exclusive false :auto-delete true)
    (lc/subscribe *channel* default-queue-name wrapped-handler :auto-ack true)))

(defmacro with-connection
  [& body]
  `(if (not (bound? #'*connection*))
     ;; If connection vars are bound
     ;; connect, execute body and disconnect.
     (binding [*connection* (initialize-connection)]
       (binding [*channel*    (initialize-channel)]
         ~@body
         (rmq/close *channel*)
         (rmq/close *connection*)))

     ;; Else, only execute body but does not disconnect.
     (do ~@body)))

(defn deliver-message
  [^String message]
  (with-connection
    (lb/publish *channel* default-exchange-name default-queue-name message
                :content-type "text/plain" :type "greetings.hi")))

<commit_msg>Print custom message when rabbitmq is down and connection exception is raised.<commit_after>(ns apio.brokers.rabbitmq
  (:require [langohr.core :as rmq]
            [langohr.channel :as lch]
            [langohr.queue :as lq]
            [langohr.consumers :as lc]
            [langohr.basic :as lb]
            [apio.concurrency.threading :as thr]
            [apio.core :as core]
            [apio.util :as util])
  (:gen-class))

(def ^{:const true} default-exchange-name "")
(def ^{:const true} default-queue-name "niwi.be.queue.test")
(def ^{:dynamic true} *connection*)
(def ^{:dynamic true} *channel*)

(defn config->rabbitmq-settings
  "Read current configuration and build
  rabbitmq connection settings."
  []
  (let [conf      (core/current-config)
        conf      (-> conf :broker)
        settings  (atom {})]

    (if (not (nil? (:vhost conf)))
      (swap! settings assoc :vhost (:vhost conf))
      (swap! settings assoc :vhost (:vhost rmq/*default-config*)))

    (if (not (nil? (:host conf)))
      (swap! settings assoc :host (:host conf))
      (swap! settings assoc :host (:host rmq/*default-config*)))

    (if (not (nil? (:port conf)))
      (swap! settings assoc :port (util/parse-int (:port conf)))
      (swap! settings assoc :port (:port rmq/*default-config*)))

    (if (not (nil? (:username conf)))
      (swap! settings assoc :username (:username conf))
      (swap! settings assoc :username (:username rmq/*default-config*)))

    (if (not (nil? (:password conf)))
      (swap! settings assoc :password (:password conf))
      (swap! settings assoc :password (:password rmq/*default-config*)))
    @settings))

(defn- wrap-message-handler
  "Private method that create lexycal scope closure
  wrapping a message handler for convert a bytes payload
  into utf-8 encoded string."
  [handler]
  (let [wrapper (fn [ch, metadata, ^bytes payload]
                  (handler (String. payload "UTF-8") metadata))]
    wrapper))

(defn initialize-connection
  []
  (let [conf (config->rabbitmq-settings)]
    (try
      (rmq/connect conf)
      (catch java.net.ConnectException e
        (do
          (.printStackTrace e)
          (thr/sleep 300)

          (println "\nSeems your connection parameters are wrong"
                   "or rabbitmq service is down!")
          (System/exit 1))))))

(defn initialize-channel [] (lch/open *connection*))

(defn attach-message-handler
  [handler]
  (let [wrapped-handler (wrap-message-handler handler)]
    (lq/declare *channel* default-queue-name :exclusive false :auto-delete true)
    (lc/subscribe *channel* default-queue-name wrapped-handler :auto-ack true)))

(defmacro with-connection
  [& body]
  `(if (not (bound? #'*connection*))
     ;; If connection vars are bound
     ;; connect, execute body and disconnect.
     (binding [*connection* (initialize-connection)]
       (binding [*channel*    (initialize-channel)]
         ~@body
         (rmq/close *channel*)
         (rmq/close *connection*)))

     ;; Else, only execute body but does not disconnect.
     (do ~@body)))

(defn deliver-message
  [^String message]
  (with-connection
    (lb/publish *channel* default-exchange-name default-queue-name message
                :content-type "text/plain" :type "greetings.hi")))


{'fraction_non_alphanumeric': 0.14566677326511032, 'fraction_numerical': 0.001439078989446754, 'mean_word_length': 2.8996259351620948, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, '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': '7124025', 'n_tokens_mistral': 1992, 'n_tokens_neox': 1901, 'n_words': 578}