WyriHaximus\Rx\ObservableWhile

Complexity

Total Complexity

7

Size/Duplication

Total Lines	57
Duplicated Lines	0 %

Coupling/Cohesion

Components	1
Dependencies	2

Test Coverage

Coverage

100%

Importance

Changes

0

2 Methods

Rating	Name	Duplication	Size	Complexity
A	__construct()	0	21	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 \SplQueue
     */
    private $queue;

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

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

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

                return;
            }

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

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

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

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

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

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