<commit_before>(ns funcatron.tron.routers.shim-router
  (:require [funcatron.tron.util :as fu]
            [cheshire.core :as json]
            [org.httpkit.server :as kit]
            [io.sarnowski.swagger1st.core :as s1st]
            [io.sarnowski.swagger1st.util.security :as s1stsec]
            [io.sarnowski.swagger1st.context :as s1ctx])
  (:import (java.net ServerSocket Socket)
           (java.util Base64 UUID)
           (java.io BufferedWriter OutputStreamWriter BufferedReader InputStreamReader OutputStream InputStream ByteArrayInputStream)))


(set! *warn-on-reflection* true)

(defonce ^:private shim-socket (atom nil))

(defonce ^:private sync-obj (Object.))



(defn send-message
  [message]
  (let [b64 (fu/encode-body message)]
    (doseq [out-x (::output @shim-socket)]
      (let [out (BufferedWriter. (OutputStreamWriter. ^OutputStream out-x "UTF-8"))]
        (locking sync-obj
          (.write out b64)
          (.write out "\n")
          (.flush out))))))

(defn- shutdown
  "Close the server-socket"
  []
  (locking sync-obj
    (let [ss @shim-socket]
      (reset! shim-socket nil)
      (when-let [inputa (::input ss)]
        (doseq
          [input inputa]
          (try
            (.close ^InputStream input)
            (catch Exception e nil)))
        )
      (when-let [outputs (::output ss)]
        (doseq [output outputs]
          (try
            (.close ^OutputStream output)
            (catch Exception e nil)))
        )
      (when-let [sockets (::socket ss)]
        (doseq [socket sockets]
          (try
            (.close ^Socket socket)
            (catch Exception e nil)))
        )
      (when-let [ss (::server-socket ss)]
        (try
          (.close ^ServerSocket ss)
          (catch Exception e nil))))))

(defn- set-swagger
  [swagger-str]
  (swap! shim-socket dissoc ::swagger)
  (when-let [sd (try
             (s1ctx/load-swagger-definition
               :yaml
               swagger-str
               )
             (catch Exception e1
               (try
                 (s1ctx/load-swagger-definition
                   :json
                   swagger-str
                   )
                 (catch Exception e2
                   (do
                     (clojure.tools.logging/log :error e1 "Failed to parse as YAML")
                     (clojure.tools.logging/log :error e2 "Failed to parse as JSON")
                     nil
                     )))))]
    (swap! shim-socket assoc ::swagger sd)
    ))

(defn do-reply
  [{:keys [replyTo] :as resp}]
  (when-let [p (get-in @shim-socket [::requests replyTo])]
    (deliver p resp)
    (swap! shim-socket #(update % ::requests dissoc replyTo))
    )
  )

(defn- listen-to-input
  "Listen to the incoming socket"
  [^InputStream input]
  (let [in (BufferedReader. (InputStreamReader. input "UTF-8"))]
    (try
      (while true
        (when-let [line (.readLine in)]
          (let [bytes (.decode (Base64/getDecoder) line)
                map (json/parse-stream (BufferedReader. (InputStreamReader. (ByteArrayInputStream. bytes) "UTF-8")))
                map (fu/keywordize-keys map)
                cmd (:cmd map)]
            (cond
              (= "setSwagger" cmd)
              (set-swagger (:swagger map))

              (= "hello" cmd)
              (clojure.tools.logging/log :info "We got hello from the client")

              (= "reply" cmd)
              (do-reply map)

              :else nil
              )
            )))
      (finally (println "Exiting input loop"))
      )))

(defn- listen-to-socket
  "Accept socket connections"
  [^ServerSocket server-socket]
  (reset! shim-socket {::server-socket server-socket
                       ::socket []
                       ::input []
                       ::requests {}
                       ::output []})
  (try
    (while (not (.isClosed server-socket))
      (let [socket (.accept server-socket)
            input (.getInputStream socket)
            output (.getOutputStream socket)
            thread (Thread. (fn [] (listen-to-input input)) "Input Listener")]
        (swap! shim-socket #(merge-with into % {::socket [socket]
                                                ::input  [input]
                                                ::output [output]}))
        (.start thread)
        )

      )
    (catch Exception e (clojure.tools.logging/log :error e "Closed server socket" ))
    (finally (shutdown)))
  )

(defn exec-app
  [op-i req]
  (println "In exec app for " op-i)
  (let [uuid (.toString (UUID/randomUUID))
        p (promise)]
    (swap! shim-socket assoc-in [::requests uuid] p)
    (send-message {:cmd     "invoke"
                   :headers (:headers req)
                   :class op-i
                   :replyTo uuid
                   :body    (if (:body req) (fu/encode-body (:body req)) nil)})
    (let [answer (deref p 10000 ::failed)]
      (if (= ::failed answer)
        {:status 500 :body "Didn't get the answer"}
        answer))))

