1 | <?php declare(strict_types=1); |
||
12 | final class ObservableWhile |
||
13 | { |
||
14 | /** |
||
15 | * @var \SplQueue |
||
16 | */ |
||
17 | private $queue; |
||
18 | |||
19 | /** |
||
20 | * @var Deferred |
||
21 | */ |
||
22 | private $deferred; |
||
23 | |||
24 | /** |
||
25 | * @var bool |
||
26 | */ |
||
27 | private $done = false; |
||
28 | |||
29 | /** |
||
30 | * @param ObservableInterface $observable |
||
31 | */ |
||
32 | 6 | public function __construct(ObservableInterface $observable) |
|
33 | { |
||
34 | 6 | $this->queue = new \SplQueue(); |
|
35 | $observable->subscribe(function ($item): void { |
||
36 | 4 | if ($this->deferred instanceof Deferred) { |
|
37 | 2 | $deferred = $this->deferred; |
|
38 | 2 | $this->deferred = null; |
|
39 | 2 | $deferred->resolve($item); |
|
40 | |||
41 | 2 | return; |
|
42 | } |
||
43 | |||
44 | 3 | $this->queue->enqueue($item); |
|
45 | }, function (Throwable $error): void { |
||
46 | 2 | if ($this->deferred instanceof Deferred) { |
|
47 | 1 | $deferred = $this->deferred; |
|
48 | 1 | $this->deferred = null; |
|
49 | 1 | $deferred->reject($error); |
|
50 | |||
51 | 1 | return; |
|
52 | } |
||
53 | |||
54 | 1 | $this->queue->enqueue($error); |
|
55 | }, function (): void { |
||
56 | 4 | $this->done = true; |
|
57 | |||
58 | 4 | if ($this->deferred instanceof Deferred) { |
|
59 | 3 | $this->deferred->resolve(); |
|
60 | 3 | $this->deferred = null; |
|
61 | } |
||
62 | 6 | }); |
|
63 | 6 | } |
|
64 | |||
65 | 6 | public function get(): PromiseInterface |
|
85 | } |
||
86 |