Innmind\AMQP\Client\Channel\Basic\Basic

Complexity

Total Complexity

12

Size/Duplication

Total Lines	148
Duplicated Lines	0 %

Test Coverage

Coverage

90.36%

Importance

Changes

0

9 Methods

Rating	Name	Duplication	Size	Complexity
A	consume()	0	21	2
A	reject()	0	10	1
A	qos()	0	11	1
A	cancel()	0	14	2
A	get()	0	26	2
A	ack()	0	10	1
A	recover()	0	11	1
A	__construct()	0	5	1
A	publish()	0	16	1

<?php
declare(strict_types = 1);

namespace Innmind\AMQP\Client\Channel\Basic;

use Innmind\AMQP\{
    Client\Channel\Basic as BasicInterface,
    Model\Basic\Ack,
    Model\Basic\Cancel,
    Model\Basic\Consume,
    Model\Basic\Get as GetCommand,
    Model\Basic\Publish,
    Model\Basic\Qos,
    Model\Basic\Recover,
    Model\Basic\Reject,
    Model\Basic\Message\Locked,
    Transport\Connection,
    Transport\Connection\MessageReader,
    Transport\Frame,
    Transport\Frame\Channel,
};
use Innmind\TimeContinuum\ElapsedPeriod;

final class Basic implements BasicInterface
{
    private $connection;
    private $channel;
    private $read;

    public function __construct(Connection $connection, Channel $channel)
    {
        $this->connection = $connection;
        $this->channel = $channel;
        $this->read = new MessageReader;
    }

    public function ack(Ack $command): BasicInterface
    {
        $this->connection->send(
            $this->connection->protocol()->basic()->ack(
                $this->channel,
                $command
            )
        );

        return $this;
    }

    public function cancel(Cancel $command): BasicInterface
    {
        $this->connection->send(
            $this->connection->protocol()->basic()->cancel(
                $this->channel,
                $command
            )
        );

        if ($command->shouldWait()) {
            $this->connection->wait('basic.cancel-ok');
        }

        return $this;
    }

    public function consume(Consume $command): Consumer
    {
        $this->connection->send(
            $this->connection->protocol()->basic()->consume(
                $this->channel,
                $command
            )
        );

        if ($command->shouldWait()) {
            $frame = $this->connection->wait('basic.consume-ok');
            $consumerTag = (string) $frame->values()->first()->original();
        } else {
            $consumerTag = $command->consumerTag();
        }

        return new Consumer\Consumer(
            $this->connection,
            $command,
            $this->channel,
            $consumerTag
        );
    }

    public function get(GetCommand $command): Get
    {
        $frame = $this
            ->connection
            ->send($this->connection->protocol()->basic()->get(
                $this->channel,
                $command
            ))
            ->wait('basic.get-ok', 'basic.get-empty');

        if ($frame->is($this->connection->protocol()->method('basic.get-empty'))) {
            return new Get\GetEmpty;
        }

        $message = ($this->read)($this->connection);

        return new Get\GetOk(
            $this->connection,
            $this->channel,
            $command,
            new Locked($message),
            $frame->values()->first()->original()->value(), //deliveryTag
            $frame->values()->get(1)->original()->first(), //redelivered
            (string) $frame->values()->get(2)->original(), //exchange
            (string) $frame->values()->get(3)->original(), //routingKey
            $frame->values()->get(4)->original()->value() //messageCount
        );
    }

    public function publish(Publish $command): BasicInterface
    {
        $this
            ->connection
            ->protocol()
            ->basic()
            ->publish(
                $this->channel,
                $command,
                $this->connection->maxFrameSize()
            )
            ->foreach(function(Frame $frame): void {
                $this->connection->send($frame);
            });

        return $this;
    }

    public function qos(Qos $command): BasicInterface
    {
        $this
            ->connection
            ->send($this->connection->protocol()->basic()->qos(
                $this->channel,
                $command
            ))
            ->wait('basic.qos-ok');

        return $this;
    }

    public function recover(Recover $command): BasicInterface
    {
        $this
            ->connection
            ->send($this->connection->protocol()->basic()->recover(
                $this->channel,
                $command
            ))
            ->wait('basic.recover-ok');

        return $this;
    }

    public function reject(Reject $command): BasicInterface
    {
        $this->connection->send(
            $this->connection->protocol()->basic()->reject(
                $this->channel,
                $command
            )
        );

        return $this;
    }
}