<commit_before>(ns marceline.storm.metrics
  (:import [backtype.storm.metric.api IMetric CountMetric MultiCountMetric])
  (:require [backtype.storm.clojure :refer (to-spec normalize-fns)]
            [backtype.storm.config :refer (TOPOLOGY-METRICS-CONSUMER-REGISTER)])
  (:gen-class))


(defmacro defmetricsconsumer
  [name & [opts & impl :as all]]
  (if-not (map? opts)
    `(defmetricsconsumer ~name {} ~@all)
    (let [prefix (gensym)
          classname (str *ns* ".consumer." name)
          state "state"
          init "init"
          [prepare-impl handle-data-points-impl & [cleanup-impl?]] impl
          cleanup-impl (or cleanup-impl? `([_#]))
          ;; options
          parallelism (:p opts 1)
          init-state (:state opts {})]
      `(do
         (gen-class :name ~classname
                    :implements [backtype.storm.metric.api.IMetricsConsumer]
                    :prefix ~prefix
                    :state ~state
                    :init ~init)
         (defn ~(symbol (str prefix "init"))
           []
           [[] (atom ~init-state)])
         (defn ~(symbol (str prefix "prepare"))
           ~@prepare-impl)
         (defn ~(symbol (str prefix "handleDataPoints"))
           ~@handle-data-points-impl)
         (defn ~(symbol (str prefix "cleanup"))
           ~@cleanup-impl)
         (def ~name
           (fn [conf#]
             (merge-with
              concat
              conf#
              {~TOPOLOGY-METRICS-CONSUMER-REGISTER
               [{"class" ~classname
                 "parallelism.hint" ~parallelism
                 "argument" nil}]})))))))

;; I don't see the value of this argument
;; it's passed during `prepare`, which
;; is also passed the entire conf map.
;; so why not pull whatever arg out of
;; the rest of the conf?

(defmacro metric
  [get-value-and-reset-impl]
  `(reify IMetric
     (getValueAndReset [this]
       (~get-value-and-reset-impl))))<commit_msg>support passing 2nd param to consumer fns for registration arg<commit_after>(ns marceline.storm.metrics
  (:import [backtype.storm.metric.api IMetric CountMetric MultiCountMetric])
  (:require [backtype.storm.clojure :refer (to-spec normalize-fns)]
            [backtype.storm.config :refer (TOPOLOGY-METRICS-CONSUMER-REGISTER)])
  (:gen-class))


(defmacro defmetricsconsumer
  [name & [opts & impl :as all]]
  (if-not (map? opts)
    `(defmetricsconsumer ~name {} ~@all)
    (let [prefix (gensym)
          classname (str *ns* ".consumer." name)
          state "state"
          init "init"
          [prepare-impl handle-data-points-impl & [cleanup-impl?]] impl
          cleanup-impl (or cleanup-impl? `([_#]))
          ;; options
          parallelism (:p opts 1)
          init-state (:state opts {})]
      `(do
         (gen-class :name ~classname
                    :implements [backtype.storm.metric.api.IMetricsConsumer]
                    :prefix ~prefix
                    :state ~state
                    :init ~init)
         (defn ~(symbol (str prefix "init"))
           []
           [[] (atom ~init-state)])
         (defn ~(symbol (str prefix "prepare"))
           ~@prepare-impl)
         (defn ~(symbol (str prefix "handleDataPoints"))
           ~@handle-data-points-impl)
         (defn ~(symbol (str prefix "cleanup"))
           ~@cleanup-impl)
         (def ~name
           (fn [conf# & [arg#]]
             (merge-with
              concat
              conf#
              {~TOPOLOGY-METRICS-CONSUMER-REGISTER
               [{"class" ~classname
                 "parallelism.hint" ~parallelism
                 "argument" arg#}]})))))))

;; I don't see the value of this argument
;; it's passed during `prepare`, which
;; is also passed the entire conf map.
;; so why not pull whatever arg out of
;; the rest of the conf?

(defmacro metric
  [get-value-and-reset-impl]
  `(reify IMetric
     (getValueAndReset [this]
       (~get-value-and-reset-impl))))
{'fraction_non_alphanumeric': 0.13528662420382165, 'fraction_numerical': 0.0007643312101910828, 'mean_word_length': 1.9832826747720365, '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': '2346667', 'n_tokens_mistral': 1206, 'n_tokens_neox': 1160, 'n_words': 305}