Do the actual work for processing a single message.

@param $queueName The queue to which the message belongs
@param $envelope The envelope containing the message to process
@return An array if [$messageSucceeded, $willRetry]
433376
protected function doOnce(string $queueName, Envelope $envelope) : array
    {
        $result = false;
        $message = $envelope->unwrap();

        $this->getLogger()->debug('Handling message {msg}', ['msg' => self::nameOf($message)]);
        try {
            $result = $this->handleMessage($message);
        } catch (Exception\MustStop $e) {
            $this->driver->ack($queueName, $envelope);
            throw $e;
        } catch (Exception\ShouldReleaseMessage $e) {
            $this->driver->release($queueName, $envelope);
            $this->getLogger()->debug('Releasing message {msg} due to {cls} exception: {err}', [
                'msg' => self::nameOf($message),
                'cls' => get_class($e),
                'err' => $e->getMessage(),
            ]);
            return [$result, true];
        } catch (\Exception $e) {
            // any other exception is simply logged. We and marked as failed
            // below. We only log here because we can't make guarantees about
            // the implementation of the handler and whether or not it actually
            // throws exceptions on failure (see PcntlForkingHandler).
            $this->getLogger()->critical('Unexpected {cls} exception handling {name} message: {msg}', [
                'cls'   => get_class($e),
                'name'  => self::nameOf($message),
                'msg'   => $e->getMessage()
            ]);
            $result = false;
        }
        $this->getLogger()->debug('Handled message {msg}', ['msg' => self::nameOf($message)]);

        if ($result) {
            $willRetry = false;
            $this->driver->ack($queueName, $envelope);
            $this->getLogger()->debug('Acknowledged message {msg}', ['msg' => self::nameOf($message)]);
        } else {
            $willRetry = $this->failed($queueName, $envelope);
            $this->getLogger()->debug('Failed message {msg}', ['msg' => self::nameOf($message)]);
        }

        return [$result, $willRetry];
    }
1233366
private function handleException(Envelope $envelope, Request $request, Message $message, \Throwable $exception): Envelope
    {
        if ($exception instanceof HasEndUserMessage) {
            $code = $exception->getCode();
            $httpCode = StatusCodeConverter::vendorToHttp($code);
            $errorName = ClassUtils::getShortName($exception);
            $errorMessage = $exception->getEndUserMessage();
            $request->attributes->set('pbjx_redact_error_message', false);
        } elseif ($exception instanceof HttpExceptionInterface) {
            $code = StatusCodeConverter::httpToVendor($exception->getStatusCode());
            $httpCode = $exception->getStatusCode();
            $errorName = ClassUtils::getShortName($exception);
            $errorMessage = $exception->getMessage();
        } elseif ($exception instanceof RequestHandlingFailed) {
            $response = $exception->getResponse();
            $code = $response->get('error_code', Code::UNKNOWN);
            $httpCode = StatusCodeConverter::vendorToHttp($code);
            $errorName = $response->get('error_name', ClassUtils::getShortName($exception));
            $errorMessage = $response->get('error_message', $exception->getMessage());
        } elseif ($exception instanceof GdbotsPbjException) {
            $code = Code::INVALID_ARGUMENT;
            $httpCode = HttpCode::HTTP_UNPROCESSABLE_ENTITY;
            $errorName = ClassUtils::getShortName($exception);
            $errorMessage = $exception->getMessage();
            // these error messages are safe to show as they only indicate schema problems
            $request->attributes->set('pbjx_redact_error_message', false);
        } else {
            $code = $exception->getCode() > 0 ? $exception->getCode() : Code::INVALID_ARGUMENT;
            $httpCode = StatusCodeConverter::vendorToHttp($code);
            $errorName = ClassUtils::getShortName($exception);
            $errorMessage = $exception->getMessage();
        }

        return $envelope
            ->set('code', $code)
            ->set('http_code', HttpCode::create($httpCode))
            ->set('error_name', $errorName)
            ->set('error_message', $errorMessage);
    }
