adopt new upstream interface

dkdhub · May 22, 2023 · af4d311 · af4d311
1 parent 529277a
commit af4d311
Show file tree

Hide file tree

Showing 4 changed files with 39 additions and 8 deletions.
diff --git a/project.clj b/project.clj
@@ -2,7 +2,7 @@
   :description "A Clojure wrapper to the Moquette Broker library"
   :url "https://github.com/dkdhub/clj-mqtt-broker"
   :license {:name "Built In Project License"}
-  :dependencies [[io.moquette/moquette-broker "0.17-SNAPSHOT-1"
+  :dependencies [[io.moquette/moquette-broker "0.17-SNAPSHOT"
                   :exclusions [com.bugsnag/bugsnag
                                org.slf4j/slf4j-api
                                org.slf4j/slf4j-log4j12]]])
diff --git a/src-clj/clj_mqtt_broker/core.clj b/src-clj/clj_mqtt_broker/core.clj
@@ -46,7 +46,9 @@
   (close [o])
   (send [o from to data qos retain?])
   (clients [o])
-  (disconnect [o ^String client]))
+  (disconnect
+    [o ^String client]
+    [o ^String client ^Boolean flush?]))
 
 (deftype Broker [^IBroker instance]
   CljBroker
@@ -66,9 +68,10 @@
     (if (instance? AdvancedBroker instance)
       (map #(->> % (into {}) clojure.walk/keywordize-keys) (.clients instance))
       nil))
-  (disconnect [_ client]
+  (disconnect [_ client] (disconnect _ client false))
+  (disconnect [_ client flush?]
     (if (instance? AdvancedBroker instance)
-      (.disconnect instance client)
+      (.disconnect instance client (boolean flush?))
       false)))
 
 (defmulti create-broker (fn [x] (class x)))

diff --git a/src-java/com/dkdhub/mqtt_broker/AdvancedBroker.java b/src-java/com/dkdhub/mqtt_broker/AdvancedBroker.java
@@ -89,11 +89,11 @@ public void send(String from, String topic, byte[] data, MqttQoS qos, Boolean re
         }
     }
 
-    public boolean disconnect(String client) {
+    public boolean disconnect(String client, boolean flush) {
         if (client == null) return false;
 
-        LOG.debug("Will disconnect client {}", client);
-        boolean res = m_server.disconnectClient(client);
+        LOG.debug("Will disconnect client {}, flush state: {}", client, flush);
+        boolean res = flush ? m_server.disconnectClient(client) : m_server.disconnectAndPurgeClientState(client);
         LOG.debug("Disconnect result: {}", res);
         return res;
     }

diff --git a/test/basic.clj b/test/basic.clj
@@ -130,7 +130,7 @@
 
 (deftest check-advanced-disconnects
   (testing "Checking advanced client for disconnects")
-  (log/info "--------- MQTT Advanced Broker check disconnect ---------")
+  (log/info "--------- MQTT Advanced Broker check disconnect (keep state) ---------")
   (is (let [b (Broker. (AdvancedBroker. advanced-config))
             results (atom [])]
         (start b (TraceHandlers. "3456"))
@@ -145,6 +145,34 @@
                 :when client-id]
           (swap! results conj (disconnect b client-id)))
 
+        ;; TODO: add here check state test part
+
+        (println "======> MQTT Advanced Broker goes down .........")
+        (println "======> MQTT Clients:" (clients b))
+        (Thread/sleep 1000)
+
+        (stop b)
+        (println "======> MQTT Results:" @results)
+        (or (empty? @results)
+            (every? true? @results))))
+
+  (log/info "--------- MQTT Advanced Broker check disconnect (state flush) ---------")
+  (is (let [b (Broker. (AdvancedBroker. advanced-config))
+            results (atom [])]
+        (start b (TraceHandlers. "3456"))
+
+        (println "======> MQTT Advanced Broker wait .........")
+        (Thread/sleep 10000)
+        (println "======> MQTT Clients:" (clients b))
+        (println "======> MQTT Advanced Broker disconnecting .........")
+
+        (doseq [client (clients b)
+                :let [client-id (:id client)]
+                :when client-id]
+          (swap! results conj (disconnect b client-id true)))
+
+        ;; TODO: add here check state test part
+
         (println "======> MQTT Advanced Broker goes down .........")
         (println "======> MQTT Clients:" (clients b))
         (Thread/sleep 1000)