Darkilliant\MqProcessBundle\Message\Publisher\AmqpMessagePublisher

Complexity

Total Complexity

6

Size/Duplication

Total Lines	47
Duplicated Lines	0 %

Importance

Changes

0

5 Methods

Rating	Name	Duplication	Size	Complexity
A	publish()	0	7	2
A	countMessages()	0	3	1
A	flush()	0	3	1
A	__construct()	0	9	1
A	finalize()	0	3	1

<?php

declare(strict_types=1);

namespace Darkilliant\MqProcessBundle\Message\Publisher;

use PhpAmqpLib\Channel\AMQPChannel;
use PhpAmqpLib\Message\AMQPMessage;


class AmqpMessagePublisher
{
    /** @var AMQPChannel */
    private $channel;
    /** @var string */
    private $queue;
    /** @var string */
    private $exchange;
    /** @var int */
    private $batch = 0;
    /** @var bool */
    private $persistant;

    public function __construct(AMQPChannel $channel, string $queue, string $exchange, bool $persistant)
    {
        $this->channel = $channel;
        $this->queue = $queue;
        $this->exchange = $exchange;
        $this->persistant = $persistant;

        $this->channel->queue_declare($this->queue, false, $this->persistant, false, false);
        $this->channel->queue_bind($this->queue, $this->exchange, $this->queue);
    }

    public function publish(string $message, int $batchCount)
    {
        ++$this->batch;
        $this->channel->batch_basic_publish(new AMQPMessage($message), $this->exchange, $this->queue);

        if ($this->batch >= $batchCount) {
            $this->flush();
        }
    }

    public function finalize()
    {
        $this->flush();
    }

    public function flush()
    {
        $this->channel->publish_batch();
    }

    public function countMessages(): int
    {
        return $this->channel->queue_declare($this->queue, true)->messageCount;
    }
}