915611
private function getIronMqQueueName(GetResponseEvent $event, array&$message) {
		if (array_key_exists('_qpush_queue', $message)) {
			return $message['_qpush_queue'];
		} else if (null !== ($subscriberUrl = $event->getRequest()->headers->get('iron-subscriber-message-url'))) {
			if (preg_match('#/queues/([a-z0-9_-]+)/messages/#i', $subscriberUrl, $matches)) {
				$queue = $matches[1];
				if (substr($queue, 0, 6) == 'qpush_') {
					$queue = substr($queue, 6);
				}

				return $queue;
			}
		}

		throw new \RuntimeException('Unable to get queue name');
	}
392923
public boolean processMultipleReply(MessageEnvelope envelope) {
		if (this.event == null) {
			return false;
		}
		boolean triggered = this.event.verifyEventCondition(envelope);
		if (triggered) {
			beforeReply(envelope);
			if (this.replies != null) {
				synchronized (replies) {
					for (AutoReply reply : replies) {
						reply.reply(envelope);
					}
				}
			}
			afterReply(envelope);
		}
		return triggered;
	}
291647
@Override
	public void queueEnvelope(Envelope envelope) {
		if (envelope.hasBuffer()) {
			throw new IllegalStateException("Envelope for OutputChannel has Buffer attached.");
		}

		for (AbstractEvent event : envelope.deserializeEvents()) {
			if (event.getClass() == ChannelCloseEvent.class) {
				synchronized (this.closeLock) {
					this.receiverCloseRequested = true;
					this.closeLock.notifyAll();
				}
			} 
			else if (event instanceof AbstractTaskEvent) {
				if (LOG.isDebugEnabled()) {
					LOG.debug("OutputChannel received task event: " + event);
				}
				
				this.outputGate.deliverEvent((AbstractTaskEvent) event);
			}
			else {
				throw new RuntimeException("OutputChannel received an event that is neither close nor task event.");
			}
		}
	}
994258
public function createQueue($queueName)
    {
        $channel = $this->getChannel();
        $channel->exchange_declare($this->exchange, 'direct', false, true, false);
        $channel->queue_declare($queueName, false, true, false, false);
        $channel->queue_bind($queueName, $this->exchange, $queueName);
    }
970462
public function queue(Message $message)
	{
		// Bump the message ID
		$this->lastMessageId++;

		// Put the message in an envelope
		$envelope = new MessageEnvelope($this->lastMessageId, $message);

		// Save the message so we can update it later on
		$this->storeMessageEnvelope($envelope);

		// Queue and return the envelope
		$this->logger->debug('Queuing Apns\Message #' . $this->lastMessageId . ' to device "' . $message->getDeviceToken() . '" on Apns\Gateway with certificate "' . $this->getCertificate()->getDescription() . '"');
		$this->sendQueue->enqueue($envelope);

		return $envelope;
	}
327048
protected function createRecipient(MessageInterface $message, $processorName, $queueName)
    {
        $newMessage = clone $message;
        $properties = $message->getProperties();
        $properties[Config::PARAMETER_PROCESSOR_NAME] = $processorName;
        $properties[Config::PARAMETER_QUEUE_NAME] = $queueName;
        $newMessage->setProperties($properties);

        $queue = $this->driver->createQueue($queueName);

        return new Recipient($queue, $newMessage);
    }
380393
@Override
	public void queueEnvelope(Envelope envelope) {
		// The sequence number of the envelope to be queued
		final int sequenceNumber = envelope.getSequenceNumber();

		synchronized (this.queuedEnvelopes) {

			if (this.destroyCalled) {
				final Buffer buffer = envelope.getBuffer();
				if (buffer != null) {
					buffer.recycleBuffer();
				}
				return;
			}

			final int expectedSequenceNumber = this.lastReceivedEnvelope + 1;
			if (sequenceNumber != expectedSequenceNumber) {
				// This is a problem, now we are actually missing some data
				reportIOException(new IOException("Expected data packet " + expectedSequenceNumber + " but received " + sequenceNumber));

				// notify that something (an exception) is available
				notifyGateThatInputIsAvailable();

				if (LOG.isDebugEnabled()) {
					LOG.debug("Input channel " + this.toString() + " expected envelope " + expectedSequenceNumber
							+ " but received " + sequenceNumber);
				}

				// rescue the buffer
				final Buffer buffer = envelope.getBuffer();
				if (buffer != null) {
					buffer.recycleBuffer();
				}
			} else {

				this.queuedEnvelopes.add(envelope);
				this.lastReceivedEnvelope = sequenceNumber;
				this.lastSourceID = envelope.getSource();

				// Notify the channel about the new data. notify as much as there is (buffer plus once per event)
				if (envelope.getBuffer() != null) {
					notifyGateThatInputIsAvailable();
				}

				List<? extends AbstractEvent> events = envelope.deserializeEvents();

				if (events != null) {
					for (int i = 0; i < events.size(); i++) {
						notifyGateThatInputIsAvailable();
					}
				}
			}
		}
	}
