Resolve image caches in background. Using message queue.

add tests, add on success message, add force

.travis.yml

@@ -55,6 +55,7 @@ before_install:
   - if [ "${TRAVIS_PHP_VERSION}" != "hhvm" ] && [ "${TRAVIS_PHP_VERSION:0:3}" != "5.3" ]; then composer require --no-update --dev league/flysystem:~1.0; fi;
   - if [ "${TRAVIS_PHP_VERSION}" != "hhvm" ] && [ "${TRAVIS_PHP_VERSION:0:1}" != "7" ]; then yes "" | pecl -q install -f mongo; composer require --no-update --dev doctrine/mongodb-odm:~1.0; fi;
   - if [ "${TRAVIS_PHP_VERSION}" != "hhvm" ] && [ "${TRAVIS_PHP_VERSION:0:1}" == "7" ]; then yes "" | pecl -q install -f mongodb; travis_retry composer require --dev alcaeus/mongo-php-adapter:~1.0; composer require --no-update --dev doctrine/mongodb-odm:~1.0; fi
+  - if [ "${TRAVIS_PHP_VERSION}" != "hhvm" ] && [ "${TRAVIS_PHP_VERSION:0:3}" -ge "5.6" ]; then composer require --no-update --dev enqueue/enqueue-bundle:^0.3.6; fi;
 
 install:
   - travis_retry composer update $COMPOSER_FLAGS

Async/ResolveAllCacheProcessor.php

@@ -0,0 +1,117 @@
+<?php
+namespace Liip\ImagineBundle\Async;
+
+use Enqueue\Client\ProducerInterface;
+use Enqueue\Client\TopicSubscriberInterface;
+use Enqueue\Consumption\QueueSubscriberInterface;
+use Enqueue\Consumption\Result;
+use Enqueue\Psr\PsrContext;
+use Enqueue\Psr\PsrMessage;
+use Enqueue\Psr\PsrProcessor;
+use Enqueue\Util\JSON;
+use Liip\ImagineBundle\Imagine\Cache\CacheManager;
+use Liip\ImagineBundle\Imagine\Data\DataManager;
+use Liip\ImagineBundle\Imagine\Filter\FilterManager;
+
+class ResolveAllCacheProcessor implements PsrProcessor, TopicSubscriberInterface, QueueSubscriberInterface
+{
+    /**
+     * @var CacheManager
+     */
+    private $cacheManager;
+
+    /**
+     * @var FilterManager
+     */
+    private $filterManager;
+
+    /**
+     * @var DataManager
+     */
+    private $dataManager;
+
+    /**
+     * @var ProducerInterface
+     */
+    private $producer;
+
+    /**
+     * @param CacheManager $cacheManager
+     * @param FilterManager $filterManager
+     * @param DataManager $dataManager
+     * @param ProducerInterface $producer
+     */
+    public function __construct(
+        CacheManager $cacheManager,
+        FilterManager $filterManager,
+        DataManager $dataManager,
+        ProducerInterface $producer
+    ) {
+        $this->cacheManager = $cacheManager;
+        $this->filterManager = $filterManager;
+        $this->dataManager = $dataManager;
+        $this->producer = $producer;
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function process(PsrMessage $psrMessage, PsrContext $psrContext)
+    {
+        try {
+            $body = JSON::decode($psrMessage->getBody());
+        } catch (\Exception $e) {
+            return Result::reject($e->getMessage());
+        }
+
+        $body = array_replace(['path' => null, 'force' => false], $body);
+
+        if (false == $body['path']) {
+            return Result::reject('The message does not contain "path" but it is required.');
+        }
+
+        $path = $body['path'];
+        $results = [];
+        foreach ($this->filterManager->getFilterConfiguration()->all() as $filter => $config) {
+            if ($this->cacheManager->isStored($path, $filter) && $body['force']) {
+                $this->cacheManager->remove($path, $filter);
+            }
+
+            if (false == $this->cacheManager->isStored($path, $filter)) {
+                $binary = $this->dataManager->find($filter, $path);
+                $this->cacheManager->store(
+                    $this->filterManager->applyFilter($binary, $filter),
+                    $path,
+                    $filter
+                );
+            }
+
+            $results[$filter] = $this->cacheManager->resolve($path, $filter);
+        }
+
+        $this->producer->send(Topics::ALL_CACHE_RESOLVED, [
+            'path' => $path,
+            'filters' => $results,
+        ]);
+
+        return self::ACK;
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public static function getSubscribedTopics()
+    {
+        return [
+            Topics::RESOLVE_ALL_CACHE => ['queueName' => Topics::RESOLVE_ALL_CACHE,  'queueNameHardcoded' => true]
+        ];
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public static function getSubscribedQueues()
+    {
+        return [Topics::RESOLVE_ALL_CACHE];
+    }
+}

Async/Topics.php

@@ -0,0 +1,9 @@
+<?php
+namespace Liip\ImagineBundle\Async;
+
+class Topics
+{
+    const RESOLVE_ALL_CACHE = 'liip_imagine_resolve_all_cache';
+
+    const ALL_CACHE_RESOLVED = 'liip_imagine_all_cache_resolve_d';
+}

DependencyInjection/Configuration.php

@@ -158,6 +158,7 @@ public function getConfigTreeBuilder()
                     ->end()
                 ->end()
             ->end()
+            ->booleanNode('enqueue')->defaultFalse()->info('Enables integration with enqueue if set true. Allows resolve image caches in background by sending messages to MQ.')->end()
         ->end();
 
         return $treeBuilder;

DependencyInjection/LiipImagineExtension.php

@@ -72,6 +72,10 @@ public function load(array $configs, ContainerBuilder $container)
         $loader = new XmlFileLoader($container, new FileLocator(__DIR__.'/../Resources/config'));
         $loader->load('imagine.xml');
 
+        if ($config['enqueue']) {
+            $loader->load('enqueue.xml');
+        }
+
         $this->setFactories($container);
 
         if (interface_exists('Imagine\Image\Metadata\MetadataReaderInterface')) {

Resources/config/enqueue.xml

@@ -0,0 +1,15 @@
+<?xml version="1.0" encoding="utf-8" ?>
+<container xmlns="http://symfony.com/schema/dic/services"
+    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+    xsi:schemaLocation="http://symfony.com/schema/dic/services http://symfony.com/schema/dic/services/services-1.0.xsd">
+
+    <services>
+        <service id="liip_imagine.async.resolve_all_cache_processor" class="Liip\ImagineBundle\Async\ResolveAllCacheProcessor">
+            <argument type="service" id="liip_imagine.cache.manager" />
+            <argument type="service" id="liip_imagine.filter.manager" />
+            <argument type="service" id="liip_imagine.data.manager" />
+
+            <tag name="enqueue.client.processor" />
+        </service>
+    </services>
+</container>