WyriHaximus\Rx\ObservableWhile

Complexity

Total Complexity

7

Size/Duplication

Total Lines	56
Duplicated Lines	0 %

Coupling/Cohesion

Components	1
Dependencies	2

Test Coverage

Coverage

84.21%

Importance

Changes

0

2 Methods

Rating	Name	Duplication	Size	Complexity
A	__construct()	0	20	3
A	get()	0	14	4

<?php declare(strict_types=1);

namespace WyriHaximus\Rx;

use React\Promise\Deferred;
use React\Promise\PromiseInterface;
use Rx\ObservableInterface;
use function React\Promise\resolve;

final class ObservableWhile
{
    /**
     * @var array
     */
    private $queue = [];

    /**
     * @var Deferred
     */
    private $deferred;

    /**
     * @var bool
     */
    private $done = false;

    /**
     * @param ObservableInterface $observable
     */
    public function __construct(ObservableInterface $observable)
    {
        $observable->subscribe(function ($item) {
            if ($this->deferred instanceof Deferred) {
                $this->deferred->resolve($item);
                $this->deferred = null;

                return;
            }

            $this->queue[] = $item;
        }, null, function () {
            $this->done = true;

            if ($this->deferred instanceof Deferred) {
                $this->deferred->resolve();
                $this->deferred = null;
            }
        });
    }

    public function get(): PromiseInterface
    {
        if (count($this->queue) === 0 && $this->done === true) {
            return resolve();
        }

        if (count($this->queue) === 0) {
            $this->deferred = new Deferred();

            return $this->deferred->promise();
        }

        return resolve(array_shift($this->queue));
    }
}