1175202
public function bindQueue($queue, $exchange, $binding_key = '')
    {
        if (!isset($this->_queues[$queue])) {
            throw new InvalidKeyException(['bind `:queue` queue to `:exchange` exchange with `:binding_key` binding key failed: queue is NOT exists',
                'queue' => $queue,
                'exchange' => $exchange,
                'binding_key' => $binding_key]);
        }

        if (!isset($this->_exchanges[$exchange])) {
            throw new AmqpException(['bind `:queue` queue to `:exchange` exchange with `:binding_key` binding key failed: exchange is NOT exists',
                'queue' => $queue,
                'exchange' => $exchange,
                'binding_key' => $binding_key]);
        }

        try {
            $this->_queues[$queue]->bind($exchange, $binding_key);
        } catch (\Exception $e) {
            throw new AmqpException(['bind `:queue` queue to `:exchange` exchange with `:binding_key` binding key failed: :error',
                'queue' => $queue,
                'exchange' => $exchange,
                'binding_key' => $binding_key,
                'error' => $e->getMessage()]);
        }

        return $this;
    }
364367
public function receiveMessages($queueName, $chunkSize = 100, array $options = [])
    {
        return $this->consumer->receiveMessages($queueName, $chunkSize, $options);
    }
487302
protected String safeGetQuickReplyPayload(MessageEnvelope envelope) {
		if (envelope != null && envelope.getMessage() != null
				&& envelope.getMessage().getQuickReply() != null
				&& envelope.getMessage().getQuickReply().getPayload() != null) {
			return envelope.getMessage().getQuickReply().getPayload();
		}

		return "";
	}
568182
public function send(Message $message)
	{
		// Queue the message and flush the associated gateway
		$messageEnvelope = $this->queue($message);
		$this->flush( $message->getCertificate() );

		// Return the envelope
		return $messageEnvelope;
	}
273206
public function execute(QueueInterface $queue, Message $message): bool
    {
        $service = $this->objectManager->get($this->className);
        $this->deferMethodCallAspect->setProcessingJob(true);
        try {
            $methodName = $this->methodName;
            call_user_func_array([$service, $methodName], $this->arguments);
            return true;
        } catch (\Exception $exception) {
            throw $exception;
        } finally {
            $this->deferMethodCallAspect->setProcessingJob(false);
        }
    }
675627
public function handle(Envelope $envelope, StackInterface $stack): Envelope
    {
        $message = $envelope->getMessage();
        $msg = $message->getMessage();

        if (!$msg instanceof AbstractPushMessage) {
            return $stack->next()->handle($envelope, $stack);
        }

        if (0 === \count($envelope->all(ReceivedStamp::class))) {
            return $stack->next()->handle($envelope, $stack);
        }

        $rawMessage = $msg->getRawMessage();

        foreach ($rawMessage->getRecipients() as $recipient) {
            foreach ($this->repository->findUserEnableDevices($recipient->getUser()) as $device) {
                $this->sender->sendTo($device, $rawMessage);
            }
        }

        return $stack->next()->handle($envelope, $stack);
    }
842571
public function deleteQueue($queueName)
    {
        $request  = new DeleteQueueRequest($queueName);
        $response = new DeleteQueueResponse();

        return $this->client->sendRequest($request, $response);
    }
873573
public function consumeMessages($queue, $callback, $flags = AMQP_NOPARAM)
    {
        if (!isset($this->_queues[$queue])) {
            throw new InvalidKeyException(['consume message from queue failed: `:queue queue is NOT exists`', 'queue' => $queue]);
        }

        try {
            $this->_queues[$queue]->consume(function (\AMQPEnvelope $envelope) use ($callback, $queue) {
                return $callback(new Message($this, $queue, $envelope));
            }, $flags);
        } catch (\Exception $e) {
            throw new AmqpException('consume `:queue` queue message failed: ', $e->getMessage());
        }
    }