Enhance MQTT message handling with semaphore control and update library build to 5

Monitoring Client/module.php

@@ -98,6 +98,10 @@ class MonitoringClient extends IPSModule {
         return false;
     }
 
+    private function getSendSemaphore() {
+        return "MonitoringClientSend_" . $this->InstanceID;
+    }
+
     /**
     * Die folgenden Funktionen stehen automatisch zur Verfügung, wenn das Modul über die "Module Control" eingefügt wurden.
     * Die Funktionen werden, mit dem selbst eingerichteten Prefix, in PHP und JSON-RPC wiefolgt zur Verfügung gestellt:
@@ -401,6 +405,12 @@ class MonitoringClient extends IPSModule {
     }
 
     public function SendTopic() {
+        // Do not send regular updates while a complete force send is running.
+        if (!IPS_SemaphoreEnter($this->getSendSemaphore(), 0)) {
+            return;
+        }
+
+        try {
         $mqttId = $this->ReadPropertyInteger("MqttCLientID");
         $projectyear = $this->ReadPropertyInteger("Projectyear");
         $projectnumber = $this->ReadPropertyInteger("Projectnumber");
@@ -434,9 +444,20 @@ class MonitoringClient extends IPSModule {
 
         $topic = "Projekte" . $projectyear . "/" . $projectnumber . "/ISP" . $ispnumber . "/Name";
         $this->MqttPublish($mqttId, $topic, $projectname, false);
+        } finally {
+            IPS_SemaphoreLeave($this->getSendSemaphore());
+        }
     }
 
     public function ForceSendTopic() {
+        // Wait briefly for a currently running regular update. Further regular
+        // updates skip their run until this complete force send has finished.
+        if (!IPS_SemaphoreEnter($this->getSendSemaphore(), 5000)) {
+            IPS_LogMessage("Monitoring Client", "Force Send skipped because another send is still running.");
+            return;
+        }
+
+        try {
         $mqttId = $this->ReadPropertyInteger("MqttCLientID");
         $projectyear = $this->ReadPropertyInteger("Projectyear");
         $projectnumber = $this->ReadPropertyInteger("Projectnumber");
@@ -463,11 +484,19 @@ class MonitoringClient extends IPSModule {
                 if (!$this->MqttPublish($mqttId, $topic, $payload, false)) {
                     return;
                 }
+
+                // Limit manual/full transfers to ten MQTT messages per second.
+                // This gives the subscribed client socket time to process its
+                // own echoed publications before the next one is sent.
+                IPS_Sleep(100);
             }
         }
 
         $topic = "Projekte" . $projectyear . "/" . $projectnumber . "/ISP" . $ispnumber . "/Name";
         $this->MqttPublish($mqttId, $topic, $projectname, false);
+        } finally {
+            IPS_SemaphoreLeave($this->getSendSemaphore());
+        }
     }
 
 

library.json

@@ -8,6 +8,6 @@
         "date": 1491343200
     },
     "version": "3.0",
-    "build": 4,
+    "build": 5,
     "date": 0
 }