Lines 100.00% 14 / 14
Functions and Methods 100.00% 2 / 2
Classes and Traits 100.00% 1 / 1
Name Lines Functions and Methods CRAP Classes and Traits
TakeOperator 100.00% 14 / 14 100.00% 2 / 2 5 100.00% 1 / 1
 __construct 100.00% 1 / 1 100.00% 1 / 1 1
 attach 100.00% 13 / 13 100.00% 1 / 1 4
1<?php
2
3declare(strict_types=1);
4
5namespace Stewart\Support\Stream;
6
7use Closure;
8use Stewart\Contracts\Subscription;
9
10/**
11 * @internal
12 * @template T
13 * @implements StreamSource<T>
14 */
15final readonly class TakeOperator implements StreamSource
16{
17    /** @param StreamSource<T> $source */
18    public function __construct(
19        private StreamSource $source,
20        private int $count,
21    ) {}
22
23    public function attach(SubscriptionScope $scope, Closure $downstream): Subscription
24    {
25        $remaining = $this->count;
26
27        if ($remaining <= 0) {
28            $scope->close();
29
30            return CancelledSubscription::createUnique();
31        }
32
33        $stage = $scope->claimStage();
34
35        return $this->source->attach($scope, static function (mixed $event) use ($scope, $stage, $downstream, &$remaining): void {
36            if ($remaining <= 0) {
37                return;
38            }
39
40            --$remaining;
41
42            try {
43                $downstream($event);
44            } finally {
45                if ($remaining <= 0) {
46                    $scope->closeWhenDownstreamSettles($stage);
47                }
48            }
49        });
50    }
51}