Support Dead Lettering

[KinaneD]

Producer

public function __construct(Connection $connection, $exchangeName, $exchangeType, $queueName, $routingKey, $deadLetterExchangeName, $deadLetterExchangeType, $deadLetterQueueName, $deadLetterRoutingKey, $passive = false, $durable = true, $autoDelete = false, $deliveryMode = 2)
{
   $this->connection = $connection;
   $this->exchangeName = $exchangeName;
   $this->exchangeType = $exchangeType;
   $this->queueName = $queueName;
   $this->passive = $passive;
   $this->durable = $durable;
   $this->autoDelete = $autoDelete;
   $this->deliveryMode = $deliveryMode;
   $this->routingKey = $routingKey;           //Since these should match
   $this->bindingKey = $routingKey;           //Since these should match
   $this->deadLetterExchangeName = $deadLetterExchangeName;
   $this->deadLetterExchangeType = $deadLetterExchangeType;
   $this->deadLetterQueueName = $deadLetterQueueName;
   $this->deadLetterRoutingKey = $deadLetterRoutingKey;
}

public function publish($data)
{
// Declare DLX
if(isset($this->deadLetterExchangeName)) {
   $this->connection->getChannel()->exchange_declare($this->deadLetterExchangeName, $this->deadLetterExchangeType);

   $this->connection->getChannel()->queue_declare($this->deadLetterQueueName, $this->passive, $this->durable, false, $this->autoDelete);

   $this->connection->getChannel()->queue_bind($this->deadLetterQueueName, $this->deadLetterExchangeName);
}

   $this->connection->getChannel()->exchange_declare($this->exchangeName, $this->exchangeType, $this->passive, $this->durable, $this->autoDelete);

   $this->connection->getChannel()->queue_declare($this->queueName, $this->passive, $this->durable, false, $this->autoDelete, false, ['x-dead-letter-exchange' => ['S', $this->deadLetterExchangeName], 'x-dead-letter-routing-key' => ['S', $this->deadLetterRoutingKey]]);

   $this->connection->getChannel()->queue_bind($this->queueName, $this->exchangeName, $this->bindingKey);

   $msg = new AMQPMessage($data, ['delivery_mode' => $this->deliveryMode]);

   $this->connection->getChannel()->basic_publish($msg, $this->exchangeName, $this->routingKey);
}

Consumer

public function __construct(Connection $connection, $exchangeName, $exchangeType, $queueName, $bindingKey, $deadLetterExchangeName, $deadLetterExchangeType, deadLetterQueueName, $deadLetterRoutingKey, $passive = false, $durable = true, $autoDelete = false, $deliveryMode = 2)
{
   $this->connection = $connection;
   $this->exchangeName = $exchangeName;
   $this->exchangeType = $exchangeType;
   $this->queueName = $queueName;
   $this->passive = $passive;
   $this->durable = $durable;
   $this->autoDelete = $autoDelete;
   $this->deliveryMode = $deliveryMode;
   $this->bindingKey = $bindingKey;
   $this->deadLetterExchangeName = $deadLetterExchangeName;
   $this->deadLetterExchangeType = $deadLetterExchangeType;
   $this->deadLetterQueueName = $deadLetterQueueName;
   $this->deadLetterRoutingKey = $deadLetterRoutingKey;
}

public function listenToQueue($handlerClass, ExceptionHandler $exceptionHandler)
{
// Declare DLX
if(isset($this->deadLetterExchangeName)) {
   $this->connection->getChannel()->exchange_declare($this->deadLetterExchangeName, $this->deadLetterExchangeType);

   $this->connection->getChannel()->queue_declare($this->deadLetterQueueName, $this->passive, $this->durable, false, $this->autoDelete);

   $this->connection->getChannel()->queue_bind($this->deadLetterQueueName, $this->deadLetterExchangeName);
}

   $this->connection->getChannel()->exchange_declare($this->exchangeName, $this->exchangeType, $this->passive, $this->durable, $this->autoDelete);

   $this->connection->getChannel()->queue_declare($this->queueName, $this->passive, $this->durable, false, $this->autoDelete, false, ['x-dead-letter-exchange' => ['S', $this->deadLetterExchangeName], 'x-dead-letter-routing-key' => ['S', $this->deadLetterRoutingKey]]);

   $this->connection->getChannel()->queue_bind($this->queueName, $this->exchangeName, $this->bindingKey);

$handler = new $handlerClass;

$callback = function ($msg) use ($handler, $exceptionHandler) {
   //
};

$this->connection->getChannel()->basic_qos(null, 1, null);
$this->connection->getChannel()->basic_consume($queue_name, '', false, false, false, false, $callback);

while (count($this->connection->getChannel()->callbacks)) {
    $this->connection->getChannel()->wait();
}
}