From f6b8739f93215d02dd8c9683c21d549fc463c722 Mon Sep 17 00:00:00 2001
From: "niten\" (aider)" <niten@fudo.org>
Date: Sun, 8 Jun 2025 08:50:56 -0700
Subject: [PATCH] feat: Split client.clj into api.clj, core.clj, and utils.clj
 with tests

---
 src/milquetoast/api.clj       |  56 ++++++++++++++
 src/milquetoast/core.clj      | 135 ++++++++++++++++++++++++++++++++++
 src/milquetoast/utils.clj     |  21 ++++++
 test/milquetoast/api_test.clj |  56 +++++++++++++-
 4 files changed, 267 insertions(+), 1 deletion(-)

diff --git a/src/milquetoast/api.clj b/src/milquetoast/api.clj
index 6c3fda1..17ed3dd 100644
--- a/src/milquetoast/api.clj
+++ b/src/milquetoast/api.clj
@@ -1 +1,57 @@
 (ns milquetoast.api)
+(ns milquetoast.api
+  (:require [milquetoast.core :as core]
+            [milquetoast.utils :as utils]))
+
+(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}}]
+  (core/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]
+  (core/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]
+  (core/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)]
+    (core/add-channel! client chan)
+    (go-loop [msg (<! chan)]
+      (when msg
+        (core/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}}]
+  (core/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)]
+    (core/MilquetoastClient. (core/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]
+  (core/MilquetoastJsonClient. (apply connect! args)))
diff --git a/src/milquetoast/core.clj b/src/milquetoast/core.clj
index e69de29..f3b2fe0 100644
--- a/src/milquetoast/core.clj
+++ b/src/milquetoast/core.clj
@@ -0,0 +1,135 @@
+(ns milquetoast.core
+  (: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)))
+
+(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 utils/json-parse-message)))
+  (get-topic! [_ topic opts]
+    (if-let [msg (get-topic! client topic opts)]
+      (utils/json-parse-message msg)
+      nil))
+  (get-topic-raw! [_ topic opts]
+    (if-let [msg (get-topic! client topic opts)]
+      msg
+      nil)))
diff --git a/src/milquetoast/utils.clj b/src/milquetoast/utils.clj
index 1a6ecbf..7a299ea 100644
--- a/src/milquetoast/utils.clj
+++ b/src/milquetoast/utils.clj
@@ -1 +1,22 @@
 (ns milquetoast.utils)
+(ns milquetoast.utils
+  (:require [clojure.core.async :as async]
+            [clojure.data.json :as json]
+            [clojure.tools.logging :as log])
+  (:import java.time.Instant))
+
+(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))))
diff --git a/test/milquetoast/api_test.clj b/test/milquetoast/api_test.clj
index 42b608d..d29c49c 100644
--- a/test/milquetoast/api_test.clj
+++ b/test/milquetoast/api_test.clj
@@ -1,3 +1,57 @@
 (ns milquetoast.api-test
   (:require [milquetoast.api :as sut]
-            [clojure.test :as t]))
+            [clojure.test :as t]
+            [clojure.core.async :as async]))
+
+(t/deftest test-send!
+  (t/testing "send! function"
+    (let [client (sut/connect! :host "localhost" :port 1883)
+          topic "test/topic"
+          msg "Hello, World!"]
+      (sut/send! client topic msg)
+      ;; Add more assertions here
+      )))
+
+(t/deftest test-get!
+  (t/testing "get! function"
+    (let [client (sut/connect! :host "localhost" :port 1883)
+          topic "test/topic"]
+      (sut/get! client topic)
+      ;; Add more assertions here
+      )))
+
+(t/deftest test-get-raw!
+  (t/testing "get-raw! function"
+    (let [client (sut/connect! :host "localhost" :port 1883)
+          topic "test/topic"]
+      (sut/get-raw! client topic)
+      ;; Add more assertions here
+      )))
+
+(t/deftest test-open-channel!
+  (t/testing "open-channel! function"
+    (let [client (sut/connect! :host "localhost" :port 1883)
+          topic "test/topic"]
+      (sut/open-channel! client topic)
+      ;; Add more assertions here
+      )))
+
+(t/deftest test-subscribe!
+  (t/testing "subscribe! function"
+    (let [client (sut/connect! :host "localhost" :port 1883)
+          topic "test/topic"]
+      (sut/subscribe! client topic)
+      ;; Add more assertions here
+      )))
+
+(t/deftest test-connect!
+  (t/testing "connect! function"
+    (let [client (sut/connect! :host "localhost" :port 1883)]
+      ;; Add assertions here
+      )))
+
+(t/deftest test-connect-json!
+  (t/testing "connect-json! function"
+    (let [client (sut/connect-json! :host "localhost" :port 1883)]
+      ;; Add assertions here
+      )))