Fetch messages from the queue within the store scope.
@param string $queue Name of the AMQP queue
@param Mage_Core_Model_Store $store
@return self
969146
protected function _consumeQueue($queue, Mage_Core_Model_Store $store)
    {
        $this->_api = $this->_helper->getSdkAmqp($queue, $store);
        $this->_api->openConnection();
        $this->_api->getChannel()->basic_consume($this->_helper->_processQueueName($queue, $this->_coreHelper->getConfigModel($store)), '', false, false, false, false, array($this, 'process'));

        $timeout = $this->_helper->getConfigModel($store)->queueTimeout ? $this->_helper->getConfigModel($store)->queueTimeout : 5;
        while (count($this->_api->getChannel()->callbacks)) {
                try{
                    $this->_api->getChannel()->wait(null, false , $timeout);
                }catch(\PhpAmqpLib\Exception\AMQPTimeoutException $e){
                    $this->_api->getChannel()->close();
		    $this->_api->closeConnection();
                    break;
                }
        }

        return $this;
    }
786736
public static function get_Hstore( &$store )
    {
        if( $store instanceof Hstore ) {
            return $store;
        }
        return !is_null($store)
            ? new Hstore($store)
            : null
            ;
    }
100124
public function delete(QueueMessage $message)
    {
        $queue = $message->getQueueName() ?: $this->queue;

        $query = $this->db->getQuery(true);

        $query->delete($query->quoteName($this->table))
            ->where('id = :id')
            ->where('queue = :queue')
            ->bind('id', $message->getId())
            ->bind('queue', $queue);

        $this->db->setQuery($query)->execute();

        return $this;
    }
1352318
public function getStoreLevelAmqpConfigurations(Mage_Core_Model_Store $store)
    {
        $coreConfig = $this->_coreHelper->getConfigModel($store);
        $amqpConfig = $this->_helper->getConfigModel($store);
        return array(
            'store_id' => $coreConfig->storeId,
            'username' => $amqpConfig->username,
            'password' => $amqpConfig->password,
        );
    }
458952
protected function storeSnapshotForQueue($queue)
    {
        $data = $this->baseSnapshotData($key = 'queue:'.$queue);

        $this->connection()->zadd(
            'snapshot:'.$key, $time = Chronos::now()->getTimestamp(), json_encode([
                'throughput' => $data['throughput'],
                'runtime' => $data['runtime'],
                'wait' => app(WaitTimeCalculator::class)->calculateFor($queue),
                'time' => $time,
            ])
        );

        $this->connection()->zremrangebyrank(
            'snapshot:'.$key, 0, -25
        );
    }
763298
public function pop($queue = null)
	{
		$queue = $this->getQueue($queue);

		$response = $this->sqs->receiveMessage(
			['QueueUrl' => $queue, 'AttributeNames' => ['ApproximateReceiveCount']]
		);

		if (count($response['Messages']) > 0) {
			if ($this->jobCreator) {
				return call_user_func($this->jobCreator, $this->container, $this->sqs, $queue, $response);
			} else {
				return new SqsJob($this->container, $this->sqs, $queue, $response['Messages'][0]);
			}
		}
	}
1239456
protected function getNextMessage($queue)
    {
        $queue = (array) $queue;

        foreach ($queue as $queueName) {
            if ($message = $this->manager->pop($queueName)) {
                return $message;
            }
        }

        return null;
    }
1246423
public function filterByQueue($queue = self::QUEUE_NO)
    {
        if (!in_array($queue, $this->schema('queue'))) {
            throw new ApiException(sprintf('Invalid queue status for %s.', __METHOD__));
        }

        $realms = $this->getRealms();

        return $this->filter($realms, 'queue', $queue);
    }
9713
public function delete(QueueMessage $message)
    {
        $this->client->deleteMessage(
            [
                'QueueUrl' => $this->getQueueUrl($message->getQueueName()),
                'ReceiptHandle' => $this->getReceiptHandle($message),
            ]
        );

        return $this;
    }
571402
public function delete(QueueMessage $message)
    {
        $queue = $message->getQueueName() ?: $this->queue;

        $sql = 'DELETE FROM ' . $this->table .
            ' WHERE id = :id AND queue = :queue';

        $stat = $this->pdo->prepare($sql);
        $stat->bindValue(':id', $message->getId());
        $stat->bindValue(':queue', $queue);

        $stat->execute();

        return $this;
    }
398250
public static function pop($queue)
    {
        $item = self::redis()->lpop('queue:' . $queue);
        if (!$item) {
            return;
        }

        return json_decode($item, true);
    }
