Lines 100.00% 12 / 12
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% 11 / 11 100.00% 1 / 1 3
15final readonly class DistinctUntilChangedOperator implements StreamSource
16{
17    /**
18     * @param StreamSource<T> $source
19     * @param Closure(T): mixed $key
20     * @param Closure(T): string $partition
21     */
22    public function __construct(
23        private StreamSource $source,
24        private Closure $key,
25        private Closure $partition,
26    ) {}
27
28    public function attach(SubscriptionScope $scope, Closure $downstream): Subscription
29    {
30        $key = $this->key;
31        $partition = $this->partition;
32        /** @var array<string, mixed> $previousByPartition */
33        $previousByPartition = [];
34
35        return $this->source->attach($scope, static function (mixed $event) use ($key, $partition, $downstream, &$previousByPartition): void {
36            $current = $key($event);
37            $partitionKey = $partition($event);
38
39            if (\array_key_exists($partitionKey, $previousByPartition) && $previousByPartition[$partitionKey] === $current) {
40                return;
41            }
42
43            $previousByPartition[$partitionKey] = $current;
44
45            $downstream($event);
46        });
47    }
48}