CMS 2.0

2026-01-06 10:02:25 +01:00
parent f685c2490a
commit b52d3a11be
111 changed files with 12830 additions and 76 deletions
@@ -0,0 +1,232 @@
+<?php
+
+declare(strict_types=1);
+
+namespace PhpMqtt\Client\Repositories;
+
+use PhpMqtt\Client\Contracts\Repository;
+use PhpMqtt\Client\Exceptions\PendingMessageAlreadyExistsException;
+use PhpMqtt\Client\Exceptions\PendingMessageNotFoundException;
+use PhpMqtt\Client\Exceptions\RepositoryException;
+use PhpMqtt\Client\PendingMessage;
+use PhpMqtt\Client\PublishedMessage;
+use PhpMqtt\Client\Subscription;
+
+/**
+ * Provides an in-memory implementation which manages message ids, subscriptions and pending messages.
+ * Instances of this type do not persist any data and are only meant for simple uses cases.
+ *
+ * @package PhpMqtt\Client\Repositories
+ */
+class MemoryRepository implements Repository
+{
+    private int $nextMessageId = 1;
+
+    /** @var array<int, PendingMessage> */
+    private array $pendingOutgoingMessages = [];
+
+    /** @var array<int, PendingMessage> */
+    private array $pendingIncomingMessages = [];
+
+    /** @var array<int, Subscription> */
+    private array $subscriptions = [];
+
+    /**
+     * {@inheritDoc}
+     */
+    public function reset(): void
+    {
+        $this->nextMessageId           = 1;
+        $this->pendingOutgoingMessages = [];
+        $this->pendingIncomingMessages = [];
+        $this->subscriptions           = [];
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    public function newMessageId(): int
+    {
+        if (count($this->pendingOutgoingMessages) >= 65535) {
+            // This should never happen, as the server receive queue is
+            // normally smaller than the actual total number of message ids.
+            // Also, when using MQTT 5.0 the server can specify a smaller
+            // receive queue size (mosquitto for example has 20 by default),
+            // so the client has to implement the logic to honor this
+            // restriction and fallback to the protocol limit.
+            throw new RepositoryException('No more message identifiers available. The queue is full.');
+        }
+
+        while (isset($this->pendingOutgoingMessages[$this->nextMessageId])) {
+            $this->nextMessageId++;
+            if ($this->nextMessageId > 65535) {
+                $this->nextMessageId = 1;
+            }
+        }
+
+        return $this->nextMessageId;
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    public function countPendingOutgoingMessages(): int
+    {
+        return count($this->pendingOutgoingMessages);
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    public function getPendingOutgoingMessage(int $messageId): ?PendingMessage
+    {
+        return $this->pendingOutgoingMessages[$messageId] ?? null;
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    public function getPendingOutgoingMessagesLastSentBefore(?\DateTime $dateTime = null): array
+    {
+        $result = [];
+
+        foreach ($this->pendingOutgoingMessages as $pendingMessage) {
+            if ($pendingMessage->getLastSentAt() < $dateTime) {
+                $result[] = $pendingMessage;
+            }
+        }
+
+        return $result;
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    public function addPendingOutgoingMessage(PendingMessage $message): void
+    {
+        if (isset($this->pendingOutgoingMessages[$message->getMessageId()])) {
+            throw new PendingMessageAlreadyExistsException($message->getMessageId());
+        }
+
+        $this->pendingOutgoingMessages[$message->getMessageId()] = $message;
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    public function markPendingOutgoingPublishedMessageAsReceived(int $messageId): bool
+    {
+        if (!isset($this->pendingOutgoingMessages[$messageId]) ||
+            !$this->pendingOutgoingMessages[$messageId] instanceof PublishedMessage) {
+            throw new PendingMessageNotFoundException($messageId);
+        }
+
+        return $this->pendingOutgoingMessages[$messageId]->markAsReceived();
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    public function removePendingOutgoingMessage(int $messageId): bool
+    {
+        if (!isset($this->pendingOutgoingMessages[$messageId])) {
+            return false;
+        }
+
+        unset($this->pendingOutgoingMessages[$messageId]);
+        return true;
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    public function countPendingIncomingMessages(): int
+    {
+        return count($this->pendingIncomingMessages);
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    public function getPendingIncomingMessage(int $messageId): ?PendingMessage
+    {
+        return $this->pendingIncomingMessages[$messageId] ?? null;
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    public function addPendingIncomingMessage(PendingMessage $message): void
+    {
+        if (isset($this->pendingIncomingMessages[$message->getMessageId()])) {
+            throw new PendingMessageAlreadyExistsException($message->getMessageId());
+        }
+
+        $this->pendingIncomingMessages[$message->getMessageId()] = $message;
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    public function removePendingIncomingMessage(int $messageId): bool
+    {
+        if (!isset($this->pendingIncomingMessages[$messageId])) {
+            return false;
+        }
+
+        unset($this->pendingIncomingMessages[$messageId]);
+        return true;
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    public function countSubscriptions(): int
+    {
+        return count($this->subscriptions);
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    public function addSubscription(Subscription $subscription): void
+    {
+        // Remove a potentially existing subscription for this topic filter.
+        $this->removeSubscription($subscription->getTopicFilter());
+
+        $this->subscriptions[] = $subscription;
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    public function getSubscriptionsMatchingTopic(string $topicName): array
+    {
+        $result = [];
+
+        foreach ($this->subscriptions as $subscription) {
+            if (!$subscription->matchesTopic($topicName)) {
+                continue;
+            }
+
+            $result[] = $subscription;
+        }
+
+        return $result;
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    public function removeSubscription(string $topicFilter): bool
+    {
+        foreach ($this->subscriptions as $index => $subscription) {
+            if ($subscription->getTopicFilter() === $topicFilter) {
+                unset($this->subscriptions[$index]);
+                return true;
+            }
+        }
+
+        return false;
+    }
+}