<commit_before>(ns dragonmark.circulate.core
  #+cljs (:require-macros [cljs.core.async.macros :refer [go]]
                          [schema.core :as sc])
  (:require
   [dragonmark.util.core :as du]
   [schema.core :as sc]
   [cognitect.transit :as t]
   #+clj [cljs.analyzer :as cljs-analyzer]
   #+clj [clojure.core.async :as async :refer [go chan timeout]]
   #+cljs [cljs.core.async :as async :refer [chan timeout]]

   #+clj [clojure.core.async.impl.protocols :as impl]
   #+clj [clojure.core.async.impl.channels :as chans]

   #+cljs [cljs.core.async.impl.protocols :as impl]
   #+cljs [cljs.core.async.impl.channels :as chans]
   )

  #+clj (:import
         [clojure.core.async.impl.channels ManyToManyChannel]
         [clojure.core.async.impl.protocols Channel]
         [java.io Writer])
  )

(defonce env-root
  (atom nil))

(def i-got-it
  "Return this from the message handler if the
  answer is going to be sent back to the return channel
  ansynchronously"
  :dragonmark.circulate:i_got_it)

(defn- error-or-answer
  "If the answer is not {:error something} wrap in {:answer value}"
  [value]
  (or
   (:error value)
   {:answer value}))

(defprotocol EnvInfo
  "Carries the environment around"
  (env-info [_] "Gets the environment atom")
  (locate-service [_ service] "locates the service in the delegation chain"))