(defn resolve-app
  [req]
  (let [operationId (get req "operationId")]
    (partial exec-app operationId)
    ))

(defn make-app
  [swagger]
  (-> {:definition     swagger
       :chain-handlers (list)}
      (s1st/discoverer)
      (s1st/mapper)
      (s1st/parser)
      (s1st/protector {"oauth2" (s1stsec/allow-all)})
      (s1st/executor :resolver resolve-app))
  )

(defn setup
  "Set up a listner from a Shim"
  [^long port]
  (locking sync-obj
    (shutdown)
    (let [socket (ServerSocket. port)
          thread (Thread. (fn [] (listen-to-socket socket)) "Socket Acceptor")]
      (.start thread)
      )
    )
  )

(defn http-handler
  [req]
  (if-let [swagger (::swagger @shim-socket)]
    (let [app (make-app swagger)]
      (app req))
    {:status 404 :body "I'm swagger-less"}
    ))

(defonce end-server (atom nil))

(defn run-server
  []<commit_msg>Got the shim mostly working<commit_after>(ns funcatron.tron.routers.shim-router
  (:require [funcatron.tron.util :as fu]
            [cheshire.core :as json]
            [org.httpkit.server :as kit]
            [io.sarnowski.swagger1st.core :as s1st]
            [io.sarnowski.swagger1st.util.security :as s1stsec]
            [io.sarnowski.swagger1st.context :as s1ctx])
  (:import (java.net ServerSocket Socket SocketException)
           (java.util Base64 UUID)
           (java.io BufferedWriter OutputStreamWriter BufferedReader InputStreamReader OutputStream InputStream ByteArrayInputStream IOException)))


(set! *warn-on-reflection* true)

(defonce ^:private shim-socket (atom nil))

(defonce ^:private sync-obj (Object.))





(defn- shutdown-socket
  "Close the server-socket"
  []
  (locking sync-obj
    (let [ss @shim-socket]
      (swap! shim-socket #(-> %
                              (dissoc ::swagger)
                              (dissoc ::input)
                              (dissoc ::output)
                              (dissoc ::socket)))
      (when-let [inputa (::input ss)]
        (doseq
          [input [inputa]]
          (try
            (.close ^InputStream input)
            (catch Exception e nil)))
        )
      (when-let [outputs (::output ss)]
        (doseq [output [outputs]]
          (try
            (.close ^OutputStream output)
            (catch Exception e nil)))
        )
      (when-let [sockets (::socket ss)]
        (doseq [socket [sockets]]
          (try
            (.close ^Socket socket)
            (catch Exception e nil)))
        )
      )))

(defn send-message
  [message]
  (let [b64 (fu/encode-body message)]
    (try
      (doseq [out-x (filter identity [(::output @shim-socket)])]
        (let [out (BufferedWriter. (OutputStreamWriter. ^OutputStream out-x "UTF-8"))]
          (locking sync-obj
            (.write out b64)
            (.write out "\n")
            (.flush out))))
      (catch IOException io
        (do
          (clojure.tools.logging/log :error io "Failed to send message")
          (shutdown-socket)
          ))
      )))

(defn- shutdown
  "Close the server-socket"
  []
  (locking sync-obj
    (let [ss @shim-socket]
      (reset! shim-socket nil)
      (when-let [inputa (::input ss)]
        (doseq
          [input [inputa]]
          (try
            (.close ^InputStream input)
            (catch Exception e nil)))
        )
      (when-let [outputs (::output ss)]
        (doseq [output [outputs]]
          (try
            (.close ^OutputStream output)
            (catch Exception e nil)))
        )
      (when-let [sockets (::socket ss)]
        (doseq [socket [sockets]]
          (try
            (.close ^Socket socket)
            (catch Exception e nil)))
        )
      (when-let [ss (::server-socket ss)]
        (try
          (.close ^ServerSocket ss)
          (catch Exception e nil))))))

(defn- set-swagger
  [swagger-str]
  (swap! shim-socket dissoc ::swagger)
  (when-let [sd (try
                  (s1ctx/load-swagger-definition
                    :yaml
                    swagger-str
                    )
                  (catch Exception e1
                    (try
                      (s1ctx/load-swagger-definition
                        :json
                        swagger-str
                        )
                      (catch Exception e2
                        (do
                          (clojure.tools.logging/log :error e1 "Failed to parse as YAML")
                          (clojure.tools.logging/log :error e2 "Failed to parse as JSON")
                          nil
                          )))))]
    (swap! shim-socket assoc ::swagger sd)
    ))

