WyriHaximus\Recoil\QueueCaller

Complexity

Total Complexity

4

Size/Duplication

Total Lines	44
Duplicated Lines	0 %

Coupling/Cohesion

Components	1
Dependencies	3

Test Coverage

Coverage

91.3%

Importance

Changes

0

2 Methods

Rating	Name	Duplication	Size	Complexity
A	__construct()	0	4	1
A	call()	0	28	3

<?php declare(strict_types=1);

namespace WyriHaximus\Recoil;

use Recoil\Kernel;
use Rx\ObservableInterface;
use function WyriHaximus\Rx\observableWhile;

final class QueueCaller implements QueueCallerInterface
{
    /**
     * @var Kernel
     */
    private $kernel;

    /**
     * @param Kernel $kernel
     */
    public function __construct(Kernel $kernel)
    {
        $this->kernel = $kernel;
    }

    public function call(ObservableInterface $observable): State
    {
        $state = new State();
        $this->kernel->execute(function () use ($observable, $state) {
            $state->onNext(State::STARTED);
            yield;
            $observableWhile = observableWhile($observable);
            $state->onNext(State::WAITING);
            /** @var Call $call */
            while ($call = (yield $observableWhile->get())) {
                try {
                    $state->onNext(State::BUSY);
                    $callable = $call->getCallable();
                    $arguments = $call->getArguments();
                    $value = yield $callable(...$arguments);
                    $call->resolve($value);
                } catch (\Throwable $et) {
                    $call->reject($et);
                } finally {
                    unset($callable, $arguments, $call);
                    $state->onNext(State::WAITING);
                }
            }
            $state->onNext(State::DONE);
        });

        return $state;
    }
}