В статье я создал атрибут, чтобы помечать операции API для выполнения в фоновом режиме. Когда аннотированная операция вызывалась как фоновая, ее выполнение задерживалось с помощью symfony messenger, клиент получал в ответ код состояния HTTP 202 Accepted.

В этой статье я внесу некоторые изменения, которые позволят операции уведомлять пользователя о завершении. Чтобы реализовать это, я буду использовать mercure, так как symfony очень хорошо с ним интегрируется.

Я собираюсь опустить код из некоторых классов, чтобы сосредоточиться только на модификациях. Если вы хотите посмотреть весь код, вы можете найти его в предыдущих статьях: Часть 1, Часть 2, Часть 3, Часть 4, Часть 5.

Установка mercure

Прежде всего, давайте начнем с установки mercure с помощью composer:

composer require mercure

Symfony предоставляет нам установку хаба mercure из docker. После установки мы можем взглянуть на файл docker-compose.yaml где мы можем увидеть сервис mercure. Чтобы запустить хаб, нам нужно всего лишь выполнить эту команду в корне нашего проекта:

docker-compose up -d

Она создаст и запустит контейнер для каждого сервиса вашего docker-compose файла.

Подготовка операции

Теперь, когда у нас запущен наш хаб mercure, мы должны указать операции, о завершении которых мы хотим уведомить пользователя. Для этого создадим следующий интерфейс:

interface ApiOperationNotificationInterface
{
    public function getNotificationData(): string;
    public function getTopic(): string;
}

Операции, которые реализуют вышеуказанный интерфейс, должны будут реализовать два метода:

getNotificationData(): возвращает данные для отправки в mercure.

getTopic(): возвращает топик, в который были отправлены данные.

Согласно документации, топики должны соответствовать формату Интернационализированный идентификатор ресурса (IRL). В нашем случае я собираюсь отформатировать топики следующим образом:

https://<mydomain>/<userIdenfier>/<subject> где:

mydomain: домен, который мы выбираем.
userIdentifier: идентификатор пользователя.
subject: о каком типе операции мы уведомляем, например, “платежи”.

Благодаря userIdentifier мы можем разделять топики по каждому пользователю, чтобы каждый пользователь подписывался на свои топики.

Давайте создадим фоновую операцию, которая реализует APINotificationOperationInterface:

#[IsBackground]
class SendPaymentOperation implements ApiOperationInterface, ApiOperationNotificationInterface
{

    private SentPaymentNotification $notification;

    public function perform(ApiInput $apiInput): ApiOutput
    {
        
        // Какой-нибудь код, который обрабатывает платежи
        
        $this->notification = new SentPaymentNotification('Your payment has been processed successfully', SentPaymentNotification::OK);
        return new ApiOutput([], Response::HTTP_OK);
    }

    public function getName(): string
    {
        return 'SendPaymentOperation';
    }

    public function getInput(): string
    {
        return '';
    }

    public function getGroup(): ?string
    {
        return null;
    }


    public function getNotificationData(): string
    {
        return $this->notification;
    }

    public function getTopic(): string
    {
        return 'https://topics.domain.com/%s/payments';
    }
}

Операция использует объект класса SentPaymentNotification для хранения данных уведомлений, которые мы хотим отправить клиенту. Давайте посмотрим, как это выглядит:

class SentPaymentNotification
{
    const OK = 'OK';
    const KO = 'KO';

    public function __construct(
        private readonly string $message,
        private readonly string $status
    ){ }

    public function __toString(): string
    {
        return json_encode(['message' => $this->message, 'status' => $this->status]);
    }
}

Как мы можем видеть, SendPaymentNotification просто встраивает json с параметрами, которые он получает через конструктор, в метод __toString. Используя этот способ, мы можем сделать для класса строковое представление, которое мы хотим.

Внесение изменений в обработчик операций и обработчик сообщений

