public/milquetoast
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Changes per Aider

Browse Source
This commit is contained in:
niten
2025-06-08 09:33:51 -07:00
parent 86d3747185
commit 5a843fd6e5
8 changed files with 25 additions and 214 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+2
View File
@@ -9,3 +9,5 @@ target/
result
.lsp/
.clj-kondo/
.aider*
.env
deps.edn
+1 -1
View File
@@ -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 {
src/milquetoast/client.clj
+1 -204
View File
@@ -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)
src/milquetoast/core.clj
+12 -5
View File
@@ -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)))
src/milquetoast/utils.clj
+1 -3
View File
@@ -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 []
test/milquetoast/client.clj → test/milquetoast/client_test.clj
+1 -1
View File
@@ -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]))
test/milquetoast/core_test.clj
+3
View File
@@ -0,0 +1,3 @@
(ns milquetoast.core-test
(:require [milquetoast.core :as sut]
[clojure.test :as t]))
test/milquetoast/utils_test.clj
+4
View File
@@ -0,0 +1,4 @@
(ns milquetoast.utils-test
(:require [milquetoast.utils :as sut]
[clojure.test :as t]))