AmqpWorkers\BatchProducer

Complexity

Total Complexity

8

Size/Duplication

Total Lines	48
Duplicated Lines	0 %

Coupling/Cohesion

Components	1
Dependencies	4

Importance

Changes	1
Bugs	0	Features	1

3 Methods

Rating	Name	Duplication	Size	Complexity
A	factory()	0	8	1
B	produce()	0	23	6
A	flush()	0	5	1

<?php

namespace AmqpWorkers;

use PhpAmqpLib\Connection\AbstractConnection;

/**
 * BatchProducer treats given message as a list of messages and send them to RabbitMQ channel
 *
 * @package AmqpWorkers
 * @author Alex Panshin <deadyaga@gmail.com>
 * @since 1.0
 */
class BatchProducer extends Producer
{
    private $limit = 0;
    private $queued = 0;

    public static function factory(AbstractConnection $connection, $limit = 100)
    {
        /** @var BatchProducer $producer */
        $producer = parent::factory($connection);
        $producer->limit = $limit;

        return $producer;
    }

    /**
     * @param array|\Traversable $messages
     * @throws \AmqpWorkers\Exception\ProducerNotProperlyConfigured
     */
    public function produce($messages)
    {
        $this->selfCheck();

        $channel = $this->getChannel();
        foreach ($messages as $payload) {
            if ($this->isExchange()) {
                $channel->batch_basic_publish($this->createMessage($payload), $this->exchange->getName());
            } else {
                $channel->batch_basic_publish($this->createMessage($payload), '', $this->queue->getName());
            }

            $this->queued++;

            if ($this->limit && $this->limit <= $this->queued) {
                $this->flush();
            }
        }

        if ($this->queued) {
            $this->flush();
        }
    }

    protected function flush()
    {
        $this->getChannel()->publish_batch();
        $this->queued = 0;
    }
}