Resolve image caches in background. Using message queue.

.travis.yml

@@ -30,14 +30,16 @@ matrix:
       env: SYMFONY_VERSION=2.3.x-dev
     - php: 5.6
       env: SYMFONY_VERSION=2.7.x-dev
+    - php: 5.6
+      env: SYMFONY_VERSION=2.8.x-dev WITH_ENQUEUE=true
     - php: 7.1
       env: SYMFONY_VERSION=3.0.x-dev
     - php: 7.1
       env: SYMFONY_VERSION=3.1.x-dev
     - php: 7.1
       env: SYMFONY_VERSION=3.2.x-dev
     - php: 7.1
-      env: SYMFONY_VERSION=3.3.x-dev
+      env: SYMFONY_VERSION=3.3.x-dev WITH_ENQUEUE=true
     - php: 7.1
       env: SYMFONY_VERSION=dev-master
   allow_failures:
@@ -55,6 +57,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 [ "$WITH_ENQUEUE" = true ] ; then composer require --no-update --dev enqueue/enqueue-bundle:^0.3.6; fi;
 
 install:
   - travis_retry composer update $COMPOSER_FLAGS

Async/CacheResolved.php

@@ -0,0 +1,71 @@
+<?php
+namespace Liip\ImagineBundle\Async;
+
+use Enqueue\Util\JSON;
+
+class CacheResolved implements \JsonSerializable
+{
+    /**
+     * @var string
+     */
+    private $path;
+
+    /**
+     * @var \string[]
+     */
+    private $uris;
+
+    /**
+     * @param string $path
+     * @param string[]|null $uris
+     */
+    public function __construct($path, array $uris)
+    {
+        $this->path = $path;
+        $this->uris = $uris;
+    }
+
+    /**
+     * @return string
+     */
+    public function getPath()
+    {
+        return $this->path;
+    }
+
+    /**
+     * @return \string[]
+     */
+    public function getUris()
+    {
+        return $this->uris;
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function jsonSerialize()
+    {
+        return array('path' => $this->path, 'uris' => $this->uris);
+    }
+
+    /**
+     * @param string $json
+     *
+     * @return static
+     */
+    public static function jsonDeserialize($json)
+    {
+        $data = JSON::decode($json);
+
+        if (empty($data['path'])) {
+            throw new \LogicException('The message does not contain "path" but it is required.');
+        }
+
+        if (empty($data['uris'])) {
+            throw new \LogicException('The message uris must not be empty array.');
+        }
+
+        return new static($data['path'], $data['uris']);
+    }
+}

Async/ResolveCache.php

@@ -0,0 +1,86 @@
+<?php
+namespace Liip\ImagineBundle\Async;
+
+use Enqueue\Util\JSON;
+
+class ResolveCache implements \JsonSerializable
+{
+    /**
+     * @var string
+     */
+    private $path;
+
+    /**
+     * @var array|null|\string[]
+     */
+    private $filters;
+
+    /**
+     * @var bool
+     */
+    private $force;
+
+    /**
+     * @param string $path
+     * @param string[]|null $filters
+     * @param bool $force
+     */
+    public function __construct($path, array $filters = null, $force = false)
+    {
+        $this->path = $path;
+        $this->filters = $filters;
+        $this->force = $force;
+    }
+
+    /**
+     * @return string
+     */
+    public function getPath()
+    {
+        return $this->path;
+    }
+
+    /**
+     * @return null|\string[]
+     */
+    public function getFilters()
+    {
+        return $this->filters;
+    }
+
+    /**
+     * @return bool
+     */
+    public function isForce()
+    {
+        return $this->force;
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function jsonSerialize()
+    {
+        return array('path' => $this->path, 'filters' => $this->filters, 'force' => $this->force);
+    }
+
+    /**
+     * @param string $json
+     *
+     * @return static
+     */
+    public static function jsonDeserialize($json)
+    {
+        $data = array_replace(array('path' => null, 'filters' => null, 'force' => false), JSON::decode($json));
+
+        if (false == $data['path']) {
+            throw new \LogicException('The message does not contain "path" but it is required.');
+        }
+
+        if (false == (is_null($data['filters']) || is_array($data['filters']))) {
+            throw new \LogicException('The message filters could be either null or array.');
+        }
+
+        return new static($data['path'], $data['filters'], $data['force']);
+    }
+}

Async/ResolveCacheProcessor.php

@@ -0,0 +1,108 @@
+<?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 Liip\ImagineBundle\Imagine\Cache\CacheManager;
+use Liip\ImagineBundle\Imagine\Data\DataManager;
+use Liip\ImagineBundle\Imagine\Filter\FilterManager;
+
+class ResolveCacheProcessor 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 {
+            $message = ResolveCache::jsonDeserialize($psrMessage->getBody());
+        } catch (\Exception $e) {
+            return Result::reject($e->getMessage());
+        }
+
+        $filters = $message->getFilters() ?: array_keys($this->filterManager->getFilterConfiguration()->all());
+        $path = $message->getPath();
+        $results = array();
+        foreach ($filters as $filter) {
+            if ($this->cacheManager->isStored($path, $filter) && $message->isForce()) {
+                $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::CACHE_RESOLVED, new CacheResolved($path, $results));
+
+        return self::ACK;
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public static function getSubscribedTopics()
+    {
+        return array(
+            Topics::RESOLVE_CACHE => array('queueName' => Topics::RESOLVE_CACHE,  'queueNameHardcoded' => true),
+        );
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public static function getSubscribedQueues()
+    {
+        return array(Topics::RESOLVE_CACHE);
+    }
+}

Async/Topics.php

@@ -0,0 +1,9 @@
+<?php
+namespace Liip\ImagineBundle\Async;
+
+class Topics
+{
+    const RESOLVE_CACHE = 'liip_imagine_resolve_cache';
+
+    const CACHE_RESOLVED = 'liip_imagine_cache_resolved';
+}

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')) {

LiipImagineBundle.php

@@ -11,6 +11,8 @@
 
 namespace Liip\ImagineBundle;
 
+use Enqueue\Bundle\DependencyInjection\Compiler\AddTopicMetaPass;
+use Liip\ImagineBundle\Async\Topics;
 use Liip\ImagineBundle\DependencyInjection\Compiler\FiltersCompilerPass;
 use Liip\ImagineBundle\DependencyInjection\Compiler\LoadersCompilerPass;
 use Liip\ImagineBundle\DependencyInjection\Compiler\LocatorsCompilerPass;
@@ -53,5 +55,12 @@ public function build(ContainerBuilder $container)
         $extension->addLoaderFactory(new StreamLoaderFactory());
         $extension->addLoaderFactory(new FileSystemLoaderFactory());
         $extension->addLoaderFactory(new FlysystemLoaderFactory());
-    }
+
+        if (class_exists('Enqueue\Bundle\DependencyInjection\Compiler\AddTopicMetaPass')) {
+            $container->addCompilerPass(AddTopicMetaPass::create()
+                ->add(Topics::RESOLVE_CACHE, 'Send message to this topic when you want to resolve image\'s cache.')
+                ->add(Topics::CACHE_RESOLVED, 'The topic contains messages about resolved image\'s caches')
+            );
+        }
+     }
 }

Resources/config/enqueue.xml

@@ -0,0 +1,16 @@
+<?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_cache_processor" class="Liip\ImagineBundle\Async\ResolveCacheProcessor">
+            <argument type="service" id="liip_imagine.cache.manager" />
+            <argument type="service" id="liip_imagine.filter.manager" />
+            <argument type="service" id="liip_imagine.data.manager" />
+            <argument type="service" id="enqueue.producer" />
+
+            <tag name="enqueue.client.processor" />
+        </service>
+    </services>
+</container>