Add connected event (handler) (#152)

* Added an connected event (handler list)

* Prevent adding duplicated subscriptions

* Fix phpcs

* Pass info to connected event handlers if is auto-reconnect

* Prevent double subscriptions in MemoryRepository

* Add tests for the new connected event handlers

* Revert "Fix phpcs"

This reverts commit 10a9d42.

* Remove trailing whitespace

---------

Co-authored-by: Marvin Mall <[email protected]>

src/Concerns/OffersHooks.php

@@ -23,6 +23,9 @@ trait OffersHooks
     /** @var \SplObjectStorage|array<\Closure> */
     private $messageReceivedEventHandlers;
 
+    /** @var \SplObjectStorage|array<\Closure> */
+    private $connectedEventHandlers;
+
     /**
      * Needs to be called in order to initialize the trait.
      *
@@ -33,6 +36,7 @@ protected function initializeEventHandlers(): void
         $this->loopEventHandlers            = new \SplObjectStorage();
         $this->publishEventHandlers         = new \SplObjectStorage();
         $this->messageReceivedEventHandlers = new \SplObjectStorage();
+        $this->connectedEventHandlers       = new \SplObjectStorage();
     }
 
     /**
@@ -266,4 +270,77 @@ private function runMessageReceivedEventHandlers(string $topic, string $message,
             }
         }
     }
+
+    /**
+     * Registers an event handler which is called when the client established a connection to the broker.
+     * This also includes manual reconnects as well as auto-reconnects by the client itself.
+     *
+     * The event handler is passed the MQTT client as first argument,
+     * followed by a flag which indicates whether an auto-reconnect occurred as second argument.
+     *
+     * Example:
+     * ```php
+     * $mqtt->registerConnectedEventHandler(function (
+     *     MqttClient $mqtt,
+     *     bool $isAutoReconnect
+     * ) use ($logger) {
+     *     if ($isAutoReconnect) {
+     *         $logger->info("Client successfully auto-reconnected to the broker.);
+     *     } else {
+     *         $logger->info("Client successfully connected to the broker.");
+     *     }
+     * });
+     * ```
+     *
+     * Multiple event handlers can be registered at the same time.
+     *
+     * @param \Closure $callback
+     * @return MqttClient
+     */
+    public function registerConnectedEventHandler(\Closure $callback): MqttClient
+    {
+        $this->connectedEventHandlers->attach($callback);
+
+        /** @var MqttClient $this */
+        return $this;
+    }
+
+    /**
+     * Unregisters a connected event handler which prevents it from being called in the future.
+     *
+     * This does not affect other registered event handlers. It is possible
+     * to unregister all registered event handlers by passing null as callback.
+     *
+     * @param \Closure|null $callback
+     * @return MqttClient
+     */
+    public function unregisterConnectedEventHandler(\Closure $callback = null): MqttClient
+    {
+        if ($callback === null) {
+            $this->connectedEventHandlers->removeAll($this->connectedEventHandlers);
+        } else {
+            $this->connectedEventHandlers->detach($callback);
+        }
+
+        /** @var MqttClient $this */
+        return $this;
+    }
+
+    /**
+     * Runs all the registered connected event handlers.
+     * Each event handler is executed in a try-catch block to avoid spilling exceptions.
+     *
+     * @param bool $isAutoReconnect
+     * @return void
+     */
+    private function runConnectedEventHandlers(bool $isAutoReconnect): void
+    {
+        foreach ($this->connectedEventHandlers as $handler) {
+            try {
+                call_user_func($handler, $this, $isAutoReconnect);
+            } catch (\Throwable $e) {
+                $this->logger->error('Connected hook callback threw exception.', ['exception' => $e]);
+            }
+        }
+    }
 }

src/ConnectionSettings.php

@@ -6,8 +6,8 @@
 
 /**
  * The settings used during connection to a broker.
- * 
- * This class is immutable and all setters return a clone of the original class because 
+ *
+ * This class is immutable and all setters return a clone of the original class because
  * connection settings must not change once passed to MqttClient.
  *
  * @package PhpMqtt\Client

src/MqttClient.php

@@ -148,10 +148,11 @@ public function connect(
      * Connect to the MQTT broker using the configured settings.
      *
      * @param bool $useCleanSession
+     * @param bool $isAutoReconnect
      * @return void
      * @throws ConnectingToBrokerFailedException
      */
-    protected function connectInternal(bool $useCleanSession = false): void
+    protected function connectInternal(bool $useCleanSession = false, bool $isAutoReconnect = false): void
     {
         try {
             $this->establishSocketConnection();
@@ -163,6 +164,8 @@ protected function connectInternal(bool $useCleanSession = false): void
         }
 
         $this->connected = true;
+
+        $this->runConnectedEventHandlers($isAutoReconnect);
     }
 
     /**
@@ -414,7 +417,7 @@ protected function reconnect(): void
 
         for ($i = 1; $i <= $maxReconnectAttempts; $i++) {
             try {
-                $this->connectInternal();
+                $this->connectInternal(false, true);
 
                 return;
             } catch (ConnectingToBrokerFailedException $e) {

src/Repositories/MemoryRepository.php

@@ -191,6 +191,9 @@ public function countSubscriptions(): int
      */
     public function addSubscription(Subscription $subscription): void
     {
+        // Remove a potentially existing subscription for this topic filter.
+        $this->removeSubscription($subscription->getTopicFilter());
+
         $this->subscriptions[] = $subscription;
     }
 
