@param MessageInterface $message
@param string $processorName
@param string $queueName

@return Recipient
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);
    }
1373152
public function getReplyReader($thread, $queueName, $queueConfigId)
    {
        if (!isset($this->replyReaders[$thread])) {
            $configuration = [
                RouterConfig::REPLY_READER_AUTO_ACK         => false,
                RouterConfig::COMMON_QUEUE_NAME             => $queueName,
                RouterConfig::COMMON_QUEUE_CONFIGURATION_ID => $queueConfigId,
            ];
            
            $pool = new RouterReplyReader($this, $thread, $configuration);
            $pool->setup();
            
            $this->replyReaders[$thread] = $pool;
        }
        
        return $this->replyReaders[$thread];
    }
79246
public function addMessageToQueue(Message $message, $name)
    {
        $q      = $this->getQueue($name);
        $qid    = $this->getQueueId($q->getName());
        $body   = base64_encode(serialize($message));
        $md5    = md5($body);

        $sql = 'INSERT INTO ' . $this->messageTable . '
            (queue_id, body, created, timeout, md5)
            VALUES
            (:queue_id, :body, :created, :timeout, :md5)
            ';
        $stmt = $this->getDb()->prepare($sql);
        $stmt->bindParam(':queue_id', $qid, \PDO::PARAM_INT);
        $stmt->bindParam(':body', $body, \PDO::PARAM_STR);
        $stmt->bindParam(':md5', $md5, \PDO::PARAM_STR);
        $stmt->bindValue(':created', time(), \PDO::PARAM_INT);
        $stmt->bindValue(':timeout', 30, \PDO::PARAM_INT);
        $stmt->execute();
        return true;
    }
432271
public function send(MessageInterface $message, $skipErrors = true) {

        $mobitex = $this->getClient();
        
        foreach ($message->getRecipient() as $recipient) {
            try {
                $mobitex->sendMessage($recipient, $message->getText());
            }
            catch (Exception $e) {
                $this->addError(new SendingError($recipient, $e->getCode(), $e->getMessage()));
                if (!$skipErrors) {
                    throw new \RuntimeException($e->getMessage(), $e->getCode());
                }
            }
        }
        return $this->getErrors()->count() === 0;
    }
1267632
protected function addApplicationHeaders(AMQPMessage $message, QueueSendMessageTransfer $queueSendMessageTransfer): QueueSendMessageTransfer
    {
        $messageProperties = $message->get_properties();

        if (isset($messageProperties['application_headers'])) {
            $headers = $messageProperties['application_headers'];
            $queueSendMessageTransfer->setHeaders($headers->getNativeData());
        }

        return $queueSendMessageTransfer;
    }
895003
public function consumeOne(MessageInterface $message)
    {
        $consumeEvent = new ConsumeEvent($message);

        try {
            $this->dispatcher->dispatch(BarbeQEvents::PRE_CONSUME, $consumeEvent);

            $message->start();

            $this->messageDispatcher->dispatch($message->getQueue(), $consumeEvent);

            $message->complete();
            $this->adapter->onSuccess($message);

            $this->dispatcher->dispatch(BarbeQEvents::POST_CONSUME, $consumeEvent);
        } catch(\Exception $e) {
            $message->completeWithError();
            $this->adapter->onError($message);

            $this->dispatcher->dispatch(BarbeQEvents::POST_CONSUME, $consumeEvent);

            // TODO
            throw new ConsumerIndigestionException("Error while consuming a message", 0, $e);
        }
    }
1119686
public function getRequestReader ($queueName) 
    {
        $forwardQueueName = $this->_prepareQueueName($queueName);
        
        $server = $this->_getServerOrFail($forwardQueueName, RouterConfig::SERVER_TAG_DEFAULT);
    
        return $server->getReplyReader(self::THREAD_MAIN, $this->replyQueueName, RouterConfig::CONFIG_REPLY_CONSUMER);
    }
1314946
public function push(QueueMessage $message)
    {
        $queue = $message->getQueueName() ?: $this->queue;

        return $this->client->useTube($queue)->put(
            json_encode($message),
            PheanstalkInterface::DEFAULT_PRIORITY,
            $message->getDelay(),
            $this->timeout
        );
    }
