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

Passed

Pull Request — 0.4 (#35)

by jean

created 2018-07-17 19:22 UTC

AmqpMessagePublisher::__construct() A

↳ Parent: AmqpMessagePublisher

Complexity

Conditions	1
Paths	1

Size

Total Lines	9
Code Lines	6

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	7
CRAP Score	1

Importance

Changes

Metric	Value
cc	1
eloc	6
nc	1
nop	4
dl	0
loc	9
ccs	7
cts	7
cp	1
crap	1
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();
        }
    }

    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];
    }

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


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	4	public function __construct(AMQPChannel $channel, string $queue, string $exchange, bool $persistant)
25		{
26	4	$this->channel = $channel;
27	4	$this->queue = $queue;
28	4	$this->exchange = $exchange;
29	4	$this->persistant = $persistant;
30
31	4	$this->channel->queue_declare($this->queue, false, $this->persistant, false, false);
32	4	$this->channel->queue_bind($this->queue, $this->exchange, $this->queue);
33	4	}
34
35	1	public function publish(Message $message, int $batchCount)
36		{
37	1	++$this->batch;
38	1	$this->channel->batch_basic_publish($this->buildMessage($message), $this->exchange, $this->queue);
39
40	1	if ($this->batch >= $batchCount) {
41	1	$this->flush();
42		}
43	1	}
44
45	1	public function finalize()
46		{
47	1	$this->flush();
48	1	}
49
50	2	public function flush()
51		{
52	2	$this->channel->publish_batch();
53	2	}
54
55	1	public function countMessages(): int
56		{
57	1	return $this->channel->queue_declare($this->queue, true)[1];
58		}
59
60	1	private function buildMessage(Message $message)
61		{
62	1	return new AMQPMessage($message->getBody());
63		}
64		}
65

jean-pasqualini / import

Pull Request — 0.4 (#35)

AmqpMessagePublisher::__construct() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like