Support Symfony Messenger Component

[stephanvierkant]

This bundle can resolve images in background by using Enqueue. Any plans to support Symfony Messenger?

[michellesanver]

Love this idea! Currently no plans for it, but I don't see why it shouldn't go that direction. So if anyone feels motivated, feel free to add Symfony Messenger support! :)

[firstred]

I am going to look into this. Enqueue doesn't play well with my current stack, but Messenger does. I expect that I am not the only one for whom Messenger will be a great alternative.

What would need to be updated?

My guess right now:

• It looks like we are going to need some parameter to enable messenger support.
• Make ResolveCacheProcessor work with both Enqueue and Messenger.
• Suggest Symfony Messenger in composer.json.
• Enqueue already is decoupled from this bundle, which is excellent! No changes needed here.
• Documentation: https://symfony.com/doc/2.0/bundles/LiipImagineBundle/resolve-cache-images-in-background.html

Is there anything missing?

[dbu]

unless it ends up being very little code, i think this would best be a separate component that can be installed and maintained separately. if the bundle only has an interface, with the default implementation directly handling the job, it should be possible to plug whatever handler one wants.

then again, messenger is super flexible behind the scene, and possibly the messenger integration code is actually quite lean... i guess give it a try with that in mind and then lets consider when we see the merge request.

i would not want this bundle to provide both enqueue and messenger, that would be too much.

[firstred]

That's a great approach, actually! Thanks for your input.

As far as I understand, background processing is nothing more than a fire-and-forget approach that defers the processing to a consumer. This means that an interface that accepts just the payload (ResolveCache object) should be sufficient for any implementation to send it over a message bus and resolve the cache.

i would not want this bundle to provide both enqueue and messenger; that, that would be too much.

Indeed, it's hard enough already to figure out where Enqueue is used. To stay backward compatible, I would vote to keep a (deprecated) Enqueue implementation in the library and wire it by default. The Messenger (and a copy of the Enqueue) handler can then be kept in separate bundles to keep things simple.

[firstred]

If we choose that path, it might be a good idea to take #1237 into account.

[dbu]

good point. though #1237 would not be possible without a BC break (or introducing a new interface while phasing out the old one). i think such a change would merit a new major version, so basically if you have a proposal for the new interface signature and a symfony messenger implementation that validates that the new interface fulfills the requirements for good async processing, that would be very cool!

[firstred]

It looks like that is the only option. With the dependency on the Enqueue producer gone, we now need something else to signal that we are going 'async.' In the actual implementation, that will likely come down to dispatching the message (the Messenger way) or sending a command (the Enqueue way).

The new interface (the 'dispatcher,' or 'async signal' -- have yet to find the right name) also has the advantage that we could introduce an option to enable background processing by default.

Now that we have our own 'dispatcher,' we should 'dispatch' just enough information for the implementation to be able to decide which route (bus) to take or when to 'dispatch,' for example. The Symfony Messenger makes it possible to set priorities. I can imagine users will want to give image processing a lower priority than, e.g., sending emails.

ResolveCacheProcessor will likely have to be deprecated, because it is coupled too tightly with Enqueue. A more agnostic version that only contains the processing can then used directly in the new components. Let the components handle the transport, LiipImagineBundle the processing.

The Commands and Topic classes seem to be meant for Enqueue. We can keep them for backward compatibility, but I don't see how they can be useful in future versions. I would recommend to deprecate them as well.

[firstred]

Thinking about using a Promises/A replacement to replace the receive function in step 4. This makes sure you can wait for the background process to finish in a modern and clean way.

Although it is not recommended to halt execution to wait for the process to finish (remember that we want to go async), I can understand that there would be several use cases for it. You can give processing a very low priority with Symfony's Messenger, so there will be cases when the promise does not resolve in time and the request times out before the cache is resolved. To take into account that processing can take a very long time, we have to add a wait timeout.

ReactPHP's Promises/A implementation looks like the right candidate.
https://github.com/reactphp/promise
Together with its support for wait timeouts it should be a breeze for implementations to replace the current receive functionality. Additionally, it supports native PHP event loop extensions, which improves performance, too.

With this Promises/A implementation:

$producer = $container->get('enqueue.producer');
$producer->sendCommand(Commands::RESOLVE_CACHE, new ResolveCache('the/path/img.png'));
$replyMessage = $reply->receive(20000); // wait for 20 sec

would then very likely become:

$replyPromise = $this->dispatcher->resolveInBackground(new ResolveCache('the/path/img.png'))
$loop = React\EventLoop\Factory::create();
Timer\timeout($replyPromise, 20.0, $loop)
    ->then(function ($value) {
        // the operation finished within 10.0 seconds
    })
    ->otherwise(function (Timer\TimeoutException $error) {
        // the operation has failed due to a timeout
    })
    ->otherwise(function ($error) {
        // the input operation has failed due to some other error
    })
;

Too bad there isn't a Promises PSR, yet. If that were the case, we could slim this down even further by just returning a PSR promise and let you choose whatever Promises library you like.

With this last step complete, I think there is now enough information to start drafting the first abstraction.

[firstred]