428699
public function buildNotifications(MessageInterface $message)
    {
        $notifications  = new \ArrayIterator();
        $recipientQueue = new \SplQueue();
        $recipientChunk = new \ArrayIterator();

        foreach ($message->getRecipientDeviceCollection() as $recipient) {
            $recipientChunk->append($recipient);

            if ($recipientChunk->count() >= $message->getMaxRecipientsPerMessage()) {
                $recipientQueue->enqueue($recipientChunk);
                $recipientChunk = new \ArrayIterator();
            }
        }

        if ($recipientChunk->count()) {
            $recipientQueue->enqueue($recipientChunk);
        }

        while (!$recipientQueue->isEmpty()) {
            $notification = $this->createNotification($recipientQueue->dequeue(), $message);
            $notifications->append($notification);
        }

        return $notifications;
    }
1046658
public function encodeMessage($message, $queue)
    {
        $messageFactory = $this->getMessageFactory();

        if (!$this->messageFactory->accepts($message)) {
            throw new Exception\RuntimeException(
                sprintf(
                    '%s does not accept message of type "%s"',
                    get_class($messageFactory),
                    (is_object($message) ? get_class($message) : gettype($message))
                )
            );
        }

        return new BernardDefaultMessage(
            $queue,
            array(
                self::MESSAGE_CLASS_KEY => get_class($message),
                self::MESSAGE_KEY => $this->getMessageFactory()->toArray($message),
            )
        );
    }
1076085
public function get_eventobject_for_processor($processorname) {
        // This is done for Backwards compatibility. We should consider throwing notices here in future versions and requesting
        // them to use proper api.

        $eventdata = new \stdClass();
        foreach ($this->properties as $prop) {
            $func = "get_$prop";
            $eventdata->$prop = method_exists($this, $func) ? $this->$func($processorname) : $this->$prop;
        }
        return $eventdata;
    }
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);
    }
580083
public function addMessageToQueue($message, $queue)
    {
        $this->queue->sendMessage([
            'QueueUrl' => $this->getQueueURL($queue),
            'MessageBody' => $message,
        ]);
    }
368805
public function send(Message $message, string $connectionName = null): MessageEnvelope
    {
        return $this->getNotifierForMessage($message)->send($message, $connectionName);
    }
105187
protected function addToQueue(Message $message, string $connectionName = null): MessageEnvelope
    {
        $envelope = $this->createMessageEnvelope($message);
        if ($connectionName === null) {
            $connectionName = $this->connectionHandler->getDefaultConnection()->getType();
        }
        if (!array_key_exists($connectionName, $this->queues)) {
            $this->queues[$connectionName] = new MessageEnvelopeQueueImpl();
        }
        $this->queues[$connectionName]->add($envelope);

        return $envelope;
    }
259877
public function collectQueueMessages(AMQPMessage $message)
    {
        $queueSendMessageTransfer = new QueueSendMessageTransfer();
        $queueSendMessageTransfer->setBody($message->getBody());
        $queueSendMessageTransfer = $this->addApplicationHeaders($message, $queueSendMessageTransfer);

        $queueReceiveMessageTransfer = new QueueReceiveMessageTransfer();
        $queueReceiveMessageTransfer->setQueueMessage($queueSendMessageTransfer);

        $queueName = $message->delivery_info['exchange'];
        if (!$queueName) {
            $queueName = $message->delivery_info['routing_key'];
        }

        $queueReceiveMessageTransfer->setQueueName($queueName);
        $queueReceiveMessageTransfer->setDeliveryTag($message->delivery_info['delivery_tag']);

        $this->collectedMessages[] = $queueReceiveMessageTransfer;
    }
1033142
public function push(QueueMessage $message)
    {
        $queue = $message->getQueueName() ?: $this->queue;

        $this->queueDeclare($queue);

        $options = [
            'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT,
        ];

        if ($message->getDelay() > 0) {
            $options['application_headers'] = new AMQPTable(
                [
                    'x-delay' => $message->getDelay(),
                ]
            );
        }

        $msg = new AMQPMessage(json_encode($message), $options);

        $this->channel->basic_publish($msg, '', $queue);

        return 1;
    }