@@ -217,16 +220,13 @@ public function getSubscriptionsMatchingTopic(string $topicName): array
      */
     public function removeSubscription(string $topicFilter): bool
     {
-        $result = false;
-
         foreach ($this->subscriptions as $index => $subscription) {
             if ($subscription->getTopicFilter() === $topicFilter) {
                 unset($this->subscriptions[$index]);
-                $result = true;
-                break;
+                return true;
             }
         }
 
-        return $result;
+        return false;
     }
 }

tests/Feature/ConnectedEventHandlerTest.php

@@ -0,0 +1,116 @@
+<?php
+
+/** @noinspection PhpUnhandledExceptionInspection */
+
+declare(strict_types=1);
+
+namespace Tests\Feature;
+
+use PhpMqtt\Client\MqttClient;
+use Tests\TestCase;
+
+/**
+ * Tests that the connected event handler work as intended.
+ *
+ * @package Tests\Feature
+ */
+class ConnectedEventHandlerTest extends TestCase
+{
+    public function test_connected_event_handlers_are_called_every_time_the_client_connects_successfully(): void
+    {
+        $client = new MqttClient($this->mqttBrokerHost, $this->mqttBrokerPort, 'test-connected-event-handler');
+
+        $handlerCallCount = 0;
+        $handler = function () use (&$handlerCallCount) {
+            $handlerCallCount++;
+        };
+
+        $client->registerConnectedEventHandler($handler);
+        $client->connect();
+
+        $this->assertSame(1, $handlerCallCount);
+
+        $client->disconnect();
+        $client->connect();
+
+        $this->assertSame(2, $handlerCallCount);
+
+        $client->disconnect();
+        $client->connect();
+
+        $this->assertSame(3, $handlerCallCount);
+
+        $client->disconnect();
+    }
+
+    public function test_connected_event_handlers_can_be_unregistered_and_will_not_be_called_anymore(): void
+    {
+        $client = new MqttClient($this->mqttBrokerHost, $this->mqttBrokerPort, 'test-connected-event-handler');
+
+        $handlerCallCount = 0;
+        $handler = function () use (&$handlerCallCount) {
+            $handlerCallCount++;
+        };
+
+        $client->registerConnectedEventHandler($handler);
+        $client->connect();
+
+        $this->assertSame(1, $handlerCallCount);
+
+        $client->unregisterConnectedEventHandler($handler);
+        $client->disconnect();
+        $client->connect();
+
+        $this->assertSame(1, $handlerCallCount);
+
+        $client->registerConnectedEventHandler($handler);
+        $client->disconnect();
+        $client->connect();
+
+        $this->assertSame(2, $handlerCallCount);
+
+        $client->unregisterConnectedEventHandler($handler);
+        $client->disconnect();
+        $client->connect();
+
+        $this->assertSame(2, $handlerCallCount);
+
+        $client->disconnect();
+    }
+
+    public function test_connected_event_handlers_can_throw_exceptions_which_does_not_affect_other_handlers_or_the_application(): void
+    {
+        $client = new MqttClient($this->mqttBrokerHost, $this->mqttBrokerPort, 'test-connected-event-handler');
+
+        $handlerCallCount = 0;
+        $handler1 = function () use (&$handlerCallCount) {
+            $handlerCallCount++;
+        };
+        $handler2 = function () {
+            throw new \Exception('Something went wrong!');
+        };
+
+        $client->registerConnectedEventHandler($handler1);
+        $client->registerConnectedEventHandler($handler2);
+
+        $client->connect();
+
+        $this->assertSame(1, $handlerCallCount);
+
+        $client->disconnect();
+    }
+
+    public function test_connected_event_handler_is_passed_the_mqtt_client_and_the_auto_reconnect_flag_as_arguments(): void
+    {
+        $client = new MqttClient($this->mqttBrokerHost, $this->mqttBrokerPort, 'test-connected-event-handler');
+
+        $client->registerConnectedEventHandler(function ($mqttClient, $isAutoReconnect) {
+            $this->assertInstanceOf(MqttClient::class, $mqttClient);
+            $this->assertIsBool($isAutoReconnect);
+            $this->assertFalse($isAutoReconnect);
+        });
+
+        $client->connect();
+        $client->disconnect();
+    }
+}

tests/Feature/PublishEventHandlerTest.php

@@ -10,7 +10,7 @@
 use Tests\TestCase;
 
 /**
- * Tests that the loop event handler work as intended.
+ * Tests that the publish event handler work as intended.
  *
  * @package Tests\Feature
  */