CommandListener.php

<?php

namespace App;

use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;

class CommandListener
{
    /**
     * @var AMQPStreamConnection
     */
    private $connection;

    /**
     * @var string
     */
    private $exchangeName;

    /**
     * @var string
     */
    private $exchangeType = 'direct';

    /**
     * @var string
     */
    private $queueName;

    /**
     * @var CommandMap
     */
    private $commandMap;

    public function __construct(AMQPStreamConnection $connection, CommandMap $map, string $exchange, string $queue)
    {
        $this->connection = $connection;
        $this->commandMap = $map;
        $this->exchangeName = $exchange;
        $this->queueName = $queue;
    }

    /**
     * @param string $exchangeType
     * @return CommandListener
     */
    public function setExchangeType(string $exchangeType): CommandListener
    {
        $this->exchangeType = $exchangeType;
        return $this;
    }
    
    public function listen()
    {
        $channel = $this->connection->channel();
        $channel->exchange_declare($this->exchangeName, $this->exchangeType, false, true, false);

        echo " [*] Waiting for messages. To exit press CTRL + C..." . PHP_EOL;

        $callback = function (AMQPMessage $msg) {
            echo ' [x] ', $msg->delivery_info['routing_key'], ':', $msg->body, "\n";

            $className = $this->commandMap->getClassName($msg->delivery_info['routing_key']);
            $payload = json_decode($msg->body);
            $handler = new $className($payload);
            $handler->handle();
        };

        foreach ($this->commandMap->getRoutingKeys() as $key) {
            $channel->queue_bind($this->queueName, $this->exchangeName, $key);
        }

        $channel->basic_consume($this->queueName, '', false, true, false, false, $callback);

        while ($channel->is_consuming()) {
            $channel->wait();
        }

        $channel->close();
        $this->connection->close();
    }
}

CommandMap.php

<?php

namespace App;

use OutOfBoundsException;

class CommandMap
{
    /**
     * @var array
     */
    private $classMap = [];

    public function registerRoutingKey(string $key, string $className)
    {
        $this->classMap[$key] = $className;
        return $this;
    }

    public function getRoutingKeys(): array
    {
        return array_keys($this->classMap);
    }

    /**
     * @param string $key
     * @return string
     */
    public function getClassName(string $key): string
    {
        if (!isset($this->classMap[$key])) {
            throw new OutOfBoundsException('Classmap with key "'. $key .'" don\'t exists.');
        }

        return $this->classMap[$key];
    }
}

consumer.php

<?php

require_once __DIR__ . '/vendor/autoload.php';

use App\WelcomeClientMail;
use App\CommandMap;
use GoSale\RabbitMQ\Connection;
use App\CommandListener;

/**
 * Load application environment from .env file
 */
$dotenv = Dotenv\Dotenv::create(__DIR__);
$dotenv->load();

$connection = (new Connection(
    getenv('RABBITMQ_HOST'),
    getenv('RABBITMQ_PORT'),
    getenv('RABBITMQ_DEFAULT_USER'),
    getenv('RABBITMQ_DEFAULT_PASS')
))->connect();

$commandMap = new CommandMap();
$commandMap->registerRoutingKey('client.mail.welcome', WelcomeClientMail::class);

$commandListener = new CommandListener($connection, $commandMap, 'notification', 'mail');
$commandListener->listen();