<commit_before>(ns kademlia.core
  (require [aleph.udp :as udp]
           [manifold.stream :as s]
           [taoensso.nippy :as nippy])
  (:gen-class))

;; (def socket (udp/socket {:port 12312}))

;; @(s/put! @socket {:host "localhost", :port 12312, :message (nippy/freeze {:foo :bar})})

;; (nippy/thaw (:message @(s/take! @socket)))

;; (s/close! @socket)

(defn -main
  "I don't do a whole lot ... yet."
  [& args]
  (println "Hello, World!"))
<commit_msg>Playing with APIs, wrote a ping test.<commit_after>(ns kademlia.core
  (require [aleph.udp :as udp]
           [manifold.stream :as s]
           [manifold.deferred :as d]
           [taoensso.nippy :as nippy])
  (:gen-class))

(def port 1665)
(def bind-timeout 5000) ;; milliseconds

(defn uuid []
  (java.util.UUID/randomUUID))

(defn send!
  "Returns a deferred that will eval to true if we successfully sent the message otherwise false."
  [socket host port msg]
  (s/put! socket
          {:host host
           :port port
           :message (nippy/freeze msg)}))

(defn recv-handler
  "Handler for all incoming messages.
  * socket: the socket receiving the message, we can use this to send a message back.
  * msg: The received message of format {:host [the host that sent the message]
                                         :port [the port that sent the message]
                                         :message [the message contents]}"
  [socket msg]
  (let [{:keys [host port message]} msg
        parsed-message (nippy/thaw message)
        ]
    (case (:type parsed-message)
      :ping (send! socket host port {:type :ack})
      )

    )
  )

(defn bind-socket
  "Returns a deferred socket, all messages coming into the socket will be consumed by
   recv-handler. If port is 0 we will bind to an open port."
  ([] (let [socket (deref (udp/socket {:port 0}) bind-timeout :timeout)]
        (s/consume (partial #'recv-handler socket) socket)
        socket) )
  ([port] (let [socket (deref (udp/socket {:port port}) bind-timeout :timeout)]
            (s/consume (partial #'recv-handler socket) socket)
            socket)))

(defonce socket (bind-socket port))

(defn socket->port
  "Get a port from a socket."
  [socket]
  (let [socket (:aleph/channel (meta socket))]
    (when (nil? socket)
      (throw (new Exception "socket is not a channel, maybe you need to defer it?")))
    (.getPort (.localAddress socket))))

;; TESTS

(use 'clojure.test)

(defmacro with-stream
  "Shamelessly modified with-open only it works on deferred streams instead."
  [bindings & body]
  (cond
    (= (count bindings) 0) `(do ~@body)
    (symbol? (bindings 0)) `(let ~(subvec bindings 0 2)
                              (try
                                (with-stream ~(subvec bindings 2) ~@body)
                                (finally
                                  (s/close! ~(bindings 0)))))
    :else (throw (IllegalArgumentException.
                   "with-stream only allows Symbols in bindings"))))

(deftest ping-test
  (testing "when we ping do we get an ack"
    (with-stream [application-socket (bind-socket)
                  test-socket        @(udp/socket {:port 0})
                  
                  ]
      (let [application-port (socket->port application-socket)
            application-host "localhost"
            ]
        (println application-host application-port (socket->port test-socket))
        @(send! test-socket application-host application-port {:type :ping})
        (is (= {:type :ack}
               (nippy/thaw (:message @(s/take! test-socket)))))))))

{'fraction_non_alphanumeric': 0.1305446378439079, 'fraction_numerical': 0.007580011229646266, 'mean_word_length': 2.250912408759124, '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': 5, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4854138', 'n_tokens_mistral': 1154, 'n_tokens_neox': 1068, 'n_words': 364}