<commit_before>(ns ^:no-doc onyx.coordinator.distributed
  (:require [clojure.core.async :refer [>!! <!! chan]]
            [com.stuartsierra.component :as component]          
            [taoensso.timbre :as timbre]
            [ring.adapter.jetty :as jetty]
            [onyx.system :refer [onyx-coordinator]]))

(defmulti dispatch-request
  (fn [coordinator request] (:uri request)))

(defmethod dispatch-request "/submit-job"
  [coordinator request]
  (let [data (read-string (slurp (:body request)))]
    (>!! (:planning-ch-head (:coordinator coordinator)) data)))

(defmethod dispatch-request "/register-peer"
  [coordinator request]
  (let [data (read-string (slurp (:body request)))]
    (>!! (:born-peer-ch-head (:coordinator coordinator)) data)))

(defn handler [coordinator]
  (fn [request]
    (dispatch-request coordinator request)
    {:status 200
     :headers {"content-type" "text/text"}
     :body "ok"}))

(defrecord CoordinatorServer [opts]
  component/Lifecycle
  (start [component]
    (let [coordinator (component/start (onyx-coordinator opts))]
      (taoensso.timbre/info "Starting Coordinator Netty server")
      (assoc component
        :coordinator coordinator
        :server (jetty/run-jetty (handler coordinator) {:port (:onyx-port opts)
                                                        :join? false}))))
  
  (stop [component]
    (taoensso.timbre/info "Stopping Coordinator Netty server")

    (component/stop (:coordinator component))
    (.stop (:server component))

    component))

(defn coordinator-server [opts]
  (map->CoordinatorServer {:opts opts}))
<commit_msg>Don't require user to bring Component into their project.<commit_after>(ns ^:no-doc onyx.coordinator.distributed
  (:require [clojure.core.async :refer [>!! <!! chan]]
            [com.stuartsierra.component :as component]          
            [taoensso.timbre :as timbre]
            [ring.adapter.jetty :as jetty]
            [onyx.system :refer [onyx-coordinator]]))

(defmulti dispatch-request
  (fn [coordinator request] (:uri request)))

(defmethod dispatch-request "/submit-job"
  [coordinator request]
  (let [data (read-string (slurp (:body request)))]
    (>!! (:planning-ch-head (:coordinator coordinator)) data)))

(defmethod dispatch-request "/register-peer"
  [coordinator request]
  (let [data (read-string (slurp (:body request)))]
    (>!! (:born-peer-ch-head (:coordinator coordinator)) data)))

(defn handler [coordinator]
  (fn [request]
    (dispatch-request coordinator request)
    {:status 200
     :headers {"content-type" "text/text"}
     :body "ok"}))

(defrecord CoordinatorServer [opts]
  component/Lifecycle
  (start [component]
    (let [coordinator (component/start (onyx-coordinator opts))]
      (taoensso.timbre/info "Starting Coordinator Netty server")
      (assoc component
        :coordinator coordinator
        :server (jetty/run-jetty (handler coordinator) {:port (:onyx-port opts)
                                                        :join? false}))))
  
  (stop [component]
    (taoensso.timbre/info "Stopping Coordinator Netty server")

    (component/stop (:coordinator component))
    (.stop (:server component))

    component))

(defn coordinator-server [opts]
  (map->CoordinatorServer {:opts opts}))

(defn start-distributed-coordinator [opts]
  (component/start (coordinator-server opts)))

(defn stop-distributed-coordinator [coordinator]
  (component/stop coordinator))

{'fraction_non_alphanumeric': 0.14028463549230322, 'fraction_numerical': 0.0017426662794074934, 'mean_word_length': 3.251851851851852, 'pattern_counts': {'":': 0, '<': 5, '<?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': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5920201', 'n_tokens_mistral': 1084, 'n_tokens_neox': 1025, 'n_words': 259}