QoS 1 does only acknowledge if the handlers did not throw an Exception

php-mqtt · Nov 8, 2023 · c2980f3 · c2980f3
1 parent 4d6ecec
commit c2980f3
Show file tree

Hide file tree

Showing 2 changed files with 19 additions and 11 deletions.
diff --git a/src/Concerns/OffersHooks.php b/src/Concerns/OffersHooks.php
@@ -255,20 +255,23 @@ public function unregisterMessageReceivedEventHandler(\Closure $callback = null)
      * @param string $message
      * @param int    $qualityOfService
      * @param bool   $retained
-     * @return void
+     * @return bool False if an Event Handler threw an Exception, True otherwise.
      */
-    private function runMessageReceivedEventHandlers(string $topic, string $message, int $qualityOfService, bool $retained): void
+    private function runMessageReceivedEventHandlers(string $topic, string $message, int $qualityOfService, bool $retained): bool
     {
+        $one_callback_failed = false;
         foreach ($this->messageReceivedEventHandlers as $handler) {
             try {
                 call_user_func($handler, $this, $topic, $message, $qualityOfService, $retained);
             } catch (\Throwable $e) {
+                $one_callback_failed = true;
                 $this->logger->error('Received message hook callback threw exception for received message on topic [{topic}].', [
                     'topic' => $topic,
                     'exception' => $e,
                 ]);
             }
         }
+        return !$one_callback_failed;
     }
 
     /**

diff --git a/src/MqttClient.php b/src/MqttClient.php
@@ -744,11 +744,6 @@ protected function handleMessage(Message $message): void
     {
         // PUBLISH (incoming)
         if ($message->getType()->equals(MessageType::PUBLISH())) {
-            if ($message->getQualityOfService() === self::QOS_AT_LEAST_ONCE) {
-                // QoS 1.
-                $this->sendPublishAcknowledgement($message->getMessageId());
-            }
-
             if ($message->getQualityOfService() === self::QOS_EXACTLY_ONCE) {
                 // QoS 2, part 1.
                 try {
@@ -772,7 +767,13 @@ protected function handleMessage(Message $message): void
             }
 
             // For QoS 0 and QoS 1 we can deliver right away.
-            $this->deliverPublishedMessage($message->getTopic(), $message->getContent(), $message->getQualityOfService());
+            if ($this->deliverPublishedMessage($message->getTopic(), $message->getContent(), $message->getQualityOfService())) {
+                if ($message->getQualityOfService() === self::QOS_AT_LEAST_ONCE) {
+                    // QoS 1: Acknowledge only if the handlers did not throw an Exception.
+                    $this->sendPublishAcknowledgement($message->getMessageId());
+                }
+            }
+
             return;
         }
 
@@ -930,9 +931,9 @@ protected function allQueuesAreEmpty(): bool
      * @param string $message
      * @param int    $qualityOfServiceLevel
      * @param bool   $retained
-     * @return void
+     * @return bool False if an Event Handler threw an Exception, True otherwise.
      */
-    protected function deliverPublishedMessage(string $topic, string $message, int $qualityOfServiceLevel, bool $retained = false): void
+    protected function deliverPublishedMessage(string $topic, string $message, int $qualityOfServiceLevel, bool $retained = false): bool
     {
         $subscribers = $this->repository->getSubscriptionsMatchingTopic($topic);
 
@@ -943,6 +944,8 @@ protected function deliverPublishedMessage(string $topic, string $message, int $
             'subscribers' => count($subscribers),
         ]);
 
+        $one_callback_failed = false;
+
         foreach ($subscribers as $subscriber) {
             if ($subscriber->getCallback() === null) {
                 continue;
@@ -951,6 +954,8 @@ protected function deliverPublishedMessage(string $topic, string $message, int $
             try {
                 call_user_func($subscriber->getCallback(), $topic, $message, $retained, $subscriber->getMatchedWildcards($topic));
             } catch (\Throwable $e) {
+                $one_callback_failed = true;
+
                 $this->logger->error('Subscriber callback threw exception for published message on topic [{topic}].', [
                     'topic' => $topic,
                     'message' => $message,
@@ -959,7 +964,7 @@ protected function deliverPublishedMessage(string $topic, string $message, int $
             }
         }
 
-        $this->runMessageReceivedEventHandlers($topic, $message, $qualityOfServiceLevel, $retained);
+        return $this->runMessageReceivedEventHandlers($topic, $message, $qualityOfServiceLevel, $retained) && !$one_callback_failed;
     }
 
     /**