rabbit-queue-bundle/Command/UpdateDefinitionCommand.php at 758cf36b16c07534615abdc8d79262f3d16ea23e · wakeapp/rabbit-queue-bundle · 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
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
<?php

declare(strict_types=1);

namespace Wakeapp\Bundle\RabbitQueueBundle\Command;

use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Exchange\AMQPExchangeType;
use PhpAmqpLib\Wire\AMQPTable;
use Symfony\Component\Console\Command\Command;
use Symfony\Component\Console\Input\InputInterface;
use Symfony\Component\Console\Output\OutputInterface;
use Wakeapp\Bundle\RabbitQueueBundle\Definition\DefinitionInterface;
use Wakeapp\Bundle\RabbitQueueBundle\Enum\ExchangeEnum;
use Wakeapp\Bundle\RabbitQueueBundle\Enum\QueueTypeEnum;
use Wakeapp\Bundle\RabbitQueueBundle\Exception\RouteStructureException;

class UpdateDefinitionCommand extends Command
{
    protected static $defaultName = 'rabbit:definition:update';

    private AMQPStreamConnection $connection;

    /**
     * @var DefinitionInterface[]
     */
    private iterable $definitionList;

    /**
     * @required
     */
    public function dependencyInjection(
        AMQPStreamConnection $connection
    ): void {
        $this->connection = $connection;
    }

    /**
     * @param DefinitionInterface[]|iterable $definitionList
     */
    public function setDefinitionList(iterable $definitionList): void
    {
        $this->definitionList = $definitionList;
    }

    /**
     * {@inheritdoc}
     */
    protected function configure(): void
    {
        $this
            ->setDescription('Run migration')
            ->setHelp('This command allows you to update schema of queues')
        ;
    }

    /**
     * {@inheritDoc}
     *
     * @throws RouteStructureException
     */
    protected function execute(InputInterface $input, OutputInterface $output): int
    {
        $routersToInit = [];
        $initializedRouters = [];

        foreach ($this->definitionList as $definition) {
            if ($definition->getQueueType() & QueueTypeEnum::ROUTER) {
                if (method_exists($definition, 'dependsOn') && !empty($definition->dependsOn())) {
                    $routersToInit[$definition::getQueueName()] = $definition;
                } else {
                    $definition->init($this->connection);
                    $initializedRouters[] = $definition::getQueueName();
                }
            }
        }

        $successLoop = true;
        while ($successLoop && !empty($routersToInit)) {
            $successLoop = false;

            foreach ($routersToInit as $router) {
                if (empty(array_diff($router->dependsOn(), $initializedRouters))) {
                    $successLoop = true;
                    $router->init($this->connection);
                    unset($routersToInit[$router::getQueueName()]);
                    $initializedRouters[] = $router::getQueueName();
                }
            }
        }

        if (!$successLoop) {
            throw new RouteStructureException('Router definitions have cyclic dependencies');
        }


        foreach ($this->definitionList as $definition) {
            if ($definition->getQueueType() & QueueTypeEnum::ROUTER) {
                continue;
            }

            $definition->init($this->connection);

            $this->bindRetryExchange($definition);
        }

        return 0;
    }

    private function bindRetryExchange(DefinitionInterface $definition): void
    {
        $queueName = $definition::getQueueName();
        $channel = $this->connection->channel();

        $channel->exchange_declare(
            ExchangeEnum::RETRY_EXCHANGE,
            'x-delayed-message',
            false,
            true,
            false,
            false,
            false,
            new AMQPTable(['x-delayed-type' => AMQPExchangeType::DIRECT])
        );

        $channel->queue_bind($queueName, ExchangeEnum::RETRY_EXCHANGE, $queueName);
    }
}