(defn- process
  "Processes a message sent to the root channel"
  [channel message env]
  (let [return (:_return message)
        command (:_cmd message)
        local? (-> message meta :local boolean)
        answer
        (case command
          "list" {:answer (->> @env :services
                               (filter #(or local? (-> % second :public)))
                               (map first))}

          "add" (if local?
                  (let [old (get-in @env [:services (:service message)])]
                    (when old (go (async/>! old {:_cmd "removed_from_root"})))
                    (swap! env update-in [:services]
                           assoc (:service message)
                           {:channel (:channel message)
                            :public (-> message :public boolean)})
                    {:answer true})
                  {:error "Cannot add a service without a :local in the meta of the message"})

          "remove" (if local?
                     (let [old (get-in @env [:services (:service message)])]
                       (when old
                         (go (async/>! old {:_cmd "removed_from_root"})))
                       (swap! env update-in [:services] dissoc
                              (:service message)))

                     {:error "Cannot add a service without a :local in the meta of the message"})

          "get-service"
          (let [result (-> @env :services (get (:service message)))
                result (if (and result
                                (or (not local?)
                                    (:public result)))
                         (:channel result)
                         nil)]
            (if result
              {:answer result}
              {:error
               (str "Service " (:service message)
                    " not found")}))

          "locate-service"
          (let [result (locate-service channel (:service message))
                result (if (and result
                                (or (not local?)
                                    (:public result)))
                         (:channel result)
                         nil)]
            (if result
              {:answer result}
              {:error
               (str "Service " (:service message)
                    " not found")}))

          "name"
          {:answer (:name @env)}

          "add-handler"
          (if local?
            ((swap!
              env
              assoc-in [:commands (:command message)] (:function message))
             {:answer true})
            {:error "Cannot add handler without a :local in the meta of the message"})

          "remove-handler"
          (if local?
            ((swap!
              env
              update-in [:commands] dissoc (:command message))
             {:answer true})
            {:error "Cannot remove handler without a :local in the meta of the message"})

          "delegate"
          {:answer (:delegate @env)}

          (or
           ;; try local custom functions
           (some-> (get (:commands @env) command)
                   (apply [message env])
                   error-or-answer)

           ;; or send it to the delegate
           (if-let [delegate  (:delegate @env)]
            (do
              (async/put! delegate message)
              i-got-it)
            nil
            )

           {:error
            (str "Unabled to process command: "
                 command)}))]
    (when (and answer
               return
               (not (= answer i-got-it))) (go (async/>! return answer)))))




(defn build-root-channel
  "Builds a root channel (usually put in env-root,
  but can be stand-alone so there can be multiple
  root channels in a single address space... nice for testing.
  The commands parameter is a map of String/function where the
  String is the name of the command and the function is a two
  parameter function (message and env) that returns a value to
  send to the answer channel or :dragonmark.circulate:i_got_it"
  ([] (build-root-channel {} (du/next-guid) nil))
  ([commands] (build-root-channel commands (du/next-guid) nil))
  ([commands name delegate]
     (let [c (chan 5)
           env (atom {:services
                      {'root {:channel c
                              :public true}}

                      :name name

                      :commands commands

                      :delegate delegate})

           chan+env
           (reify
             EnvInfo
             (env-info [_] env)
             (locate-service [_ service]
               (or
                (-> @env :services (get service))
                (if-let [delegate (:delegate env)]
                  (and
                   (satisfies? EnvInfo delegate)
                   (locate-service delegate service)))))

             chans/MMC
             (cleanup [_] (chans/cleanup c))
             (abort [_] (chans/abort c))
<commit_msg>Fixed an issue with locate-service<commit_after>(ns dragonmark.circulate.core
  #+cljs (:require-macros [cljs.core.async.macros :refer [go]]
                          [schema.core :as sc])
  (:require
   [dragonmark.util.core :as du]
   [schema.core :as sc]
   [cognitect.transit :as t]
   #+clj [cljs.analyzer :as cljs-analyzer]
   #+clj [clojure.core.async :as async :refer [go chan timeout]]
   #+cljs [cljs.core.async :as async :refer [chan timeout]]

   #+clj [clojure.core.async.impl.protocols :as impl]
   #+clj [clojure.core.async.impl.channels :as chans]

   #+cljs [cljs.core.async.impl.protocols :as impl]
   #+cljs [cljs.core.async.impl.channels :as chans]
   )

  #+clj (:import
         [clojure.core.async.impl.channels ManyToManyChannel]
         [clojure.core.async.impl.protocols Channel]
         [java.io Writer])
  )

(defonce env-root
  (atom nil))

(def i-got-it
  "Return this from the message handler if the
  answer is going to be sent back to the return channel
  ansynchronously"
  :dragonmark.circulate:i_got_it)

(defn- error-or-answer
  "If the answer is not {:error something} wrap in {:answer value}"
  [value]
  (or
   (:error value)
   {:answer value}))

(defprotocol EnvInfo
  "Carries the environment around"
  (env-info [_] "Gets the environment atom")
  (locate-service [_ service] "locates the service in the delegation chain"))

(defn- process
  "Processes a message sent to the root channel"
  [channel message env]
  (let [return (:_return message)
        command (:_cmd message)
        local? (-> message meta :local boolean)
        answer
        (case command
          "list" {:answer (->> @env :services
                               (filter #(or local? (-> % second :public)))
                               (map first))}

          "add" (if local?
                  (let [old (get-in @env [:services (:service message)])]
                    (when old (go (async/>! old {:_cmd "removed_from_root"})))
                    (swap! env update-in [:services]
                           assoc (:service message)
                           {:channel (:channel message)
                            :public (-> message :public boolean)})
                    {:answer true})
                  {:error "Cannot add a service without a :local in the meta of the message"})

          "remove" (if local?
                     (let [old (get-in @env [:services (:service message)])]
                       (when old
                         (go (async/>! old {:_cmd "removed_from_root"})))
                       (swap! env update-in [:services] dissoc
                              (:service message)))

                     {:error "Cannot add a service without a :local in the meta of the message"})

          "get-service"
          (let [result (-> @env :services (get (:service message)))
                result (if (and result
                                (or (not local?)
                                    (:public result)))
                         (:channel result)
                         nil)]
            (if result
              {:answer result}
              {:error
               (str "Service " (:service message)
                    " not found")}))

          "locate-service"
          (let [result (locate-service channel (:service message))
                result (if (and result
                                (or (not local?)
                                    (:public result)))
                         (:channel result)
                         nil)]
            (if result
              {:answer result}
              {:error
               (str "Service " (:service message)
                    " not found")}))

          "name"
          {:answer (:name @env)}

          "add-handler"
          (if local?
            ((swap!
              env
              assoc-in [:commands (:command message)] (:function message))
             {:answer true})
            {:error "Cannot add handler without a :local in the meta of the message"})

          "remove-handler"
          (if local?
            ((swap!
              env
              update-in [:commands] dissoc (:command message))
             {:answer true})
            {:error "Cannot remove handler without a :local in the meta of the message"})

          "delegate"
          {:answer (:delegate @env)}

          (or
           ;; try local custom functions
           (some-> (get (:commands @env) command)
                   (apply [message env])
                   error-or-answer)

           ;; or send it to the delegate
           (if-let [delegate  (:delegate @env)]
            (do
              (async/put! delegate message)
              i-got-it)
            nil
            )

           {:error
            (str "Unabled to process command: "
                 command)}))]
    (when (and answer
               return
               (not (= answer i-got-it))) (go (async/>! return answer)))))




(defn build-root-channel
  "Builds a root channel (usually put in env-root,
  but can be stand-alone so there can be multiple
  root channels in a single address space... nice for testing.
  The commands parameter is a map of String/function where the
  String is the name of the command and the function is a two
  parameter function (message and env) that returns a value to
  send to the answer channel or :dragonmark.circulate:i_got_it"
  ([] (build-root-channel {} (du/next-guid) nil))
  ([commands] (build-root-channel commands (du/next-guid) nil))
  ([commands name delegate]
     (let [c (chan 5)
           env (atom {:services
                      {'root {:channel c
                              :public true}}

                      :name name

                      :commands commands

                      :delegate delegate})

           chan+env
           (reify
             EnvInfo
             (env-info [_] env)
             (locate-service [_ service]
               (let [ret
                     (or
                      (-> @env :services (get service))
                      (if-let [delegate (:delegate @env)]
                        (do
                          (and
                           (instance? dragonmark.circulate.core.EnvInfo delegate)
                           (locate-service delegate service)))))]
                 ret
                 ))

             chans/MMC
             (cleanup [_] (chans/cleanup c))
             (abort [_] (chans/abort c))

{'fraction_non_alphanumeric': 0.11063060229595849, 'fraction_numerical': 0.00015725743041358704, 'mean_word_length': 1.442673324371039, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 25, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 12, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1192878', 'n_tokens_mistral': 3514, 'n_tokens_neox': 3312, 'n_words': 1114}