. * * @author Daniil Gentili * @copyright 2016-2025 Daniil Gentili * @license https://opensource.org/licenses/AGPL-3.0 AGPLv3 * @link https://docs.madelineproto.xyz MadelineProto documentation */ namespace danog\MadelineProto\Reactive; use danog\Loop\GenericLoop; use SplQueue; /** * @template T * * @internal * * @implements Subscriber */ final class Actor implements Subscriber { /** @var SplQueue */ private readonly SplQueue $queue; private GenericLoop $loop; public function __construct( /** @var Subscriber $subscriber */ private readonly Subscriber $subscriber ) { /** @var SplQueue */ $this->queue = new SplQueue; $this->queue->setIteratorMode(SplQueue::IT_MODE_DELETE); $this->__wakeup(); } public function __sleep() { return ['queue', 'subscriber']; } public function __wakeup(): void { $this->loop = new GenericLoop(function (): ?float { foreach ($this->queue as $item) { if (\count($item) === 1) { $this->subscriber->onAttach($item[0]); } else { $this->subscriber->onStateChange($item[0], $item[1]); } } return GenericLoop::PAUSE; }, ''); } #[\Override] public function onAttach($initState): void { $this->queue->enqueue([$initState]); if ($this->loop->isRunning()) { $this->loop->resume(true); } else { $this->loop->start(); } } #[\Override] public function onStateChange($prevState, $state): void { $this->queue->enqueue([$prevState, $state]); if ($this->loop->isRunning()) { $this->loop->resume(true); } else { $this->loop->start(); } } #[\Override] public function __toString(): string { return 'Actor<' . $this->subscriber::class . '>'; } }