777623
public function get($queue = "", $noAck = false)
    {
        if ($this->getDeferred !== null) {
            throw new ChannelException("Another 'basic.get' already in progress. You should use 'basic.consume' instead of multiple 'basic.get'.");
        }

        $response = $this->getImpl($queue, $noAck);

        if ($response instanceof PromiseInterface) {
            $this->getDeferred = new Deferred();

            $response->done(function ($frame) {
                if ($frame instanceof MethodBasicGetEmptyFrame) {
                    // deferred has to be first nullified and then resolved, otherwise results in race condition
                    $deferred = $this->getDeferred;
                    $this->getDeferred = null;
                    $deferred->resolve(null);

                } elseif ($frame instanceof MethodBasicGetOkFrame) {
                    $this->getOkFrame = $frame;
                    $this->state = ChannelStateEnum::AWAITING_HEADER;

                } else {
                    throw new \LogicException("This statement should never be reached.");
                }
            });

            return $this->getDeferred->promise();

        } elseif ($response instanceof MethodBasicGetEmptyFrame) {
            return null;

        } elseif ($response instanceof MethodBasicGetOkFrame) {
            $this->state = ChannelStateEnum::AWAITING_HEADER;

            $headerFrame = $this->getClient()->awaitContentHeader($this->getChannelId());
            $this->headerFrame = $headerFrame;
            $this->bodySizeRemaining = $headerFrame->bodySize;
            $this->state = ChannelStateEnum::AWAITING_BODY;

            while ($this->bodySizeRemaining > 0) {
                $bodyFrame = $this->getClient()->awaitContentBody($this->getChannelId());

                $this->bodyBuffer->append($bodyFrame->payload);
                $this->bodySizeRemaining -= $bodyFrame->payloadSize;

                if ($this->bodySizeRemaining < 0) {
                    $this->state = ChannelStateEnum::ERROR;
                    $this->client->disconnect(Constants::STATUS_SYNTAX_ERROR, $errorMessage = "Body overflow, received " . (-$this->bodySizeRemaining) . " more bytes.");
                    throw new ChannelException($errorMessage);
                }
            }

            $this->state = ChannelStateEnum::READY;

            $message = new Message(
                null,
                $response->deliveryTag,
                $response->redelivered,
                $response->exchange,
                $response->routingKey,
                $this->headerFrame->toArray(),
                $this->bodyBuffer->consume($this->bodyBuffer->getLength())
            );

            $this->headerFrame = null;

            return $message;

        } else {
            throw new \LogicException("This statement should never be reached.");
        }
    }
297227
private function getConfiguration($queue = null)
    {
        if (null === $queue) {
            foreach ($this->queues as $queue) {
                $configuration = $this->getConfigurationByQueue($queue);

                if (null !== $configuration) {
                    return $configuration;
                }
            }

            return null;
        }

        return $this->getConfigurationByQueue($queue);
    }
815495
public function getPrefetchShipping($store = null)
    {
        return $this->getScopeConfig()->isSetFlag(
            self::XML_PATH_PREFETCH_SHIPPING,
            ScopeInterface::SCOPE_STORE,
            $store
        );
    }
1280128
public function isActive($store = null)
    {
        return $this->getScopeConfig()->isSetFlag(
            self::XML_PATH_ACTIVE,
            ScopeInterface::SCOPE_STORE,
            $store
        );
    }
559010
public function subscribe($queue)
    {
        if ($this->currentEnvelope) {
            throw new \RuntimeException('Don\'t be greedy and process the message you already have!');
        }

        $this->queue = $this->manager->getQueue((string) $queue);
    }
1037368
public function getJsonMessage($queue, $auto_ack = false)
    {
        if (!isset($this->_queues[$queue])) {
            throw new InvalidKeyException(['retrieve message from queue failed: `:queue queue is NOT exists`', 'queue' => $queue]);
        }

        try {
            $envelope = $this->_queues[$queue]->get($auto_ack ? AMQP_AUTOACK : AMQP_NOPARAM);
        } catch (\Exception $e) {
            throw new AmqpException(['retrieve message from `:queue` queue failed: :error', 'queue' => $queue, 'error' => $e->getMessage()]);
        }

        if ($envelope !== false) {
            $json = json_decode($envelope->getBody(), true);
            if ($json === null) {
                throw new AmqpException(['json_decode `:1` queue `:2` message failed: :3', $queue, $envelope->getBody(), json_last_error_msg()]);
            }

            $json[self::MESSAGE_METADATA] = ['queue' => $queue, 'delivery_tag' => $envelope->getDeliveryTag(), 'is_redelivery' => $envelope->isRedelivery()];

            return $json;
        } else {
            return false;
        }
    }