From 5a843fd6e503726c720d6743dbed1f5de906d67c Mon Sep 17 00:00:00 2001
From: niten <niten@fudo.org>
Date: Sun, 8 Jun 2025 09:33:51 -0700
Subject: [PATCH] Changes per Aider

---
 .gitignore                                    |   2 +
 deps.edn                                      |   2 +-
 src/milquetoast/client.clj                    | 205 +-----------------
 src/milquetoast/core.clj                      |  17 +-
 src/milquetoast/utils.clj                     |   4 +-
 .../{client.clj => client_test.clj}           |   2 +-
 test/milquetoast/core_test.clj                |   3 +
 test/milquetoast/utils_test.clj               |   4 +
 8 files changed, 25 insertions(+), 214 deletions(-)
 rename test/milquetoast/{client.clj => client_test.clj} (98%)
 create mode 100644 test/milquetoast/core_test.clj
 create mode 100644 test/milquetoast/utils_test.clj

diff --git a/.gitignore b/.gitignore
index 88823c4..66f455f 100644
--- a/.gitignore
+++ b/.gitignore
@@ -9,3 +9,5 @@ target/
 result
 .lsp/
 .clj-kondo/
+.aider*
+.env
diff --git a/deps.edn b/deps.edn
index 03589c6..ae8e5c1 100644
--- a/deps.edn
+++ b/deps.edn
@@ -5,7 +5,7 @@
         org.clojure/core.async { :mvn/version "1.8.741" }
         org.clojure/data.json { :mvn/version "2.4.0" }
         org.eclipse.paho/org.eclipse.paho.client.mqttv3 { :mvn/version "1.2.5" }
