invoker = $invoker; $this->configuration = $configuration; $this->massConsumerEnvelopeCallback = $massConsumerEnvelopeCallback; $this->registry = $registry; } /** * @inheritdoc */ public function process($maxNumberOfMessages = null) { $this->registry->register('isSecureArea', true, true); $queue = $this->configuration->getQueue(); $maxIdleTime = $this->configuration->getMaxIdleTime(); $sleep = $this->configuration->getSleep(); if (!isset($maxNumberOfMessages)) { $queue->subscribe($this->getTransactionCallback($queue)); } else { $this->invoker->invoke( $queue, $maxNumberOfMessages, $this->getTransactionCallback($queue), $maxIdleTime, $sleep ); } $this->registry->unregister('isSecureArea'); } /** * Get transaction callback. This handles the case of async. * * @param QueueInterface $queue * @return \Closure */ private function getTransactionCallback(QueueInterface $queue) { $callbackInstance = $this->massConsumerEnvelopeCallback->create( [ 'configuration' => $this->configuration, 'queue' => $queue, ] ); return function (EnvelopeInterface $message) use ($callbackInstance) { $callbackInstance->execute($message); }; } }