BenTools\MercurePHP\Storage\Redis\RedisStorage

Complexity

Total Complexity

17

Size/Duplication

Total Lines	132
Duplicated Lines	0 %

Test Coverage

Coverage

97.44%

Importance

Changes

0

10 Methods

Rating	Name	Duplication	Size	Complexity
A	findSubscriptionsBySubscriber()	0	17	2
A	removeSubscriptions()	0	11	2
A	findNextMessages()	0	32	4
A	storeMessage()	0	10	1
A	storeLastEventId()	0	10	2
A	createSubscriptionKey()	0	6	1
A	getLastEventId()	0	3	1
A	retrieveMessagesAfterId()	0	3	1
A	__construct()	0	4	1
A	storeSubscriptions()	0	11	2

<?php

namespace BenTools\MercurePHP\Storage\Redis;

use BenTools\MercurePHP\Model\Message;
use BenTools\MercurePHP\Model\Subscription;
use BenTools\MercurePHP\Security\TopicMatcher;
use BenTools\MercurePHP\Storage\StorageInterface;
use Clue\React\Redis\Client as AsynchronousClient;
use Predis\Client as SynchronousClient;
use React\Promise\PromiseInterface;

use function React\Promise\all;
use function React\Promise\resolve;

final class RedisStorage implements StorageInterface
{
    private AsynchronousClient $async;
    private SynchronousClient $sync;

    public function __construct(AsynchronousClient $asyncClient, SynchronousClient $syncClient)
    {
        $this->async = $asyncClient;
        $this->sync = $syncClient;
    }

    public function retrieveMessagesAfterId(string $id, array $subscribedTopics): PromiseInterface
    {
        return resolve($this->findNextMessages($id, $subscribedTopics));
    }

    public function storeMessage(string $topic, Message $message): PromiseInterface
    {
        $id = $message->getId();
        $payload = \json_encode($message, \JSON_THROW_ON_ERROR);

        /** @phpstan-ignore-next-line */
        return $this->async->set('data:' . $id, $topic . \PHP_EOL . $payload)
            ->then(fn() => $this->getLastEventId())
            ->then(fn(?string $lastEventId) => $this->storeLastEventId($lastEventId, $id))
            ->then(fn() => $id);
    }

    public function storeSubscriptions(array $subscriptions): PromiseInterface
    {
        $promises = [];

        foreach ($subscriptions as $subscription) {
            $key = $this->createSubscriptionKey($subscription);
            /** @phpstan-ignore-next-line */
            $promises[] = $this->async->set($key, \json_encode($subscription, \JSON_THROW_ON_ERROR));
        }

        return all($promises);
    }

    public function removeSubscriptions(iterable $subscriptions): PromiseInterface
    {
        $promises = [];

        foreach ($subscriptions as $subscription) {
            $key = $this->createSubscriptionKey($subscription);
            /** @phpstan-ignore-next-line */
            $promises[] = $this->async->del($key);
        }

        return all($promises);
    }

    public function findSubscriptionsBySubscriber(string $subscriber): PromiseInterface
    {
        $keyPattern = \sprintf(
            'subscription:%s:*',
            $subscriber,
        );

        /** @phpstan-ignore-next-line */
        return $this->async->keys($keyPattern)
            ->then(
                function (array $keys) {
                    $promises = [];
                    foreach ($keys as $key) {
                        $promises[] = $this->async->get($key); /** @phpstan-ignore-line */
                    }

                    return all($promises);
                }
            );
    }

    private function getLastEventId(): PromiseInterface
    {
        return $this->async->get('Last-Event-ID'); /** @phpstan-ignore-line */
    }

    private function storeLastEventId(?string $previousEventId, string $newEventId): PromiseInterface
    {
        $promise = $this->async->set('Last-Event-ID', $newEventId); /** @phpstan-ignore-line */

        if (null === $previousEventId) {
            return $promise;
        }

        /** @phpstan-ignore-next-line */
        return $promise->then(fn() => $this->async->set('next:' . $previousEventId, $newEventId));
    }

    private function findNextMessages(string $id, array $subscribedTopics): iterable
    {
        $nextId = $this->sync->get('next:' . $id);

        /** @phpstan-ignore-next-line */
        if (null === $nextId) {
            return [];
        }

        $payload = $this->sync->get('data:' . $nextId);

        /** @phpstan-ignore-next-line */
        if (null === $payload) {
            return [];
        }

        $item = \explode(\PHP_EOL, $payload);
        $topic = \array_shift($item);
        $message = Message::fromArray(
            \json_decode(
                \implode(\PHP_EOL, $item),
                true,
                512,
                \JSON_THROW_ON_ERROR
            )
        );

        if (TopicMatcher::matchesTopicSelectors($topic, $subscribedTopics)) {
            yield $topic => $message;
        }

        yield from $this->findNextMessages($message->getId(), $subscribedTopics); // Sync client needed because of this
    }

    private function createSubscriptionKey(Subscription $subscription): string
    {
        return \sprintf(
            'subscription:%s:%s',
            $subscription->getSubscriber(),
            \hash('crc32', $subscription->getId())
        );
    }
}