Обновление сущности в обработчике сообщений (Symfony)

Question

Я пытаюсь перенести свою тяжелую обработку на Messenger .

Это работает, но я хотел бы показать процент прогресса для моих пользователей. Я создал объект DownloadTask и пытаюсь обновить его во время процесса, но он не работает. Моя сущность не обновлена ​​в базе данных.

У вас есть какие-либо идеи?

Я все еще учусь, поэтому надеюсь, что мой код не сильно повредит вашим глазам.

<?php

namespace App\MessageHandler;

use App\Message\GetSellerList;
use App\Repository\DownloadTaskRepository;
use App\Service\EbayDL;
use Doctrine\DBAL\ConnectionException;
use Doctrine\ORM\EntityManagerInterface;
use DTS\eBaySDK\Trading\Types\GetSellerListResponseType;
use Psr\Log\LoggerInterface;
use Symfony\Component\Messenger\Handler\MessageHandlerInterface;
use GuzzleHttp\Promise;

class GetSellerListHandler implements MessageHandlerInterface
{
    /**
     * @var EbayDL
     */
    private $ebayDL;
    /**
     * @var EntityManagerInterface
     */
    private $entityManager;
    /**
     * @var LoggerInterface
     */
    private $logger;
    /**
     * @var \App\Entity\DownloadTask
     */
    private $task;
    /**
     * @var DownloadTaskRepository
     */
    private $downloadTaskRepository;

    public function __construct(EbayDL $ebayDL, EntityManagerInterface $entityManager, DownloadTaskRepository $downloadTaskRepository, LoggerInterface $logger)
    {
        $this->ebayDL = $ebayDL;
        $this->entityManager = $entityManager;
        $this->logger = $logger;
        $this->downloadTaskRepository = $downloadTaskRepository;
    }

    /**
     * @param GetSellerList $getSellerList
     * @throws ConnectionException
     * @throws \Exception
     * @throws \Throwable
     */
    public function __invoke(GetSellerList $getSellerList)
    {
        $task =  $this->downloadTaskRepository->find($getSellerList->getDownloadTaskId());

        $this->clearDatabase();
        $listingInfos = $this->ebayDL->getInformation();

        $totalNumberOfPages = $listingInfos["totalNumberOfPages"];

        $promises = (function () use ($totalNumberOfPages){
            for ($page = 1; $page <= $totalNumberOfPages; $page++) {
                yield $this->ebayDL->getProductsByPageAsync($page);
            }
        })();

        $eachPromise = new Promise\EachPromise($promises, [
            'concurrency' => 6,
            'fulfilled' => function(GetSellerListResponseType $response) {
                $products = $this->ebayDL->parseSellerListResponse($response);
                foreach ($products as $product) {
                    try {
                        $this->entityManager->persist($product);
                        $this->entityManager->flush();
                    }catch (\Exception $e) {
                        $this->logger->error("Failed to store a product");
                        $this->logger->error($e->getMessage());
                        if (!$this->entityManager->isOpen()) {
                            // https://stackoverflow.com/questions/14258591/the-entitymanager-is-closed
                            $this->entityManager = $this->entityManager->create(
                                $this->entityManager->getConnection(),
                                $this->entityManager->getConfiguration()
                            );
                        }
                    }
                }
            }
        ]);

        $eachPromise->promise()->wait();

        $this->task
            ->setProgress(100)
            ->setCompletedAt(new \DateTime('NOW'));

        $this->entityManager->flush();
    }

    /**
     * @throws ConnectionException
     * @throws \Doctrine\DBAL\DBALException
     */
    private function clearDatabase()
    {
        $connection = $this->entityManager->getConnection();
        $connection->beginTransaction();

        try {
            $connection->query('SET FOREIGN_KEY_CHECKS=0');
            $connection->query('DELETE FROM product');
            $connection->query('SET FOREIGN_KEY_CHECKS=1');
            $connection->commit();
        } catch (\Exception $exception) {
            $connection->rollback();
        }
    }
}