Refactory repository interface with single pending message and subscriptions management (#33)

* Refactored Repository interface and pending message handling.

Details:

* Refactored the Repository interface.
  The new interface stores a generic serializable PendingMessage without
  knowing which type it is, using two different queues one for incoming and
  one for outgoing messages. There are currently three types of outgoing
  messages (PublishedMessage, SubscribeRequest, UnsubscribeRequest) and one
  incoming message (PublishedMessaged) handled.
* The `TopicSubscription` class has been renamed to `Subscription` and now
  represents only the subscription itself, separating it from the concept of
  SubscribeRequest which might include several topic filters in one message.
* Removed exception `UnexpectedAcknowledgementException`.
  It is expected to have duplicate packets over the wire, so there is no need
  for an expection in this case, a notice to the logger is enough.
* Added `ProtocolViolationException`.
  This explicitly marks situations where the broker is misbehaving and the
  connection should be terminated.
* Fixed some problems with the MessageProcessor.

* Removed Redis repository implementation until updated to the new Repository interface

* Fix MessageProcessor, exceptions and documentation

* Change visibility and description of DTOs

* Simplify acknowledgement parsing logic

* Change code formatting

* Change field visibility and formatting

Co-authored-by: Marvin Mall <marvin-mall@msn.com>

Author: thg2k

src/Contracts/MessageProcessor.php

@@ -8,8 +8,9 @@
 use PhpMqtt\Client\Exceptions\ConnectingToBrokerFailedException;
 use PhpMqtt\Client\Exceptions\InvalidMessageException;
 use PhpMqtt\Client\Exceptions\MqttClientException;
-use PhpMqtt\Client\Exceptions\UnexpectedAcknowledgementException;
+use PhpMqtt\Client\Exceptions\ProtocolViolationException;
 use PhpMqtt\Client\Message;
+use PhpMqtt\Client\Subscription;
 
 /**
  * Implementations of this interface provide message parsing capabilities.
@@ -42,7 +43,7 @@ public function tryFindMessageInBuffer(string $buffer, int $bufferLength, string
      * @param string $message
      * @return Message|null
      * @throws InvalidMessageException
-     * @throws UnexpectedAcknowledgementException
+     * @throws ProtocolViolationException
      * @throws MqttClientException
      */
     public function parseAndValidateMessage(string $message): ?Message;
@@ -58,11 +59,18 @@ public function parseAndValidateMessage(string $message): ?Message;
     public function buildConnectMessage(ConnectionSettings $connectionSettings, bool $useCleanSession = false): string;
 
     /**
-     * Builds a ping message.
+     * Builds a ping request message.
      *
      * @return string
      */
-    public function buildPingMessage(): string;
+    public function buildPingRequestMessage(): string;
+
+    /**
+     * Builds a ping response message.
+     *
+     * @return string
+     */
+    public function buildPingResponseMessage(): string;
 
     /**
      * Builds a disconnect message.
@@ -74,22 +82,22 @@ public function buildDisconnectMessage(): string;
     /**
      * Builds a subscribe message from the given parameters.
      *
-     * @param int    $messageId
-     * @param string $topic
-     * @param int    $qualityOfService
+     * @param int            $messageId
+     * @param Subscription[] $subscriptions
+     * @param bool           $isDuplicate
      * @return string
      */
-    public function buildSubscribeMessage(int $messageId, string $topic, int $qualityOfService): string;
+    public function buildSubscribeMessage(int $messageId, array $subscriptions, bool $isDuplicate = false): string;
 
     /**
      * Builds an unsubscribe message from the given parameters.
      *
-     * @param int    $messageId
-     * @param string $topic
-     * @param bool   $isDuplicate
+     * @param int      $messageId
+     * @param string[] $topics
+     * @param bool     $isDuplicate
      * @return string
      */
-    public function buildUnsubscribeMessage(int $messageId, string $topic, bool $isDuplicate = false): string;
+    public function buildUnsubscribeMessage(int $messageId, array $topics, bool $isDuplicate = false): string;
 
     /**
      * Builds a publish message based on the given parameters.

src/Contracts/MqttClient.php

@@ -8,8 +8,8 @@
 use PhpMqtt\Client\Exceptions\ConfigurationInvalidException;
 use PhpMqtt\Client\Exceptions\ConnectingToBrokerFailedException;
 use PhpMqtt\Client\Exceptions\DataTransferException;
-use PhpMqtt\Client\Exceptions\TopicNotSubscribedException;
-use PhpMqtt\Client\Exceptions\UnexpectedAcknowledgementException;
+use PhpMqtt\Client\Exceptions\ProtocolViolationException;
+use PhpMqtt\Client\Exceptions\RepositoryException;
 
 /**
  * An interface for the MQTT client.
@@ -57,6 +57,7 @@ public function isConnected(): bool;
      * @param bool   $retain
      * @return void
      * @throws DataTransferException
+     * @throws RepositoryException
      */
     public function publish(string $topic, string $message, int $qualityOfService = 0, bool $retain = false): void;
 
@@ -86,6 +87,7 @@ public function publish(string $topic, string $message, int $qualityOfService =
      * @param int      $qualityOfService
      * @return void
      * @throws DataTransferException
+     * @throws RepositoryException
      */
     public function subscribe(string $topic, callable $callback, int $qualityOfService = 0): void;
 
@@ -95,7 +97,6 @@ public function subscribe(string $topic, callable $callback, int $qualityOfServi
      * @param string $topic
      * @return void
      * @throws DataTransferException
-     * @throws TopicNotSubscribedException
      */
     public function unsubscribe(string $topic): void;
 
@@ -137,7 +138,7 @@ public function interrupt(): void;
      * @param int|null $queueWaitLimit
      * @return void
      * @throws DataTransferException
-     * @throws UnexpectedAcknowledgementException
+     * @throws ProtocolViolationException
      */
     public function loop(bool $allowSleep = true, bool $exitWhenQueuesEmpty = false, int $queueWaitLimit = null): void;
 

src/Contracts/Repository.php

@@ -5,10 +5,11 @@
 namespace PhpMqtt\Client\Contracts;
 
 use DateTime;
-use PhpMqtt\Client\Exceptions\PendingPublishConfirmationAlreadyExistsException;
-use PhpMqtt\Client\PublishedMessage;
-use PhpMqtt\Client\TopicSubscription;
-use PhpMqtt\Client\UnsubscribeRequest;
+use PhpMqtt\Client\Exceptions\PendingMessageAlreadyExistsException;
+use PhpMqtt\Client\Exceptions\PendingMessageNotFoundException;
+use PhpMqtt\Client\Exceptions\RepositoryException;
+use PhpMqtt\Client\PendingMessage;
+use PhpMqtt\Client\Subscription;
 
 /**
  * Implementations of this interface provide storage capabilities to an MQTT client.
@@ -31,190 +32,139 @@ interface Repository
      * but it is currently not being used (i.e. in a resend queue).
      *
      * @return int
+     * @throws RepositoryException
      */
     public function newMessageId(): int;
 
     /**
-     * Releases the given message id, allowing it to be reused in the future.
-     *
-     * @param int $messageId
-     * @return void
-     */
-    public function releaseMessageId(int $messageId): void;
-
-    /**
-     * Returns the number of registered topic subscriptions. The method does
-     * not differentiate between pending and acknowledged subscriptions.
+     * Returns the number of pending outgoing messages.
      *
      * @return int
      */
-    public function countTopicSubscriptions(): int;
-
-    /**
-     * Adds a topic subscription to the repository.
-     *
-     * @param TopicSubscription $subscription
-     * @return void
-     */
-    public function addTopicSubscription(TopicSubscription $subscription): void;
+    public function countPendingOutgoingMessages(): int;
 
     /**
-     * Get all topic subscriptions with the given message identifier.
+     * Gets a pending outgoing message with the given message identifier, if found.
      *
      * @param int $messageId
-     * @return TopicSubscription[]
+     * @return PendingMessage|null
      */
-    public function getTopicSubscriptionsWithMessageId(int $messageId): array;
+    public function getPendingOutgoingMessage(int $messageId): ?PendingMessage;
 
     /**
-     * Find a topic subscription with the given topic.
+     * Gets a list of pending outgoing messages last sent before the given date time.
      *
-     * @param string $topic
-     * @return TopicSubscription|null
-     */
-    public function getTopicSubscriptionByTopic(string $topic): ?TopicSubscription;
-
-    /**
-     * Get all topic subscriptions matching the given topic.
-     *
-     * @param string $topic
-     * @return TopicSubscription[]
-     */
-    public function getTopicSubscriptionsMatchingTopic(string $topic): array;
-
-    /**
-     * Removes the topic subscription with the given topic from the repository.
-     * Returns true if a topic subscription existed and has been removed.
-     * Otherwise, false is returned.
+     * If date time is `null`, all pending messages are returned.
      *
-     * @param string $topic
-     * @return bool
-     */
-    public function removeTopicSubscription(string $topic): bool;
-
-    /**
-     * Returns the number of pending publish messages.
+     * The messages are returned in the same order they were added to the repository.
      *
-     * @return int
+     * @param DateTime|null $dateTime
+     * @return PendingMessage[]
      */
-    public function countPendingPublishMessages(): int;
+    public function getPendingOutgoingMessagesLastSentBefore(DateTime $dateTime = null): array;
 
     /**
-     * Adds a pending published message to the repository.
+     * Adds a pending outgoing message to the repository.
      *
-     * @param PublishedMessage $message
+     * @param PendingMessage $message
      * @return void
+     * @throws PendingMessageAlreadyExistsException
      */
-    public function addPendingPublishedMessage(PublishedMessage $message): void;
+    public function addPendingOutgoingMessage(PendingMessage $message): void;
 
     /**
-     * Gets a pending published message with the given message identifier, if found.
+     * Marks an existing pending outgoing published message as received in the repository.
      *
-     * @param int $messageId
-     * @return PublishedMessage|null
-     */
-    public function getPendingPublishedMessageWithMessageId(int $messageId): ?PublishedMessage;
-
-    /**
-     * Gets a list of pending published messages last sent before the given date time.
-     *
-     * @param DateTime $dateTime
-     * @return PublishedMessage[]
-     */
-    public function getPendingPublishedMessagesLastSentBefore(DateTime $dateTime): array;
-
-    /**
-     * Marks the pending published message with the given message identifier as received.
-     * If the message has no QoS level of 2, is not found or has already been received,
-     * false is returned. Otherwise the result will be true.
+     * If the message does not exists, an exception is thrown,
+     * otherwise `true` is returned if the message was marked as received, and `false`
+     * in case it was already marked as received.
      *
      * @param int $messageId
      * @return bool
+     * @throws PendingMessageNotFoundException
      */
-    public function markPendingPublishedMessageAsReceived(int $messageId): bool;
+    public function markPendingOutgoingPublishedMessageAsReceived(int $messageId): bool;
 
     /**
-     * Removes a pending published message from the repository. If a pending message
-     * with the given identifier is found and successfully removed from the repository,
-     * `true` is returned. Otherwise `false` will be returned.
+     * Removes a pending outgoing message from the repository.
+     *
+     * If a pending message with the given identifier is found and
+     * successfully removed from the repository, `true` is returned.
+     * Otherwise `false` will be returned.
      *
      * @param int $messageId
      * @return bool
      */
-    public function removePendingPublishedMessage(int $messageId): bool;
+    public function removePendingOutgoingMessage(int $messageId): bool;
 
     /**
-     * Returns the number of pending unsubscribe requests.
+     * Returns the number of pending incoming messages.
      *
      * @return int
      */
-    public function countPendingUnsubscribeRequests(): int;
+    public function countPendingIncomingMessages(): int;
 
     /**
-     * Adds a pending unsubscribe request to the repository.
-     *
-     * @param UnsubscribeRequest $request
-     * @return void
-     */
-    public function addPendingUnsubscribeRequest(UnsubscribeRequest $request): void;
-
-    /**
-     * Gets a pending unsubscribe request with the given message identifier, if found.
+     * Gets a pending incoming message with the given message identifier, if found.
      *
      * @param int $messageId
-     * @return UnsubscribeRequest|null
+     * @return PendingMessage|null
      */
-    public function getPendingUnsubscribeRequestWithMessageId(int $messageId): ?UnsubscribeRequest;
+    public function getPendingIncomingMessage(int $messageId): ?PendingMessage;
 
     /**
-     * Gets a list of pending unsubscribe requests last sent before the given date time.
+     * Adds a pending outgoing message to the repository.
      *
-     * @param DateTime $dateTime
-     * @return UnsubscribeRequest[]
+     * @param PendingMessage $message
+     * @return void
+     * @throws PendingMessageAlreadyExistsException
      */
-    public function getPendingUnsubscribeRequestsLastSentBefore(DateTime $dateTime): array;
+    public function addPendingIncomingMessage(PendingMessage $message): void;
 
     /**
-     * Removes a pending unsubscribe requests from the repository. If a pending request
-     * with the given identifier is found and successfully removed from the repository,
-     * `true` is returned. Otherwise `false` will be returned.
+     * Removes a pending incoming message from the repository.
+     *
+     * If a pending message with the given identifier is found and
+     * successfully removed from the repository, `true` is returned.
+     * Otherwise `false` will be returned.
      *
      * @param int $messageId
      * @return bool
      */
-    public function removePendingUnsubscribeRequest(int $messageId): bool;
+    public function removePendingIncomingMessage(int $messageId): bool;
 
     /**
-     * Returns the number of pending publish confirmations.
+     * Returns the number of registered subscriptions.
      *
      * @return int
      */
-    public function countPendingPublishConfirmations(): int;
+    public function countSubscriptions(): int;
 
     /**
-     * Adds a pending publish confirmation to the repository.
+     * Adds a subscription to the repository.
      *
-     * @param PublishedMessage $message
+     * @param Subscription $subscription
      * @return void
-     * @throws PendingPublishConfirmationAlreadyExistsException
      */
-    public function addPendingPublishConfirmation(PublishedMessage $message): void;
+    public function addSubscription(Subscription $subscription): void;
 
     /**
-     * Gets a pending publish confirmation with the given message identifier, if found.
+     * Gets all subscriptions matching the given criteria.
      *
-     * @param int $messageId
-     * @return PublishedMessage|null
+     * @param string|null $topicName
+     * @param int|null    $subscriptionId
+     * @return Subscription[]
      */
-    public function getPendingPublishConfirmationWithMessageId(int $messageId): ?PublishedMessage;
+    public function getMatchingSubscriptions(string $topicName = null, int $subscriptionId = null): array;
 
     /**
-     * Removes the pending publish confirmation with the given message identifier
-     * from the repository. This is normally done as soon as a transaction has been
-     * successfully finished by the publisher.
+     * Removes the subscription with the given topic filter from the repository.
      *
-     * @param int $messageId
+     * Returns `true` if a topic subscription existed and has been removed.
+     * Otherwise, `false` is returned.
+     *
+     * @param string $topicFilter
      * @return bool
      */
-    public function removePendingPublishConfirmation(int $messageId): bool;
+    public function removeSubscription(string $topicFilter): bool;
 }