Lines 100.00% 14 / 14
Methods 100.00% 2 / 2
Classes 100.00% 1 / 1
Name Lines Methods CRAP
 __construct 100.00% 1 / 1 100.00% 1 / 1 1
 attach 100.00% 13 / 13 100.00% 1 / 1 4
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}