(defn do-reply
  [{:keys [replyTo] :as resp}]
  (when-let [p (get-in @shim-socket [::requests replyTo])]
    (deliver p resp)
    (swap! shim-socket #(update % ::requests dissoc replyTo))
    )
  )

(defn- listen-to-input
  "Listen to the incoming socket"
  [^InputStream input]
  (let [in (BufferedReader. (InputStreamReader. input "UTF-8"))
        continue (atom true)]
    (try
      (while @continue
        (let [line (.readLine in)]
          (if line
            (let [bytes (.decode (Base64/getDecoder) line)
                  map (json/parse-stream (BufferedReader. (InputStreamReader. (ByteArrayInputStream. bytes) "UTF-8")))
                  map (fu/keywordize-keys map)
                  cmd (:cmd map)]
              (cond
                (= "setSwagger" cmd)
                (set-swagger (:swagger map))

                (= "hello" cmd)
                (clojure.tools.logging/log :info "We got hello from the client")

                (= "reply" cmd)
                (do-reply map)

                :else nil
                )
              )
            (reset! continue false)
            )))
      (finally
        (shutdown-socket)
        )
      )))

(defn- listen-to-socket
  "Accept socket connections"
  [^ServerSocket server-socket]
  (shutdown)
  (reset! shim-socket {::server-socket server-socket
                       ::socket        nil
                       ::input         nil
                       ::requests      {}
                       ::output        nil})
  (try
    (while (not (.isClosed server-socket))
      (let [socket (.accept server-socket)
            input (.getInputStream socket)
            output (.getOutputStream socket)
            thread (Thread. (fn [] (listen-to-input input)) "Input Listener")]
        (swap! shim-socket #(merge % {::socket socket
                                      ::input  input
                                      ::output output}))
        (.start thread)
        )

      )
    (catch SocketException se nil)                          ;; don't worry about closed socket closing
    (catch Exception e (clojure.tools.logging/log :error e "Closed server socket"))
    (finally
      (shutdown)))
  )

(defn exec-app
  [op-i req]
  (let [req2 (dissoc req :swagger)
        req2 (assoc req2 :body (get-in req [:parameters :body]))]
    (let [uuid (.toString (UUID/randomUUID))
          p (promise)]
      (swap! shim-socket assoc-in [::requests uuid] p)
      (send-message {:cmd     "invoke"
                     :headers (:headers req)
                     :class   op-i
                     :replyTo uuid
                     :body    (if (:body req2) (json/generate-string (:body req2)) nil)})
      (let [answer (deref p 10000 ::failed)]
        (if (= ::failed answer)
          {:status 500 :headers {"Content-Type" "text/plain"} :body "Didn't get the answer"}
          (let [response (:response answer)
                response (update response :headers fu/stringify-keys)
                response (if (and (:body response) (:decodeBody answer))
                           (assoc response
                             :body
                             (ByteArrayInputStream. (.decode (Base64/getDecoder) ^String (:body response))))
                           response)
                ]
            response
            ))))))

(defn resolve-app
  [req]
  (let [operationId (get req "operationId")]
    (partial exec-app operationId)
    ))

(defn make-app
  [swagger]
  (-> {:definition     swagger
       :chain-handlers (list)}
      (s1st/discoverer)
      (s1st/mapper)
      (s1st/parser)
      (s1st/protector {"oauth2" (s1stsec/allow-all)})
      (s1st/executor :resolver resolve-app))
  )

(defn setup
  "Set up a listner from a Shim"
  [^long port]
  (locking sync-obj
    (shutdown)
    (let [socket (ServerSocket. port)
          thread (Thread. (fn [] (listen-to-socket socket)) "Socket Acceptor")]
      (.start thread)
      )
    )
  )

(defn http-handler
  [req]
  (if-let [swagger (::swagger @shim-socket)]
    (let [app (make-app swagger)
          resp (app req)]
      resp
      )
    {:status 404 :headers {"Content-Type" "text/plain"} :body "No Swagger Defined. Unable to route request\n"}
    ))

(defonce end-server (atom nil))

(defn run-server
  []
{'fraction_non_alphanumeric': 0.12707821309816394, 'fraction_numerical': 0.0064334248951857746, 'mean_word_length': 1.606443104747551, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 6, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 6, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1882639', 'n_tokens_mistral': 4275, 'n_tokens_neox': 4057, 'n_words': 1100}