Linio\Component\Queue\Adapter\RabbitAdapter

Complexity

Total Complexity

8

Size/Duplication

Total Lines	59
Duplicated Lines	0 %

Coupling/Cohesion

Components	1
Dependencies	4

Importance

Changes

0

<?php
class RabbitAdapter implements AdapterInterface
{
    protected ?AMQPChannel $channel = null;

This code did not parse for me. Apparently, there is an error somewhere around this line:

Syntax error, unexpected '?', expecting T_FUNCTION or T_CONST

    protected array $config = [];

    public function __construct(array $config = [])
    {
        $this->config = $config;
    }

    public function add(Job $job): void
    {
        $this->getChannel()->queue_declare($job->getQueue(), true, $job->isPersistent(), false, false);
        $message = new AMQPMessage($job->getPayload(), ['delivery_mode' => 2]);
        $this->getChannel()->basic_publish($message, '', $job->getQueue());
    }

    public function perform(Job $job): void
    {
        $this->getChannel()->queue_declare($job->getQueue(), true, $job->isPersistent(), false, false);
        $this->getChannel()->basic_qos(0, 1, false);
        $this->getChannel()->basic_consume($job->getQueue(), '', false, false, false, false, function ($message) use ($job): void {
            $job->setPayload($message->body);
            $job->perform();

            if ($job->isFinished()) {
                $message->delivery_info['channel']->basic_ack($message->delivery_info['delivery_tag']);
            } else {
                $this->getChannel()->basic_nack($message->delivery_info['delivery_tag'], false, true);
            }
        });

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

    public function getChannel(): AMQPChannel
    {
        if (!$this->channel) {
            $connection = new AMQPConnection($this->config['host'], $this->config['port'], $this->config['username'], $this->config['password'], $this->config['vhost']);
            $this->channel = $connection->channel();
        }

        return $this->channel;
    }

    public function setChannel(AMQPChannel $channel): void
    {
        $this->channel = $channel;
    }
}