Too bad there isn't a Promises PSR, yet. If that were the case, we could slim this down even further by just returning a PSR promise and let you choose whatever Promises library you like.

And maybe we should do just that. Add a Promises/A interface and leave it at that (but do try to support both Guzzle and ReactPHP Promises, if possible). If you still want to wait for the background processes to finish, you can then include one of those libraries yourself.

[dbu]

yeah, promises is its own can of worms... we had the same problem while designing async http requests for PSR-18. in the end we had to decide to not define async because the http psr can't define promises.

just to think about options in architecture: for https://rokka.io we use the concept that on request, you get a "quick" scaling of the image which is a bit less nice and much less memory efficient but returns within milliseconds, and that image is cached only for a short time. we then have a background process that applies all kind of fancy image optimization processes with the amazon lambda system to eventually produce a highly optimized image which gets cached for a long time.

would be nice if the architecture we build here would also allow something like that, get a quick and dirty image immediately but trigger a nice rendering in the background...

maybe if we have those 2 options (immediate or deferred), we could do a simple interface that allows to run promises underneath but also allows the default handler to do everything synchronously without needing a promise library? basically ImageRendering::ready(): bool and ImageRendering::get(): Image - and it would be up to the consumer to know if with their configuration the get() should be called or not.

[firstred]

maybe if we have those 2 options (immediate or deferred), we could do a simple interface that allows to run promises underneath but also allows the default handler to do everything synchronously without needing a promise library? basically ImageRendering::ready(): bool and ImageRendering::get(): Image - and it would be up to the consumer to know if with their configuration the get() should be called or not.

Great idea! I'll try to take that into account.

Every single time I try to post an update here I suddenly find a better design to handle all this. So I'll just try to make a PR next time. Should be pretty close now, I think.

[laferte-tech]

Hi, any update maybe ?
Thanks a lot.

[ivanbogomoloff]

Hello all.
My workaround here.

Create DTO like this

<?php

namespace App\Message;

/**
 * Class LiipImageResolveCache
 * @package App\Message
 */
class LiipImageResolveCache
{
    /** @var string */
    private $pathToImage;
    /** @var array */
    private $filters;
    /** @var bool  */
    private $force = false;
    /**
     * LiipImageResolveCache constructor.
     *
     * @param       $pathToImage
     * @param array $filters
     */
    public function __construct($pathToImage, array $filters = [], bool $force = false)
    {
        $this->pathToImage = $pathToImage;
        $this->filters = $filters;
        $this->force = $force;
    }

    public function isForce(): bool
    {
        return $this->force;
    }

    /**
     * @return mixed
     */
    public function getPath()
    {
        return $this->pathToImage;
    }

    /**
     * @return array
     */
    public function getFilters(): array
    {
        return $this->filters;
    }
}

then in file upload logic or anything else push message to queue by messenger like this:

<?php
namespace App\Service;

use Symfony\Component\Messenger\MessageBusInterface;
use App\Message\LiipImageResolveCache;

class FileManagerService {
  private $messageBus;
  public function __construct(MessageBusInterface $messageBus)
      {
          $this->messageBus = $messageBus;
      }

  public function upload() {
     //upload logic then resize
     $this->messageBus->dispatch(new LiipImageResolveCache('/path/to/file'));
  }
}

then create handler like this:

<?php
namespace App\MessageHandler;

ini_set('memory_limit', '512M');

use App\Message\LiipImageResolveCache;
use Liip\ImagineBundle\Imagine\Filter\FilterManager;
use Liip\ImagineBundle\Service\FilterService;
use Symfony\Component\Messenger\Handler\MessageHandlerInterface;

class LiipImageHandler implements MessageHandlerInterface
{
    /** @var FilterManager  */
    private $filterManager;
    /** @var FilterService */
    private $filterService;
    
    public function __construct(
        FilterManager $filterManager,
        FilterService $filterService)
    {
        $this->filterManager = $filterManager;
        $this->filterService = $filterService;
    }
    /**
     * @param LiipImageResolveCache $message
     */
    public function __invoke(LiipImageResolveCache $message)
    {
        $path = $message->getPath();
        $filters = $message->getFilters() ?: array_keys($this->filterManager->getFilterConfiguration()->all());
        foreach ($filters as $filter)
        {
            if ($message->isForce()) {
                $this->filterService->bustCache($path, $filter);
            }

            $results[$filter] = $this->filterService->getUrlOfFilteredImage($path, $filter);
        }
    }
}

then register this handler in services.yaml

App\MessageHandler\LiipImageHandler:
      arguments:
        - '@liip_imagine.filter.manager'
        - '@liip_imagine.service.filter'

then register queue in messenger for this operations:

# config/packages/messenger.yaml
framework:
    messenger:
        transports:
             liip_image:
                dsn: '%env(MESSENGER_TRANSPORT_DSN)%'

        routing:
             'App\Message\LiipImageResolveCache': liip_image

then try it

bin/console -vvv messenger:consume liip_image

[mynameisbogdan]

@ivanbogomoloff it's similar to what I did in #1360 👍

[dbu]

symfony messenger support has been added in #1360