AmqpMessagePublisher::__construct() - Code Metrics - Inspection of "Je veux scaler avec un système de message queue" - jean-pasqualini/import - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — 0.4 (#35)

by jean

created 2018-07-17 06:53 UTC

AmqpMessagePublisher::__construct() A

↳ Parent: AmqpMessagePublisher

Complexity

Conditions	1
Paths	1

Size

Total Lines	9
Code Lines	6

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	6
nc	1
nop	4
dl	0
loc	9
rs	10
c	0
b	0
f	0

<?php

declare(strict_types=1);

namespace Darkilliant\MqProcessBundle\Message\Publisher;

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

class AmqpMessagePublisher implements MessagePublisherInterface
{
    /** @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(Message $message, int $batchCount)
    {
        ++$this->batch;
        $this->channel->batch_basic_publish($this->buildMessage($message), $this->exchange, $this->queue);

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

    private function buildMessage(Message $message)
    {
        return new AMQPMessage($message->getBody());
    }

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

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

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


1			<?php
2
3			declare(strict_types=1);
4
5			namespace Darkilliant\MqProcessBundle\Message\Publisher;
6
7			use Darkilliant\MqProcessBundle\Message\Message;
8			use PhpAmqpLib\Channel\AMQPChannel;
9			use PhpAmqpLib\Message\AMQPMessage;
10
11			class AmqpMessagePublisher implements MessagePublisherInterface
12			{
13			/** @var AMQPChannel */
14			private $channel;
15			/** @var string */
16			private $queue;
17			/** @var string */
18			private $exchange;
19			/** @var int */
20			private $batch = 0;
21			/** @var bool */
22			private $persistant;
23
24			public function __construct(AMQPChannel $channel, string $queue, string $exchange, bool $persistant)
25			{
26			$this->channel = $channel;
27			$this->queue = $queue;
28			$this->exchange = $exchange;
29			$this->persistant = $persistant;
30
31			$this->channel->queue_declare($this->queue, false, $this->persistant, false, false);
32			$this->channel->queue_bind($this->queue, $this->exchange, $this->queue);
33			}
34
35			public function publish(Message $message, int $batchCount)
36			{
37			++$this->batch;
38			$this->channel->batch_basic_publish($this->buildMessage($message), $this->exchange, $this->queue);
39
40			if ($this->batch >= $batchCount) {
41			$this->flush();
42			}
43			}
44
45			private function buildMessage(Message $message)
46			{
47			return new AMQPMessage($message->getBody());
48			}
49
50			public function finalize()
51			{
52			$this->flush();
53			}
54
55			public function flush()
56			{
57			$this->channel->publish_batch();
58			}
59
60			public function countMessages(): int
61			{
62			return $this->channel->queue_declare($this->queue, true)[1];
63			}
64			}
65

jean-pasqualini / import

Pull Request — 0.4 (#35)

AmqpMessagePublisher::__construct() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like