LiipImagineBundle/Async/ResolveCacheProcessor.php at 713e36f5df353d7c5345daed5a2eefc23c103849 · liip/LiipImagineBundle · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
<?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);
    }
}