<commit_before>(ns hellhound.kafka.core
  (:require [manifold.stream :as stream])
  (:import
   [java.util Properties]
   [org.apache.kafka.streams KafkaStreams StreamsBuilder]
   [org.apache.kafka.streams.kstream KStream]))



(def _builder (atom nil))

(defn ^Properties make-config
  "Create a Properties instance using the given CONFIG hashmap."
  [config]
  (let [stream-config (Properties.)]
    (.putAll stream-config config)
    stream-config))

(defn ^StreamsBuilder builder
  []
  (when (nil? @_builder)
    (let [b (StreamsBuilder.)]
      (reset! _builder b)))
  @_builder)

(defn reset-builder!
  []
  (reset! _builder nil))

(defn ^KStream stream
  [topic]
  (.stream (builder) topic))

(defn ^KafkaStreams streams
  [config]
  (KafkaStreams. (.build (builder))
                 (make-config config)))


(def cb (reify org.apache.kafka.streams.kstream.ForeachAction
          (apply [this k v]
            (println (str "<<<<<<<<<<<<<< " k ">>>> " v)))))

(reset! _builder nil)

(def ss (stream "something"))
(def a (streams {"application.id"    "test.app"
                 "retries"           "1"
                 "bootstrap.servers" ["localhost:9092"]}))

(.foreach ss cb)
(.start a)
(.close a)<commit_msg>first working PoC of kafka stream builder has been added<commit_after>(ns hellhound.kafka.core
  (:require [manifold.stream :as stream])
  (:import
   [java.util Properties]
   [org.apache.kafka.streams KafkaStreams StreamsBuilder StreamsConfig]
   [org.apache.kafka.streams.kstream KStream]))



(def _builder (atom nil))

(defn ^Properties make-config
  "Create a Properties instance using the given CONFIG hashmap."
  [config]
  (let [stream-config (Properties.)]
    (.putAll stream-config config)
    stream-config))

(defn ^StreamsBuilder builder
  []
  (when (nil? @_builder)
    (let [b (StreamsBuilder.)]
      (reset! _builder b)))
  @_builder)

(defn reset-builder!
  []
  (reset! _builder nil))

(defn ^KStream stream
  [topic]
  (.stream (builder) topic))

(defn ^KafkaStreams streams
  [config]
  (KafkaStreams. (.build (builder))
                 (make-config config)))


(def cb (reify org.apache.kafka.streams.kstream.ForeachAction
          (apply [this k v]
            (println (str "<<<<<<<<<<<<<< " k ">>>> " v)))))

(defn xx []
  (reset! _builder nil)
  (let [ss (stream "something")
        a (streams {"application.id"    "test.app"
                    "retries"           "1"
                    (StreamsConfig/BOOTSTRAP_SERVERS_CONFIG) "localhost:9092"})]
                    ;;"bootstrap.servers" "localhost:9092"})]

    (.foreach ss cb)
    (.start a)
    a))

(defn cc [a]
  (.close a))


(def x (xx))
(cc x)
{'fraction_non_alphanumeric': 0.15520481022172117, 'fraction_numerical': 0.005261180007515971, 'mean_word_length': 2.9089574155653453, 'pattern_counts': {'":': 0, '<': 31, '<?xml version=': 0, '>': 11, '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': '4149834', 'n_tokens_mistral': 933, 'n_tokens_neox': 897, 'n_words': 235}