OldSound\RabbitMqBundle\ExecuteReceiverStrategy\BatchExecuteReceiverStrategy

Complexity

Total Complexity

11

Size/Duplication

Total Lines	56
Duplicated Lines	0 %

Importance

Changes

0

8 Methods

Rating	Name	Duplication	Size	Complexity
A	isBatchCompleted()	0	3	1
A	isBatchEmpty()	0	3	1
A	onConsumeCallback()	0	6	2
A	onMessageProcessed()	0	3	1
A	__construct()	0	3	1
A	canPrecessMultiMessages()	0	3	1
A	onCatchTimeout()	0	4	2
A	onStopConsuming()	0	4	2

<?php

namespace OldSound\RabbitMqBundle\ExecuteReceiverStrategy;

use OldSound\RabbitMqBundle\Declarations\ConsumeOptions;
use OldSound\RabbitMqBundle\ExecuteCallbackStrategy\ExecuteReceiverStrategyInterface;

This use statement conflicts with another class in this namespace, OldSound\RabbitMqBundle\ExecuteReceiverStrategy\ExecuteReceiverStrategyInterface. Consider defining an alias.

The type OldSound\RabbitMqBundle\ExecuteCallbackStrategy\ExecuteReceiverStrategyInterface was not found. Maybe you did not declare it correctly or list all dependencies?

use PhpAmqpLib\Exception\AMQPTimeoutException;
use PhpAmqpLib\Message\AMQPMessage;

class BatchExecuteReceiverStrategy extends AbstractExecuteReceiverStrategy
{
    /** @var int */
    private $batchCount;
    /** @var AMQPMessage[] */
    protected $messagesBatch = [];

    public function __construct(int $batchCount)
    {
        $this->batchCount = $batchCount;
    }

    public function canPrecessMultiMessages(): bool
    {
        return true;
    }

    public function onConsumeCallback(AMQPMessage $message): ?array
    {
        $this->messagesBatch[$message->getDeliveryTag()] = $message;

        if ($this->isBatchCompleted()) {
            return $this->execute($this->messagesBatch);
        }
    }

    public function onMessageProcessed(AMQPMessage $message)
    {
        unset($this->messagesBatch[array_search($message, $this->messagesBatch, true)]);
    }

    public function onCatchTimeout(AMQPTimeoutException $e)
    {
        if (!$this->isBatchEmpty()) {
            $this->processMessages($this->messagesBatch);

The method processMessages() does not exist on OldSound\RabbitMqBundle\ExecuteReceiverStrategy\BatchExecuteReceiverStrategy.

        }
    }

    public function onStopConsuming()
    {
        if (!$this->isBatchEmpty()) {
            $this->processMessages($this->messagesBatch);
        }
    }

    protected function isBatchCompleted(): bool
    {
        return count($this->messagesBatch) === $this->batchCount;
    }

    /**
     * @return  bool
     */
    protected function isBatchEmpty()
    {
        return count($this->messagesBatch) === 0;
    }
}