Поскольку топики требуют userIdentifier, мы должны будем передать его ApiOperationMessage, поэтому давайте добавим новый параметр в его конструктор:

class ApiOperationMessage
{
    public function __construct(
        public readonly ApiInput $apiInput,
        public readonly string $userUuid
    ){ }
}

В этом случае, ApiOperationMessageHandler будет иметь доступ к пользовательскому uuid. Давайте изменим ApiOperationHandler, таким образом, чтобы он передавал uuid пользователя ApiOperationMessage при отправке операции в фоновый режим:

class ApiOperationHandler
{
    private ApiOperationsCollection $apiOperationsCollection;

    private Security $security;
    private MessageBusInterface $bus;

    public function __construct(ApiOperationsCollection $apiOperationsCollection, Security $security, MessageBusInterface $bus)
    {
        $this->apiOperationsCollection = $apiOperationsCollection;
        $this->security = $security;
        $this->bus = $bus;
    }

    /**
     * @throws \ReflectionException
     */
    public function performOperation(ApiInput $apiInput): ApiOutput
    {
        // ....... Остальной код

        $attribute = $this->getBackgroundAttr($operationHandler);
        if($attribute){
            $stamps = ($attribute->delay > 0) ? [new DelayStamp($attribute->delay * 1000)] : [];
            $this->bus->dispatch(new ApiOperationMessage($apiInput, $this->security->getUser()->getUuid()), $stamps);
            return new ApiOutput(
                ['status' => 'Queued'],
                Response::HTTP_ACCEPTED
            );
        }



        return $operationHandler->perform($apiInput);
    }

    // .......
}

Теперь давайте посмотрим, как ApiOperationMessageHandler ведет себя, когда операция реализует ApiOperationNotificationInterface.

#[AsMessageHandler]
class ApiOperationMessageHandler
{
    public function __construct(
        private readonly ApiOperationsCollection $apiOperationsCollection,
        private readonly LockHandler $lockHandler,
        private readonly MessageBusInterface $bus,
        private readonly AttributeHelper $attributeHelper,
        private readonly HubInterface $hub
    ){ }


    /**
     * @throws \RedisException
     */
    public function __invoke(ApiOperationMessage $message): void
    {

        // Остальной код  .......

        if($operation instanceof ApiOperationNotificationInterface){

            $topic = sprintf($operation->getTopic(), Uuid::fromString($message->userUuid)->toBase32());

            $this->hub->publish(
                new Update(
                    $topic,
                    $operation->getNotificationData()
                )
            );
        }
    }
}

В конце метода __invoke, мы проверяем, реализует ли операция ApiOperationNotificationInterface.Если это так, мы генерируем топик с uuid пользователя в кодировке base32 и публикуем в хабе с данными, которые мы получаем из метода getNotificationData().

Для работы с uuids я установил symfony uid component.

Создание кастомного поставщика пользователей

Как вы помните из статьи, мы создали нашего пользователя, используя in memory users symfony, поскольку это служило нам хорошим примером в рамках той статьи.

Проблема здесь в том, что in_memory_users не разрешает никаких полей, кроме identifier и roles, поэтому мы создали собственного кастомного поставщика пользователей, чтобы иметь возможность добавлять пользователю uuid. Давайте посмотрим, как это будет выглядеть:

class UserProvider implements UserProviderInterface
{

    public function refreshUser(UserInterface $user)
    {
        // TODO: Реализовать метод refreshUser().
    }

    public function supportsClass(string $class)
    {
        return $class === User::class;
    }

    public function loadUserByIdentifier(string $identifier): UserInterface
    {
        $users = [
            [
                'identifier' => "fYDg7nMhlvxAtL7KfBnS",
                'roles'      => ["ROLE_USER", "ROLE_SUPERUSER"],
                'uuid'       => 'c5b6206c-340c-41e6-abed-18ca2258c578'
            ]
        ];

        $users = array_filter($users, fn($u) => $u['identifier'] === $identifier);
        if(count($users) == 0){
            throw new UserNotFoundException('User does not exists');
        }

        return new User(array_pop($users));
    }
}

