WyriHaximus\React\ObservableBunny\ObservableBunny

Complexity

Total Complexity

7

Size/Duplication

Total Lines	92
Duplicated Lines	0 %

Coupling/Cohesion

Components	1
Dependencies	6

Test Coverage

Coverage

80.95%

Importance

Changes

0

4 Methods

Rating	Name	Duplication	Size	Complexity
A	__construct()	0	5	1
A	consume()	0	50	3
A	cancelSubscription()	0	8	1
A	onError()	0	9	2

<?php declare(strict_types=1);

namespace WyriHaximus\React\ObservableBunny;

use Bunny\Async\Client;
use Bunny\Channel;
use Bunny\Message as BunnyMessage;
use Bunny\Protocol\MethodBasicConsumeOkFrame;
use React\EventLoop\LoopInterface;
use React\EventLoop\Timer\TimerInterface;
use React\Promise\PromiseInterface;
use Rx\Subject\Subject;
use Throwable;

final class ObservableBunny
{
    /**
     * @var LoopInterface
     */
    private $loop;

    /**
     * @var Client
     */
    private $bunny;

    /**
     * @param LoopInterface $loop
     * @param Client        $bunny
     */
    public function __construct(LoopInterface $loop, Client $bunny)
    {
        $this->loop = $loop;
        $this->bunny = $bunny;
    }

    public function consume(
        string $queue = '',
        string $consumerTag = '',
        bool $noLocal = false,
        bool $noAck = false,
        bool $exclusive = false,
        bool $nowait = false,
        array $arguments = []
    ): Subject {
        $subject = new Subject();
        $consumeArgs = [$queue, $consumerTag, $noLocal, $noAck, $exclusive, $nowait, $arguments];

        $channel = $this->bunny->channel();
        $channel->then(function (Channel $channel) use ($subject, $consumeArgs) {

The method then does only exist in React\Promise\PromiseInterface, but not in Bunny\Channel.

            /** @var string $consumerTag */
            $consumerTag = null;
            $timer = $this->loop->addPeriodicTimer(1, function () use ($channel, $subject, &$timer, &$consumerTag) {
                if (!$subject->isDisposed()) {
                    return;
                }

                $this->cancelSubscription(

It seems like you code against a concrete implementation and not the interface React\Promise\PromiseInterface as the method done() does only exist in the following implementations of said interface: React\Promise\FulfilledPromise, React\Promise\LazyPromise, React\Promise\Promise, React\Promise\RejectedPromise.

                    $timer,
                    $channel,
                    $consumerTag
                )->done([$subject, 'onComplete'], $this->onError($subject, $timer));
            });
            $channel->consume(

The method then does only exist in React\Promise\PromiseInterface, but not in Bunny\Protocol\MethodBasicConsumeOkFrame.

                function (BunnyMessage $message, Channel $channel) use ($subject, &$timer, &$consumerTag) {
                    if ($subject->isDisposed()) {
                        $channel->nack($message);
                        $this->cancelSubscription(

It seems like you code against a concrete implementation and not the interface React\Promise\PromiseInterface as the method done() does only exist in the following implementations of said interface: React\Promise\FulfilledPromise, React\Promise\LazyPromise, React\Promise\Promise, React\Promise\RejectedPromise.

                            $timer,
                            $channel,
                            $consumerTag
                        )->done([$subject, 'onComplete'], $this->onError($subject, $timer));

                        return;
                    }

                    $subject->onNext(new Message($message, $channel));
                },
                ...$consumeArgs

$consumeArgs is of type array<integer,string|boolean|array,{"0":"string","1":"string","2":"boolean","3":"boolean","4":"boolean","5":"boolean","6":"array"}>, but the function expects a string.

            )->then(function (MethodBasicConsumeOkFrame $response) use (&$consumerTag) {
                $consumerTag = $response->consumerTag;
            })->done(null, $this->onError($subject, $timer));
        })->done(null, [$subject, 'onError']);

        return $subject;
    }

    private function cancelSubscription(TimerInterface $timer, Channel $channel, string $consumerTag): PromiseInterface
    {
        $this->loop->cancelTimer($timer);

        return $channel->cancel($consumerTag)->then(function () use ($channel) {

The method then does only exist in React\Promise\PromiseInterface, but not in Bunny\Protocol\MethodBasicCancelOkFrame.

            return $channel->close();
        });
    }

    private function onError(Subject $subject, TimerInterface $timer): callable
    {
        return function (Throwable $et) use ($subject, $timer) {
            if ($this->loop->isTimerActive($timer)) {
                $this->loop->cancelTimer($timer);
            }
            $subject->onError($et);
        };
    }
}