Messenger not working if using processor on same resource in APIPlatform 3.1

Question

I am using API Platform 3.1. I need to generate thumbnails on some entities after creation. I want to use Symfony Messenger to trigger this and do it asynchronously since it could take time some to process, but I want the entity to be saved immediately.

When a resource uses both the messenger and a custom processor (to save the entity), either the messages are not created if using messenger=true, or the processor is not called if using messenger='input'.

How to reproduce

#[ORM\Entity]
#[ApiResource(
    operations: [
        new Post(
            processor: MyEntityProcessor::class,
            messenger: 'input',
            deserialize: false, 
        )
    ]
)]
class MyEntity
{
}

final class MyEntityProcessor implements ProcessorInterface
{
    public function __construct(private ProcessorInterface $persistProcessor)
    {
    }

    public function process($data, Operation $operation, array $uriVariables = [], array $context = [])
    {
        $result = $this->persistProcessor->process($data, $operation, $uriVariables, $context);
        return $result;
    }
}

final class MyEntityHandler implements MessageHandlerInterface
{
    public function __construct(private EntityManagerInterface $entityManager)
    {
    }

    public function __invoke(MyEntity $myEntity)
    {
        // my long running function

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

services.yaml

services:
    App\State\MyEntityProcessor:
        bind:
            $persistProcessor: '@api_platform.doctrine.orm.state.item_provider'

messenger.yaml

framework:
    messenger:
        transports:
            async: 'doctrine://default'

        routing:
            'App\Entity\MyEntity': async

API Platform documentation mentions in Symfony Messenger Integration:

Note: when using messenger=true ApiResource attribute in a Doctrine entity, the Doctrine Processor is not called. If you want the Doctrine Processor to be called, you should decorate a built-in state processor and implement your own logic.

It was suggested on a GitHub issue that I should decorate the messenger handler in order to save the entity. However, I need my entity to be saved immediately without waiting for it to be consumed by the messenger:consume worker.

Answer 1

A possible solution would be to dispatch the Message in your Processor after you have persisted the Entity for the first time. Maybe not as clean, since it requires more code, however you have way more control over how and when the Message is handled. In addition, you could also dispatch it with a DelayStamp.

In your processor I would dispatch the Message like this:

final class MyEntityProcessor implements ProcessorInterface
{
    
    public function __construct(
        private ProcessorInterface $persistProcessor,
        private MessageBusInterface $bus,
    ) {
    }

    public function process($data, Operation $operation, array $uriVariables = [], array $context = [])
    {
        $result = $this->persistProcessor->process($data, $operation, $uriVariables, $context);

        // $data should be your MyEntity Entity.
        $this->bus->dispatch(new GenerateThumbMessage($data), [
            // wait 5 seconds before processing
            new DelayStamp(5000),
        ]);
        
        return $result;
    }
}

Create the Message App\Message\GenerateThumbMessage.php:

class GenerateThumbMessage
{

    public function __construct(private readonly MyEntity $myEntity) {}

    public function getMyEntity(): MyEntity
    {
        return $this->myEntity;
    }

}

And then create your MessageHandler App\Message\GenerateThumbMessageHandler.php:

final class GenerateThumbMessageHandler implements MessageHandlerInterface
{
    public function __construct(private EntityManagerInterface $em) {}

    public function __invoke(GenerateThumbMessage $generateThumbMessage)
    {
        $myEntity = $generateThumbMessage->getMyEntity();
        
        // your long running function to generate a thumbnail

        $this->em->persist($myEntity);
        $this->em->flush();
    }
}

Your messenger.yaml config should look something like this:

framework:
    messenger:
        transports:
            async: 'doctrine://default'

        routing:
            'App\Message\GenerateThumbMessage': async

In the config, in stead of using the Entity, you have configured the Message to be processed async.

A similary solution can be found in the API Platform documentation