FM / components

/var/www/vhosts/nabawater/vendor/mikemadisonweb/yii2-rabbitmq/components/

/var/www/vhosts/nabawater/vendor/mikemadisonweb/yii2-rabbitmq/components

Name	Size	Mode	Actions
AbstractConnectionFactory.php	3163	0644	edit dl rm
BaseConsumer.php	3303	0644	edit dl rm
BaseRabbitMQ.php	7118	0644	edit dl rm
Consumer.php	9754	0644	edit dl rm
ConsumerInterface.php	644	0644	edit dl rm
MultipleConsumer.php	2608	0644	edit dl rm
Producer.php	2830	0644	edit dl rm
ProducerInterface.php	323	0644	edit dl rm
RabbitMQConsumerEvent.php	368	0644	edit dl rm
RabbitMQEvent.php	360	0644	edit dl rm
RabbitMQPublisherEvent.php	369	0644	edit dl rm

Edit: /var/www/vhosts/nabawater/vendor/mikemadisonweb/yii2-rabbitmq/components/MultipleConsumer.php (2608B)

<?php

namespace mikemadisonweb\rabbitmq\components;

use PhpAmqpLib\Message\AMQPMessage;

class MultipleConsumer extends Consumer
{
    protected $queues = [];

/**
     * @param $queue
     * @return string
     */
    public function getQueueConsumerTag($queue)
    {
        return sprintf('%s-%s', $this->getConsumerTag(), $queue);
    }

/**
     * @param array $queues
     */
    public function setQueues(array $queues)
    {
        $this->queues = $queues;
    }

protected function startConsuming()
    {
        if ($this->autoSetupFabric) {
            $this->setupFabric();
        }
        foreach ($this->queues as $name => $options) {
            //PHP 5.3 Compliant
            $currentObject = $this;
            $this->getChannel()->basic_consume($name, $this->getQueueConsumerTag($name), false, false, false, false, function (AMQPMessage $msg) use ($currentObject, $name) {
                $currentObject->processQueueMessage($name, $msg);
            });
        }
    }

protected function queueDeclare()
    {
        foreach ($this->queues as $name => $options) {
            $options = array_merge($this->queueOptions, $options);
            list($queueName, ,) = $this->getChannel()->queue_declare(
                $name,
                $options['passive'],
                $options['durable'],
                $options['exclusive'],
                $options['auto_delete'],
                $options['nowait'],
                $options['arguments'],
                $options['ticket']
            );
            if (isset($options['routing_keys']) && count($options['routing_keys']) > 0) {
                foreach ($options['routing_keys'] as $routingKey) {
                    $this->queueBind($queueName, $this->exchangeOptions['name'], $routingKey);
                }
            } else {
                $this->queueBind($queueName, $this->exchangeOptions['name'], $this->routingKey);
            }
        }
        $this->queueDeclared = true;
    }

/**
     * @param $queueName
     * @param AMQPMessage $msg
     * @throws \Exception
     */
    public function processQueueMessage($queueName, AMQPMessage $msg)
    {
        if (!isset($this->queues[$queueName])) {
            throw new \Exception('Queue not found!');
        }
        $this->processMessageQueueCallback($msg, $queueName, $this->queues[$queueName]['callback']);
    }

public function stopConsuming()
    {
        foreach ($this->queues as $name => $options) {
            $this->getChannel()->basic_cancel($this->getQueueConsumerTag($name), false, true);
        }
    }
}

getConsumerTag(), $queue); } /** * @param array $queues */ public function setQueues(array $queues) { $this->queues = $queues; } protected function startConsuming() { if ($this->autoSetupFabric) { $this->setupFabric(); } foreach ($this->queues as $name => $options) { //PHP 5.3 Compliant $currentObject = $this; $this->getChannel()->basic_consume($name, $this->getQueueConsumerTag($name), false, false, false, false, function (AMQPMessage $msg) use ($currentObject, $name) { $currentObject->processQueueMessage($name, $msg); }); } } protected function queueDeclare() { foreach ($this->queues as $name => $options) { $options = array_merge($this->queueOptions, $options); list($queueName, ,) = $this->getChannel()->queue_declare( $name, $options['passive'], $options['durable'], $options['exclusive'], $options['auto_delete'], $options['nowait'], $options['arguments'], $options['ticket'] ); if (isset($options['routing_keys']) && count($options['routing_keys']) > 0) { foreach ($options['routing_keys'] as $routingKey) { $this->queueBind($queueName, $this->exchangeOptions['name'], $routingKey); } } else { $this->queueBind($queueName, $this->exchangeOptions['name'], $this->routingKey); } } $this->queueDeclared = true; } /** * @param $queueName * @param AMQPMessage $msg * @throws \Exception */ public function processQueueMessage($queueName, AMQPMessage $msg) { if (!isset($this->queues[$queueName])) { throw new \Exception('Queue not found!'); } $this->processMessageQueueCallback($msg, $queueName, $this->queues[$queueName]['callback']); } public function stopConsuming() { foreach ($this->queues as $name => $options) { $this->getChannel()->basic_cancel($this->getQueueConsumerTag($name), false, true); } } }