-        org.clojure/tools.logging { :mvn/version "1.1.0" }
+        org.clojure/tools.logging { :mvn/version "1.3.0" }
         }
  :aliases {
            :test {
diff --git a/src/milquetoast/client.clj b/src/milquetoast/client.clj
index 01a7f27..3afb1a2 100644
--- a/src/milquetoast/client.clj
+++ b/src/milquetoast/client.clj
@@ -1,204 +1 @@
-(ns milquetoast.client
-  (:require [clojure.core.async :as async :refer [go go-loop <! >! alts!! <!! timeout]]
-            [clojure.data.json :as json]
-            [clojure.tools.logging :as log])
-  (:import [org.eclipse.paho.client.mqttv3 MqttClient MqttConnectOptions MqttMessage IMqttMessageListener]
-           org.eclipse.paho.client.mqttv3.persist.MemoryPersistence
-           java.time.Instant))
-
-(defn create-mqtt-client!
-  "Creates and connects an MQTT client with the provided broker URI, username, and password."
-  [& {:keys [broker-uri username password]}]
-  (let [client-id (MqttClient/generateClientId)
-        opts (doto (MqttConnectOptions.)
-               (.setCleanSession true)
-               (.setAutomaticReconnect true))]
-    (when username
-      (doto opts
-        (.setUserName username)
-        (.setPassword (char-array password))))
-    (doto (MqttClient. broker-uri client-id (MemoryPersistence.))
-      (.connect opts))))
-
-(defn retry-attempt
-  "Attempts to execute function `f`. If `f` throws a RuntimeException, logs the exception and attempts to reconnect before retrying `f`."
-  [verbose f reconnect]
-  (let [wrapped-attempt (fn []
-                          (try [true (f)]
-                               (catch RuntimeException e
-                                 (do (when verbose
-                                       (log/error e "exception"))
-                                     [false e]))))
-        max-wait     (* 5 60 1000)] ;; wait at most 5 minutes
-    (loop [[success? result] (wrapped-attempt)
-           wait-ms 1000]
-      (if success?
-        result
-        (do (when verbose
-              (log/warn "attempt failed, attempting reconnect"))
-            (reconnect)
-            (when verbose
-              (log/debug (format "sleeping %s ms" wait-ms)))
-            (<!! (timeout wait-ms))
-            (recur (wrapped-attempt) (min (* wait-ms 1.25) max-wait)))))))
-
-(defn create-message
-  "Creates an MQTT message with the provided message string and options."
-  [msg {:keys [qos retain]
-        :or {qos    1
-             retain false}}]
-  (doto (MqttMessage. (.getBytes msg))
-    (.setQos      qos)
-    (.setRetained retain)))
-
-(defn parse-message
-  "Parses an MQTT message into a map with keys :id, :qos, :retained, :duplicate, :payload, and :payload-bytes."
-  [mqtt-msg]
-  {:id        (.getId mqtt-msg)
-   :qos       (.getQos mqtt-msg)
-   :retained  (.isRetained mqtt-msg)
-   :duplicate (.isDuplicate mqtt-msg)
-   :payload   (.toString mqtt-msg)
-   :payload-bytes (.getPayload mqtt-msg)})
-
-(defprotocol IMilquetoastClient
-  "Protocol defining the core operations of a Milquetoast MQTT client."
-  (send-message!    [_ topic msg opts])
-  (add-channel!     [_ chan])
-  (stop!            [_])
-  (subscribe-topic! [_ topic opts])
-  (get-topic!       [_ topic opts])
-  (get-topic-raw!   [_ topic opts]))
-
-(defrecord MilquetoastClient
-  [client open-channels verbose]
-  IMilquetoastClient
-  (send-message! [_ topic msg opts]
-    (retry-attempt verbose
-                   #(.publish client topic (create-message msg opts))
-                   #(.reconnect client)))
-  (stop! [_]
-    (when verbose
-      (log/info
-       (str "stopping " (count @open-channels) " channels")))
-    (doseq [chan @open-channels]
-      (async/close! chan))
-    true)
-  (add-channel! [_ chan]
-    (swap! open-channels conj chan))
-  (subscribe-topic! [self topic opts]
-    (let [{:keys [buffer-size qos]
-           :or   {buffer-size 1 qos 0}} opts
-          chan (async/chan buffer-size)]
-      (add-channel! self chan)
-      (retry-attempt verbose
-                     #(.subscribe client topic qos
-                                  (proxy [IMqttMessageListener] []
-                                    (messageArrived [topic mqtt-message]
-                                      (go (>! chan (assoc (parse-message mqtt-message)
-                                                          :topic topic))))))
-                     #(.reconnect client))
-      chan))
-  (get-topic! [_ topic opts]
-    (let [{:keys [qos timeout] :or {qos 0 timeout 5}} opts
-          result-chan (async/chan)]
-      (retry-attempt verbose
-                     #(.subscribe client topic qos
-                                  (proxy [IMqttMessageListener] []
-                                    (messageArrived [topic mqtt-message]
-                                      (go (>! result-chan (assoc (parse-message mqtt-message)
-                                                                 :topic topic))
-                                          (async/close! result-chan))
-                                      (.unsubscribe client topic))))
-                     #(.reconnect client))
-      (first (alts!! [result-chan
-                      (async/timeout (* timeout 1000))]))))
-  (get-topic-raw! [c topic opts] (get-topic! c topic opts)))
-
-(defn parallelism []
-  (-> (Runtime/getRuntime)
-      (.availableProcessors)
-      (+ 1)))
-
-(defn pipe [in xf]
-  (let [out (async/chan)]
-    (async/pipeline (parallelism) out xf in)
-    out))
-
-(defn json-parse-message [msg]
-  (-> msg
-      (update :payload   (fn [payload]
-                           (json/read-str payload :key-fn keyword)))
-      (assoc  :timestamp (Instant/now))))
-
-(defrecord MilquetoastJsonClient
-  [client]
-  IMilquetoastClient
-  (send-message! [_ topic msg opts]
-    (send-message! client topic (json/write-str msg) opts))
-  (stop! [_] (stop! client))
-  (add-channel! [_ chan] (add-channel! client chan))
-  (subscribe-topic! [_ topic opts]
-    (pipe (subscribe-topic! client topic opts)
-          (map json-parse-message)))
-  (get-topic! [_ topic opts]
-    (if-let [msg (get-topic! client topic opts)]
-      (json-parse-message msg)
-      nil))
-  (get-topic-raw! [_ topic opts]
-    (if-let [msg (get-topic! client topic opts)]
-      msg
-      nil)))
-
-(defn send!
-  "Sends a message to a topic on the provided client with the specified QoS and retain options."
-  [client topic msg & {:keys [qos retain]
-                                 :or   {qos 1 retain false}}]
-  (send-message! client topic msg {:qos qos :retain retain}))
-
-(defn get!
-  "Gets a message from a topic on the provided client with the specified options."
-  [client topic & options]
-  (get-topic! client topic options))
-
-(defn get-raw!
-  "Gets a raw message from a topic on the provided client with the specified options."
-  [client topic & options]
-  (get-topic-raw! client topic options))
-
-(defn open-channel!
-  "Opens a channel for sending messages to a topic on the provided client with the specified buffer size, QoS, and retain options."
-  [client topic & {:keys [buffer-size qos retain]
-                   :or   {buffer-size 1
-                          qos         1
-                          retain      false}}]
-  (let [chan (async/chan buffer-size)]
-    (add-channel! client chan)
-    (go-loop [msg (<! chan)]
-      (when msg
-        (send-message! client topic msg {:qos qos :retain retain})
-        (recur (<! chan))))
-    chan))
-
-(defn subscribe!
-  "Subscribes to a topic on the provided client with the specified buffer size and QoS options."
-  [client topic & {:keys [buffer-size qos]
-                   :or   {buffer-size 1
-                          qos         1}}]
-  (subscribe-topic! client topic {:buffer-size buffer-size :qos qos}))
-
-(defn connect!
-  "Connects to an MQTT broker at the provided host and port with the specified scheme and verbosity options."
-  [& {:keys [host port scheme verbose]
-      :or   {verbose false
-             scheme  :tcp}
-      :as   opts}]
-  (let [broker-uri (str (name scheme) "://" host ":" port)]
-    (MilquetoastClient. (create-mqtt-client! (assoc opts :broker-uri broker-uri))
-                        (atom [])
-                        verbose)))
-
-(defn connect-json!
-  "Connects to an MQTT broker with the provided arguments and configures the client to send and receive JSON messages."
-  [& args]
-  (MilquetoastJsonClient. (apply connect! args)))
+(ns milquetoast.client)
diff --git a/src/milquetoast/core.clj b/src/milquetoast/core.clj
index ee8f3aa..8527fee 100644
--- a/src/milquetoast/core.clj
+++ b/src/milquetoast/core.clj
@@ -140,11 +140,18 @@
 
 (defn create-client
   "Creates a new MilquetoastClient instance with the provided MQTT client and options."
-  [mqtt-client & {:keys [verbose]
-                  :or   {verbose false}}]
-  (MilquetoastClient. mqtt-client (atom []) verbose))
+  [broker-uri username password & {:keys [verbose]
+                                   :or   {verbose false}}]
+  (let [mqtt-client (create-mqtt-client! :broker-uri broker-uri
+                                         :username   username
+                                         :password   password)]
+    (MilquetoastClient. mqtt-client (atom []) verbose)))
 
 (defn create-json-client
   "Creates a new MilquetoastJsonClient instance with the provided MQTT client."
-  [mqtt-client]
-  (MilquetoastJsonClient. mqtt-client))
+  [broker-uri username password & {:keys [verbose]
+                                   :or   {verbose false}}]
+  (let [mqtt-client (create-mqtt-client! :broker-uri broker-uri
+                                         :username   username
+                                         :password   password)]
+    (MilquetoastJsonClient. mqtt-client (atom []) verbose)))
diff --git a/src/milquetoast/utils.clj b/src/milquetoast/utils.clj
index 7a299ea..e715696 100644
--- a/src/milquetoast/utils.clj
+++ b/src/milquetoast/utils.clj
@@ -1,8 +1,6 @@
-(ns milquetoast.utils)
 (ns milquetoast.utils
   (:require [clojure.core.async :as async]
-            [clojure.data.json :as json]
-            [clojure.tools.logging :as log])
+            [clojure.data.json :as json])
   (:import java.time.Instant))
 
 (defn parallelism []
diff --git a/test/milquetoast/client.clj b/test/milquetoast/client_test.clj
similarity index 98%
rename from test/milquetoast/client.clj
rename to test/milquetoast/client_test.clj
index 3357600..8fb7631 100644
--- a/test/milquetoast/client.clj
+++ b/test/milquetoast/client_test.clj
@@ -1,4 +1,4 @@
-(ns milquetoast.client
+(ns milquetoast.client-test
   (:require [milquetoast.client :as sut]
             [clojure.test :as t]
             [org.eclipse.paho.client.mqttv3 MqttClient]))
diff --git a/test/milquetoast/core_test.clj b/test/milquetoast/core_test.clj
new file mode 100644
index 0000000..e14f9f3
--- /dev/null
+++ b/test/milquetoast/core_test.clj
@@ -0,0 +1,3 @@
+(ns milquetoast.core-test
+  (:require [milquetoast.core :as sut]
+            [clojure.test :as t]))
diff --git a/test/milquetoast/utils_test.clj b/test/milquetoast/utils_test.clj
new file mode 100644
index 0000000..1061261
--- /dev/null
+++ b/test/milquetoast/utils_test.clj
@@ -0,0 +1,4 @@
+(ns milquetoast.utils-test
+  (:require [milquetoast.utils :as sut]
+            [clojure.test :as t]))
+