<?php

namespace Comitium5\DesignerBundle\Command\RabbitMqConsumers;

use PhpAmqpLib\Channel\AMQPChannel;
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Wire\AMQPTable;
use Symfony\Component\Console\Command\Command;
use Symfony\Component\Console\Input\InputInterface;
use Symfony\Component\Console\Input\InputOption;
use Symfony\Component\Console\Output\OutputInterface;

/**
 * Class AbstractRabbitMqConsumerCommand
 *
 * @author Oscar Jimenez <oscarjg19.developer@gmail.com>
 * @package Comitium5\DesignerBundle\Command\RabbitMqConsumers
 */
abstract class AbstractRabbitMqConsumerCommand extends Command
{
    const DEFAULT_EXCHANGE = "comitium.topic";
    const DEFAULT_MAX_MESSAGES = 10000;

    /**
     * @var int
     */
    protected $messagesProcessed;

    /**
     * @return string
     */
    abstract protected function commandName(): string;

    /**
     * @return string
     */
    abstract protected function commandDescription(): string;

    /**
     * @param OutputInterface $output
     * @param int $maxMessages
     *
     * @return mixed
     */
    abstract protected function getClosure(OutputInterface $output, int $maxMessages);

    /**
     *
     */
    protected function configure(): void
    {
        $this
            ->setName($this->commandName())
            ->setDescription($this->commandDescription())
            ->setDefinition([
                new InputOption(
                    'host',
                    null,
                    InputOption::VALUE_REQUIRED,
                    "Rabbit host",
                    "rabbit"
                ),
                new InputOption(
                    'username',
                    null,
                    InputOption::VALUE_REQUIRED,
                    "Rabbit user name",
                    "rabbitmq"
                ),
                new InputOption(
                    'password',
                    null,
                    InputOption::VALUE_REQUIRED,
                    "Rabbit password",
                    "rabbitmq"
                ),
                new InputOption(
                    'port',
                    null,
                    InputOption::VALUE_REQUIRED,
                    "Rabbit port",
                    5672
                ),
                new InputOption(
                    'exchange',
                    null,
                    InputOption::VALUE_OPTIONAL,
                    "Rabbit exchange",
                    self::DEFAULT_EXCHANGE
                ),
                new InputOption(
                    'routing-key',
                    null,
                    InputOption::VALUE_REQUIRED,
                    "A routing key which the queue will be subscribed. Ex: comitium.site.subsite.cms.1.event.page.created"
                ),
                new InputOption(
                    'queue-name',
                    null,
                    InputOption::VALUE_REQUIRED,
                    "A queue name. Ex: comitium-clientA.create_page_on_comitium_created_page"
                ),
            ])
        ;
    }

    /**
     * @param InputInterface $input
     * @param OutputInterface $output
     *
     * @return int
     * @throws \ErrorException
     */
    protected function execute(InputInterface $input, OutputInterface $output): int
    {
        $connection = new AMQPStreamConnection(
            $input->getOption('host'),
            $input->getOption('port'),
            $input->getOption('username'),
            $input->getOption('password')
        );

        $channel = $this->buildChannel(
            $connection,
            $input->getOption('exchange'),
            $input->getOption('queue-name'),
            $input->getOption('routing-key'),
            $this->getClosure($output, self::DEFAULT_MAX_MESSAGES)
        );

        $output->writeln(
            "[*] Waiting for messages. To exit press CTRL+C",
            OutputInterface::VERBOSITY_VERBOSE
        );

        while (count($channel->callbacks)) {
            $channel->wait();
        }

        return 0;
    }

    /**
     * @param $exchange
     * @param $queueName
     *
     * @return string
     */
    protected function buildQueueName($exchange, $queueName): string
    {
        return sprintf('%s_%s', $exchange, $queueName);
    }

    /**
     * @param string $host
     * @param int $port
     * @param string $user
     * @param string $password
     *
     * @return AMQPStreamConnection
     */
    protected function rabbitConnection(string $host, int $port, string $user, string $password): AMQPStreamConnection
    {
        return new AMQPStreamConnection(
            $host,
            $port,
            $user,
            $password
        );
    }

    /**
     * @param AMQPStreamConnection $connection
     * @param $exchange
     * @param string $queueName
     * @param $route
     * @param $callback
     *
     * @return AMQPChannel
     */
    protected function buildChannel(
        AMQPStreamConnection $connection,
        string $exchange,
        string $queueName,
        string $route,
        $callback
    ): AMQPChannel {
        $channel = $connection->channel();

        $this->buildDeadLetter($channel, $exchange);

        $channel->exchange_declare(
            $exchange,
            'topic',
            false,
            true,
            false
        );

        $channel->queue_declare(
            $queueName,
            false,
            true,
            false,
            false,
            false,
            new AMQPTable([
                'x-dead-letter-exchange' => $this->buildDlxName($exchange),
                "x-queue-master-locator" => "min-masters"
            ])
        );

        $channel->queue_bind($queueName, $exchange, $route);

        // Avoid to receive more than one message if the consumer is busy
        $channel->basic_qos(
            null,
            1,
            null
        );

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

        return $channel;
    }

    /**
     * @param OutputInterface $output
     * @param $maxMessages
     *
     * @throws \Exception
     */
    protected function handleLimitMessages(OutputInterface $output, $maxMessages): void
    {
        $this->messagesProcessed++;

        if ($this->messagesProcessed > $maxMessages) {
            throw new \Exception("Max number of messages reached. Force exit to restart from supervisor");
        } else {
            $output->writeln(
                sprintf('Message processed:'. $this->messagesProcessed . ' Limit: ' . $maxMessages),
                OutputInterface::VERBOSITY_VERBOSE
            );
        }
    }

    /**
     * @param AMQPChannel $channel
     * @param $exchange
     */
    protected function buildDeadLetter(AMQPChannel $channel, $exchange): void
    {
        $queueName = $exchange . "_dlx_queue";

        $channel->exchange_declare(
            $this->buildDlxName($exchange),
            'topic',
            false,
            true,
            false
        );

        $channel->queue_declare(
            $queueName,
            false,
            true,
            false,
            false,
            false,
            new AMQPTable([
                "x-queue-master-locator" => "min-masters"
            ])
        );

        $channel->queue_bind($queueName, $this->buildDlxName($exchange), "#");
    }

    /**
     * @param $exchange
     *
     * @return string
     */
    private function buildDlxName($exchange): string
    {
        return $exchange . "_dlx";
    }
}