Этот кастомный поставщик пользователей попросту имеет массив с пользователем, который был у нас в in_memory_users, и использует функцию array_filter для поиска пользователя, у которого подходит идентификатор. Если ни один пользователь не найден, пробрасывается UserNotFoundHttpException, в противном случае возвращается объект User. Давайте посмотрим на объект User:

class User implements UserInterface
{

    private string $uuid;
    private array $roles;
    private string $identifier;

    public function __construct(array $userData)
    {
        $this->uuid = $userData['uuid'];
        $this->roles = $userData['roles'];
        $this->identifier = $userData['identifier'];
    }


    public function getRoles(): array
    {
        return $this->roles;
    }

    public function eraseCredentials()
    {

    }

    public function getUserIdentifier(): string
    {
        return $this->identifier;
    }

    public function getUuid(): string
    {
        return $this->uuid;
    }

}

Класс User реализует UserInterface Symfony. Мы должны реализовать его, так как метод loadUserByIdentifier поставщика должен возвращать объект, который реализует этот интерфейс.

Чтобы иметь возможность использовать этого поставщика, мы должны указать его в файле symfony security.yaml:

providers:
        custom_provider:
            id: App\Security\Provider\UserProvider
    firewalls:
        dev:
            pattern: ^/(_(profiler|wdt)|css|images|js)/
            security: false
        api:
            pattern: '^/api/v1/operation'
            provider: custom_provider
            access_denied_handler: App\Security\AccessDeniedHandler
            custom_authenticators:
                - App\Security\ApiTokenAuthenticator

Проверяем, можем ли мы подписаться на уведомления

Теперь, чтобы проверить, работает ли подписка на уведомления, давайте создадим маршрут, который отображает twig-шаблон, который подписывается на наш топик платежей пользователей:

<script>
    const eventSource = new EventSource("{{ mercure('https://topics.domain.com/65PRG6RD0C87KAQV8RS8H5HHBR/payments')|escape('js') }}");
    eventSource.onmessage = event => {
        // Будет вызываться каждый раз, когда сервер что-либо публикует
        console.log(JSON.parse(event.data));
    }
</script>

Это тот же скрипт, который мы можем найти в документации. Единственное изменение, которое я сделал, это изменение топика IRI. В URL топика мы можем увидеть uuid пользователя в кодировке base32 (65PRG6RD0C87KAQV8RS8H5HHBR).

Теперь давайте создадим простой маршрут, который отображает этот twig:

#[Route('/admin/topics', name: 'get_topics', methods: ['GET'])]
public function getTopics(): Response
{
    return $this->render('topics.html.twig');
}

Чтобы использовать twig-шаблон, вам необходимо установить пакет twig: composer2 требует symfony/twig-bundle. Вы можете узнать больше о symfony и twig здесь

Теперь давайте отправим запрос API для операции SendPaymentOperation, который опубликует уведомление:

Теперь давайте вызовем маршрут /admin/topics, чтобы мы могли видеть, как сообщение появляется в консоли браузера:

И все в порядке, мы можем подписаться на наши топики и видеть уведомления, отправленные в хаб. Это очень важно для фронтенд-части, так как они могут предоставлять обратную связь конечным пользователям.

В заключение статьи приглашаем на открытое занятие «Обновление до php 8.1 с помощью rector», который состоится скоро в рамках онлайн-курса "Symfony Framework". На этом уроке мы:

применим rector к коду на php 8 для автоматического использования возможностей 8.1,
рассмотрим нюансы перехода с myclabs/php-enum на built-in enum,
обсудим readonly-свойства и их влияние на тесты.

Регистрация на урок.

Отправка уведомлений из Symfony API с помощью Mercure