<commit_before>(ns geschichte.platform
  "Platform specific io operations."
  (:require [clojure.set :as set]
            [clojure.edn :as edn]
            [clojure.string :as str]
            [geschichte.platform-log :refer [debug info warn error]]
            [konserve.platform :refer [read-string-safe]]
            [hasch.benc :refer [IHashCoercion -coerce]]
            [clojure.core.async :as async
             :refer [<!! <! >! timeout chan alt! go go-loop]]
            [org.httpkit.server :refer :all]
            [http.async.client :as cli]
            [cognitect.transit :as transit])
  (:import [java.io ByteArrayInputStream ByteArrayOutputStream]))


(defn throwable? [x]
  (instance? Throwable x))

(defn throw-err [e]
  (when (throwable? e) (throw e)) e)

(defmacro <? [ch]
  `(throw-err (async/<! ~ch)))

(defmacro <!? [ch]
  `(throw-err (async/<!! ~ch)))

(defmacro go<? [& body]
  `(go (try
         ~@body
         (catch Exception e#
           e#))))

(defmacro go>? [err-chan & body]
  `(go (try
         ~@body
         (catch Exception e#
           (>! ~err-chan e#)))))

(defmacro go-loop>? [err-chan bindings & body]
  `(go (try
         (loop ~bindings
           ~@body)
         (catch Exception e#
           (>! ~err-chan e#)))))

(defmacro go-loop<? [bindings & body]
  `(go<? (loop ~bindings ~@body) ))

(defn now [] (java.util.Date.))


(defn client-connect!
  "Connects to url. Puts [in out] channels on return channel when ready.
Only supports websocket at the moment, but is supposed to dispatch on
protocol of url. tag-table is an atom"
  [url tag-table]
  (let [host (.getHost (java.net.URL. (str/replace url #"^ws" "http"))) ; HACK
        http-client (cli/create-client) ;; TODO use as singleton var?
        in (chan)
        out (chan)
        opener (chan)]
    (try
      (cli/websocket http-client url
                     :open (fn [ws]
                             (info "ws-opened" ws)
                             (go-loop [m (<! out)]
                               (when m
                                 (debug "client sending msg to:" url m)
                                 (if (= (:topic m) :binary-fetched)
                                   (cli/send ws :byte (let [out (ByteArrayOutputStream.)]
                                                        (.write out (byte 0))
                                                        (.write out (:value m))
                                                        (.toByteArray out)))
                                   (let [baos (ByteArrayOutputStream. (* 100 1024 1024))
                                         writer (transit/writer baos :json)]
                                     (.write baos (byte-array 1 (byte 1)))
                                     (transit/write writer m)
                                     (cli/send ws :byte (.toByteArray baos)))
                                   #_(cli/send ws :text (str " " (pr-str m))))
                                 (recur (<! out))))
                             (async/put! opener [in out])
                             (async/close! opener))
                     :text (fn [ws data]
                             (let [m (read-string-safe @tag-table data)]
                               (debug "client received msg from:" url m)
                               (async/put! in (with-meta m {:host host}))))
                     :byte (fn [ws ^bytes data]
                             (let [blob (java.util.Arrays/copyOfRange data 1 (count data))]
                               (case (long (aget data 0))
                                 0
                                 (let [m {:topic :binary-fetched
                                          :value blob}]
                                   (debug "client received binary blob from:"
                                          url (take 10 (map byte blob)))
                                   (async/put! in (with-meta m {:host host})))

                                 1
                                 (let [bais (ByteArrayInputStream. blob)
                                       reader (transit/reader bais :json)
                                       m (transit/read reader)]
                                   (debug "client received transit blob from:"
                                          url (take 10 (map byte blob)))
                                   (async/put! in (with-meta m {:host host}))))))
                     :close (fn [ws code reason]
                              (info "closing" ws code reason)
                              (async/close! in)
                              (async/close! out))
                     :error (fn [ws err] (error "ws-error" url err)
                              (async/put! opener (ex-info "ws-error"
                                                          {:type :websocket-connection-error
                                                           :url url
                                                           :error err}))
                              (async/close! opener)))
      (catch Exception e
        (error "client-connect error:" url e)
        (async/put! opener (ex-info "client-connect error"
                                    {:type :websocket-connection-error
                                     :url url
                                     :error e}))
        (async/close! in)
        (async/close! opener)))
    opener))


(defn create-http-kit-handler!
  "Creates a server handler described by url, e.g. wss://myhost:8443/geschichte/ws.
Returns a map to run a peer with a platform specific server handler
under :handler.  tag-table is an atom according to clojure.edn/read, it
should be the same as for the peer's store."
  ([url]
   (create-http-kit-handler! url (atom {})))
  ([url tag-table]
   (let [channel-hub (atom {})
         conns (chan)
         handler (fn [request]
                   (let [client-id (gensym)
                         in (chan)
                         out (chan)]
                     (async/put! conns [in out])
                     (with-channel request channel
                       (swap! channel-hub assoc channel request)
                       (go-loop [m (<! out)]
                         (when m
                           (if (@channel-hub channel)
                             (do
                               (debug "server sending msg:" url (pr-str m))
                               (if (= (:topic m) :binary-fetched)
                                 (send! channel ^bytes (let [out (ByteArrayOutputStream.)]
                                                         (.write out (byte 0))
                                                         (.write out (:value m))
                                                         (.toByteArray out)))
                                 (let [baos (ByteArrayOutputStream. (* 100 1024 1024))
                                       writer (transit/writer baos :json)]
                                   (.write baos (byte-array 1 (byte 1)))
                                   (transit/write writer m)
                                   (send! channel (.toByteArray baos) #_(str " " (pr-str m))))))
                             (debug "dropping msg because of closed channel: " url (pr-str m)))
                           (recur (<! out))))
                       (on-close channel (fn [status]
                                           (info "channel closed:" status)
                                           (swap! channel-hub dissoc channel)
                                           (async/close! in)))
                       (on-receive channel (fn [data]
                                             (if (string? data)
                                               (do
                                                 (debug "server received data:" url data)
                                                 (async/put! in
                                                             (with-meta
                                                               (read-string-safe @tag-table data)
                                                               {:host (:remote-addr request)})))
                                               (let [blob (java.util.Arrays/copyOfRange data 1 (count data))
                                                     host (:remote-addr request)]
                                                 (case (long (aget data 0))
                                                   0
                                                   (let [m {:topic :binary-fetched
                                                            :value blob}]
                                                     (debug "client received binary blob from:"
                                                            url (take 10 (map byte blob)))
                                                     (async/put! in (with-meta m {:host host})))

                                                   1
                                                   (let [bais (ByteArrayInputStream. blob)
                                                         reader (transit/reader bais :json)
                                                         m (transit/read reader)]
                                                     (debug "client received transit blob from:"
                                                            url (take 10 (map byte blob)))
                                                     (async/put! in (with-meta m {:host host})))))))))))]
     {:new-conns conns
      :channel-hub channel-hub
      :url url
      :handler handler})))




(defn start [peer]
  (when-let [handler (-> @peer :volatile :handler)]
    (println "starting" (:name @peer))
    (swap! peer assoc-in [:volatile :server]
           (run-server handler {:port (->> (-> @peer :volatile :url)
                                           (re-seq #":(\d+)")
                                           first
                                           second
                                           read-string)
                                :max-body (* 100 1024 1024)}))))


(defn stop [peer]
  (when-let [stop-fn (get-in @peer [:volatile :server])]
    (stop-fn :timeout 100))
  (<!! (timeout 200))
  (when-let [hub (get-in @peer [:volatile :channel-hub])]
    (reset! hub {}))
  (when-let [in (-> @peer :volatile :chans first)]<commit_msg>Try messagepack to avoid memory footprint.<commit_after>(ns geschichte.platform
  "Platform specific io operations."
  (:require [clojure.set :as set]
            [clojure.edn :as edn]
            [clojure.string :as str]
            [geschichte.platform-log :refer [debug info warn error]]
            [konserve.platform :refer [read-string-safe]]
            [hasch.benc :refer [IHashCoercion -coerce]]
            [clojure.core.async :as async
             :refer [<!! <! >! timeout chan alt! go go-loop]]
            [org.httpkit.server :refer :all]
            [http.async.client :as cli]
            [cognitect.transit :as transit])
  (:import [java.io ByteArrayInputStream ByteArrayOutputStream]))


(defn throwable? [x]
  (instance? Throwable x))

(defn throw-err [e]
  (when (throwable? e) (throw e)) e)

(defmacro <? [ch]
  `(throw-err (async/<! ~ch)))

(defmacro <!? [ch]
  `(throw-err (async/<!! ~ch)))

(defmacro go<? [& body]
  `(go (try
         ~@body
         (catch Exception e#
           e#))))

(defmacro go>? [err-chan & body]
  `(go (try
         ~@body
         (catch Exception e#
           (>! ~err-chan e#)))))

(defmacro go-loop>? [err-chan bindings & body]
  `(go (try
         (loop ~bindings
           ~@body)
         (catch Exception e#
           (>! ~err-chan e#)))))

(defmacro go-loop<? [bindings & body]
  `(go<? (loop ~bindings ~@body) ))

(defn now [] (java.util.Date.))


(defn client-connect!
  "Connects to url. Puts [in out] channels on return channel when ready.
Only supports websocket at the moment, but is supposed to dispatch on
protocol of url. tag-table is an atom"
  [url tag-table]
  (let [host (.getHost (java.net.URL. (str/replace url #"^ws" "http"))) ; HACK
        http-client (cli/create-client) ;; TODO use as singleton var?
        in (chan)
        out (chan)
        opener (chan)]
    (try
      (cli/websocket http-client url
                     :open (fn [ws]
                             (info "ws-opened" ws)
                             (go-loop [m (<! out)]
                               (when m
                                 (debug "client sending msg to:" url m)
                                 (if (= (:topic m) :binary-fetched)
                                   (cli/send ws :byte (let [out (ByteArrayOutputStream.)]
                                                        (.write out (byte 0))
                                                        (.write out (:value m))
                                                        (.toByteArray out)))
                                   (let [baos (ByteArrayOutputStream. (* 100 1024 1024))
                                         writer (transit/writer baos :msgpack)]
                                     (.write baos (byte-array 1 (byte 1)))
                                     (transit/write writer m)
                                     (cli/send ws :byte (.toByteArray baos)))
                                   #_(cli/send ws :text (str " " (pr-str m))))
                                 (recur (<! out))))
                             (async/put! opener [in out])
                             (async/close! opener))
                     :text (fn [ws data]
                             (let [m (read-string-safe @tag-table data)]
                               (debug "client received msg from:" url m)
                               (async/put! in (with-meta m {:host host}))))
                     :byte (fn [ws ^bytes data]
                             (let [blob (java.util.Arrays/copyOfRange data 1 (count data))]
                               (case (long (aget data 0))
                                 0
                                 (let [m {:topic :binary-fetched
                                          :value blob}]
                                   (debug "client received binary blob from:"
                                          url (take 10 (map byte blob)))
                                   (async/put! in (with-meta m {:host host})))

                                 1
                                 (let [bais (ByteArrayInputStream. blob)
                                       reader (transit/reader bais :msgpack)
                                       m (transit/read reader)]
                                   (debug "client received transit blob from:"
                                          url (take 10 (map byte blob)))
                                   (async/put! in (with-meta m {:host host}))))))
                     :close (fn [ws code reason]
                              (info "closing" ws code reason)
                              (async/close! in)
                              (async/close! out))
                     :error (fn [ws err] (error "ws-error" url err)
                              (async/put! opener (ex-info "ws-error"
                                                          {:type :websocket-connection-error
                                                           :url url
                                                           :error err}))
                              (async/close! opener)))
      (catch Exception e
        (error "client-connect error:" url e)
        (async/put! opener (ex-info "client-connect error"
                                    {:type :websocket-connection-error
                                     :url url
                                     :error e}))
        (async/close! in)
        (async/close! opener)))
    opener))


(defn create-http-kit-handler!
  "Creates a server handler described by url, e.g. wss://myhost:8443/geschichte/ws.
Returns a map to run a peer with a platform specific server handler
under :handler.  tag-table is an atom according to clojure.edn/read, it
should be the same as for the peer's store."
  ([url]
   (create-http-kit-handler! url (atom {})))
  ([url tag-table]
   (let [channel-hub (atom {})
         conns (chan)
         handler (fn [request]
                   (let [client-id (gensym)
                         in (chan)
                         out (chan)]
                     (async/put! conns [in out])
                     (with-channel request channel
                       (swap! channel-hub assoc channel request)
                       (go-loop [m (<! out)]
                         (when m
                           (if (@channel-hub channel)
                             (do
                               (debug "server sending msg:" url (pr-str m))
                               (if (= (:topic m) :binary-fetched)
                                 (send! channel ^bytes (let [out (ByteArrayOutputStream.)]
                                                         (.write out (byte 0))
                                                         (.write out (:value m))
                                                         (.toByteArray out)))
                                 (let [baos (ByteArrayOutputStream. (* 100 1024 1024))
                                       writer (transit/writer baos :msgpack)]
                                   (.write baos (byte-array 1 (byte 1)))
                                   (transit/write writer m)
                                   (send! channel (.toByteArray baos) #_(str " " (pr-str m))))))
                             (debug "dropping msg because of closed channel: " url (pr-str m)))
                           (recur (<! out))))
                       (on-close channel (fn [status]
                                           (info "channel closed:" status)
                                           (swap! channel-hub dissoc channel)
                                           (async/close! in)))
                       (on-receive channel (fn [data]
                                             (if (string? data)
                                               (do
                                                 (debug "server received data:" url data)
                                                 (async/put! in
                                                             (with-meta
                                                               (read-string-safe @tag-table data)
                                                               {:host (:remote-addr request)})))
                                               (let [blob (java.util.Arrays/copyOfRange data 1 (count data))
                                                     host (:remote-addr request)]
                                                 (case (long (aget data 0))
                                                   0
                                                   (let [m {:topic :binary-fetched
                                                            :value blob}]
                                                     (debug "client received binary blob from:"
                                                            url (take 10 (map byte blob)))
                                                     (async/put! in (with-meta m {:host host})))

                                                   1
                                                   (let [bais (ByteArrayInputStream. blob)
                                                         reader (transit/reader bais :msgpack)
                                                         m (transit/read reader)]
                                                     (debug "client received transit blob from:"
                                                            url (take 10 (map byte blob)))
                                                     (async/put! in (with-meta m {:host host})))))))))))]
     {:new-conns conns
      :channel-hub channel-hub
      :url url
      :handler handler})))




(defn start [peer]
  (when-let [handler (-> @peer :volatile :handler)]
    (println "starting" (:name @peer))
    (swap! peer assoc-in [:volatile :server]
           (run-server handler {:port (->> (-> @peer :volatile :url)
                                           (re-seq #":(\d+)")
                                           first
                                           second
                                           read-string)
                                :max-body (* 100 1024 1024)}))))


(defn stop [peer]
  (when-let [stop-fn (get-in @peer [:volatile :server])]
    (stop-fn :timeout 100))
  (<!! (timeout 200))
  (when-let [hub (get-in @peer [:volatile :channel-hub])]
    (reset! hub {}))
  (when-let [in (-> @peer :volatile :chans first)]
{'fraction_non_alphanumeric': 0.10393512997853566, 'fraction_numerical': 0.006200810875268304, 'mean_word_length': 0.8721314403071703, 'pattern_counts': {'":': 2, '<': 31, '<?xml version=': 0, '>': 23, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 228, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '340430', 'n_tokens_mistral': 4948, 'n_tokens_neox': 4590, 'n_words': 1480}