For each store with a unique AMQP configuration, consume messages from
each configured queue.
@return self
349439
public function processQueues()
    {
        // consume queues for each store with a unique set of AMQP configuration
        foreach ($this->_amqpConfigHelper->getQueueConfigurationScopes() as $store) {
	    $this->_store = $store;
            $this->_consumeStoreQueues($store);
        }
        return $this;
    }
23010
public function connect(): self
    {
        if ($this->pdoInstance !== null) {
            return $this;
        }

        // Build a database connection if we don't have one connected
        $this->setPdoInstance($this->getAdapter()->connect($this->getAdapterConfig()));

        return $this;
    }
1263718
private function produceQueueInstance($queueName)
    {
        if (isset($this->declaredQueueInstances[$queueName])) {
            return $this->declaredQueueInstances[$queueName];
        }

        if (isset($this->usedQueueInstances[$queueName])) {
            return $this->usedQueueInstances[$queueName];
        }

        $this->ensureChannel();
        $queue = new AMQPQueue($this->channel);
        $queue->setName($queueName);
        $this->usedQueueInstances[$queueName] = $queue;
        return $queue;
    }
259496
protected function createProcessPoolPerQueue()
    {
        return collect(explode(',', $this->options->queue))->map(function ($queue) {
            return $this->createProcessPool($this->options->withQueue($queue));
        });
    }
10951
public function getQueueConfigurationScopes()
    {
        // cache of seen, unique AMQP configurations
        $configurations = array();
        // list of stores to produce unique AMQP configuration
        $uniqueStores = array();
        foreach (Mage::app()->getStores(true) as $store) {
            $amqpConfig = $this->getStoreLevelAmqpConfigurations($store);
            if (!in_array($amqpConfig, $configurations, true)) {
                $configurations[] = $amqpConfig;
                $uniqueStores[] = $store;
            }
        }
        return $uniqueStores;
    }
1368125
protected function initInstances(): self
    {
        if (is_array($this->data) && !empty($this->data)) {
            //Iterates and instantiate records
            $iterator = new RecordIterator($this->data, $this->class, $this->orm);

            foreach ($iterator as $item) {
                if (in_array($item, $this->instances, true)) {
                    //Skip duplicates
                    continue;
                }

                $this->instances[] = $item;
            }
        }

        //Memory free
        $this->data = null;

        return $this;
    }
32170
public function apply()
    {
        if (!isset($_SESSION) || empty($_SESSION)) {
            return $this;
        }


        foreach ($this->messageBlocks as $messageBlock) {
            foreach ($messageBlock->getMessagesByStorage() as $storageType => $messages) {
                $this->addMessagesByStorageType($storageType, $messages);
            }
        }

        foreach ($this->scheduledMessages as $storageType => $messages) {
            $this->addMessagesByStorageType($storageType, $messages);
        }

        $types = array_keys($this->getMessageTypes());
        $updatedStorages = [];

        foreach ($types as $type) {
            $storage = $this->getStorageByMessageType($type);
            if ($storage && $storage->getMessages(false)->getItems()) {
                $updatedStorages[] = $type;
            }
        }


        if ($updatedStorages) {
            Mage::getSingleton('ecomdev_varnish/cookie')->set(
                self::COOKIE_NAME, implode(',', $updatedStorages)
            );
        }

        return $this;
    }
418599
@Bean
    public Queue dmfReceiverQueue() {
        return new Queue(amqpProperties.getReceiverQueue(), true, false, false,
                amqpDeadletterProperties.getDeadLetterExchangeArgs(amqpProperties.getDeadLetterExchange()));
    }
982140
protected function initQueue()
	{
		if ($this->beingDestroyed) {
			throw new \RuntimeException(self::ERROR_BEING_DESTROY);
		}
		$this->idleQueue = new \SplQueue();
		$this->taskQueue = new \SplQueue();
		
		return $this;
	}
444642
private function queueConfigs() {
        $config = $this->getConfig();
        if (!isset($config['schedulers'])) {
            yield $config;
            return;
        }

        foreach ($config['schedulers'] as $child_config) {
            yield mergeConfigOptions($config, $child_config);
        }
    }
196309
public function reset()
    {
        $this->messages = [];

        foreach (self::$when as $when) {
            $this->messages[$when] = [];
        }

        return $this;
    }
360843
protected function registerManager()
    {
        $this->app->bindShared('queue', function($app)
        {
            // Once we have an instance of the queue manager, we will register the various
            // resolvers for the queue connectors. These connectors are responsible for
            // creating the classes that accept queue configs and instantiate queues.
            $manager = new QueueManager($app);

            $this->registerConnectors($manager);

            return $manager;
        });

        $this->app->singleton('queue.connection', function ($app) {
            return $app['queue']->connection();
        });
    }
797800
public function createSqsPolicy()
    {
        $arn = $this->sqs->getQueueArn($this->queueUrl);

        return json_encode([
            'Version'   => '2008-10-17',
            'Id'        =>  sprintf('%s/SQSDefaultPolicy', $arn),
            'Statement' => [
                [
                    'Sid'       => 'SNSPermissions',
                    'Effect'    => 'Allow',
                    'Principal' => ['AWS' => '*'],
                    'Action'    => 'SQS:SendMessage',
                    'Resource'  => $arn
                ]
            ]
        ]);
    }
373401
public function reload()
    {
        $fresh = \Pheasant::instance()->finderFor($this)
            ->find($this->className(), $this->identity()->toCriteria())
            ->one()
            ;

        $this->_data = $fresh->_data;

        return $this;
    }
347117
public function getMessageQueue()
	{
		// For empty queue, if messages exists in the session, enqueue them.
		if (!count($this->_messageQueue))
		{
			$session = JFactory::getSession();
			$sessionQueue = $session->get('application.queue');

			if (count($sessionQueue))
			{
				$this->_messageQueue = $sessionQueue;
				$session->set('application.queue', null);
			}
		}

		return $this->_messageQueue;
	}
1365451
public function createQueue()
    {
        $attributes = [
            'VisibilityTimeout'             => $this->options['message_timeout'],
            'MessageRetentionPeriod'        => $this->options['message_expiration'],
            'ReceiveMessageWaitTimeSeconds' => $this->options['receive_wait_time']
        ];

        if ($this->isQueueFIFO()) {
            $attributes['FifoQueue'] = 'true';
            $attributes['ContentBasedDeduplication'] = $this->options['content_based_deduplication'] === true
                ? 'true'
                : 'false';
        }

        $result = $this->sqs->createQueue(['QueueName' => $this->getNameWithPrefix(), 'Attributes' => $attributes]);

        $this->queueUrl = $result->get('QueueUrl');

        $key = $this->getNameWithPrefix() . '_url';
        $this->cache->save($key, $this->queueUrl);

        $this->log(200, "Created SQS Queue", ['QueueUrl' => $this->queueUrl]);

        if ($this->options['push_notifications']) {

            $policy = $this->createSqsPolicy();

            $this->sqs->setQueueAttributes([
                'QueueUrl'      => $this->queueUrl,
                'Attributes'    => [
                    'Policy'    => $policy,
                ]
            ]);

            $this->log(200, "Created Updated SQS Policy");
        }
    }
1192820
protected function createQueue(AmqpContext $context, $queueName)
    {
        $queue = $context->createQueue($queueName);
        $context->declareQueue($queue);
        $context->purgeQueue($queue);

        return $queue;
    }