<commit_before>(ns apio.cli
  (:import (java.lang Thread)
           (apio Queue))
  (:require [apio.core :as core]
            [apio.concurrency.threading :as thr]
            [apio.concurrency.semaphore :as sem]
            [apio.concurrency.queue :as q]
            [apio.tasks :as tasks]
            [apio.brokers.core :as brk])
  (:gen-class))

(defn dispatch-one-task
  [pool task semaphore]
  (sem/acquire semaphore)
  (let [task-wrapper (fn []
                       (task)
                       (sem/release semaphore))]
    (thr/spawn pool task-wrapper)))

(defn task-dispatcher
  [^Queue queue]
  (let [numworkers  (core/max-workers)
        pool        (thr/start-pool numworkers)
        semaphore   (sem/semaphore numworkers)]
    (loop []
      (let [task (q/rcv queue)]
          (do (dispatch-one-task pool task semaphore) (recur))))
    (thr/shutdown-pool pool)))

(defn messages-dispatcher
  "Broker callback for receive messages."
  [^Queue queue]
  (let [dispatcher (fn [^String message & args]
                     (let [unit (tasks/exec-unit-from-message message)]
                       (if (not (nil? unit))
                         (q/snd queue unit)
                         (println "- Message:" message " does not corresponds to any task."))))]
    dispatcher))

(defn -main
  [path & args]
  (core/with-config path
    (let [queue   (q/queue (core/max-prefetch))
          thr     (thr/thread #(task-dispatcher queue))]

      ;; Add hook for keyboard interruption (sigint) and
      ;; properly close RabbitMQ connection.
      ;; (let [hook (proxy [Thread] [] (run [] (mq/finish-connection mq-conn)))]
      ;;   (.addShutdownHook (Runtime/getRuntime) hook))

      (brk/with-connection-and-handler (messages-dispatcher queue)
        (thr/sleep 1000)
        (tasks/send-task "hellow-world" 1 2 3)
        (thr/sleep 1000)
        (tasks/send-task "hellow-world" 4 5 6)
        (thr/sleep 1000)
        (tasks/send-task "hellow-world2" 4 5 6)
        (thr/join thr)))))
<commit_msg>Remove unused imports from cli module.<commit_after>(ns apio.cli
  (:import apio.Queue)
  (:require [apio.core :as core]
            [apio.concurrency.threading :as thr]
            [apio.concurrency.semaphore :as sem]
            [apio.concurrency.queue :as q]
            [apio.tasks :as tasks]
            [apio.brokers.core :as brk])
  (:gen-class))

(defn dispatch-one-task
  [pool task semaphore]
  (sem/acquire semaphore)
  (let [task-wrapper (fn []
                       (task)
                       (sem/release semaphore))]
    (thr/spawn pool task-wrapper)))

(defn task-dispatcher
  [^Queue queue]
  (let [numworkers  (core/max-workers)
        pool        (thr/start-pool numworkers)
        semaphore   (sem/semaphore numworkers)]
    (loop []
      (let [task (q/rcv queue)]
          (do (dispatch-one-task pool task semaphore) (recur))))
    (thr/shutdown-pool pool)))

(defn messages-dispatcher
  "Broker callback for receive messages."
  [^Queue queue]
  (let [dispatcher (fn [^String message & args]
                     (let [unit (tasks/exec-unit-from-message message)]
                       (if (not (nil? unit))
                         (q/snd queue unit)
                         (println "- Message:" message " does not corresponds to any task."))))]
    dispatcher))

(defn -main
  [path & args]
  (core/with-config path
    (let [queue   (q/queue (core/max-prefetch))
          thr     (thr/thread #(task-dispatcher queue))]

      ;; Add hook for keyboard interruption (sigint) and
      ;; properly close RabbitMQ connection.
      ;; (let [hook (proxy [Thread] [] (run [] (mq/finish-connection mq-conn)))]
      ;;   (.addShutdownHook (Runtime/getRuntime) hook))

      (brk/with-connection-and-handler (messages-dispatcher queue)
        (thr/sleep 1000)
        (tasks/send-task "hellow-world" 1 2 3)
        (thr/sleep 1000)
        (tasks/send-task "hellow-world" 4 5 6)
        (thr/sleep 1000)
        (tasks/send-task "hellow-world2" 4 5 6)
        (thr/join thr)))))

{'fraction_non_alphanumeric': 0.13540372670807455, 'fraction_numerical': 0.01093167701863354, 'mean_word_length': 2.070938215102975, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, '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': '5272328', 'n_tokens_mistral': 1372, 'n_tokens_neox': 1289, 'n_words': 334}