. * * @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\Stream\Transport; use Amp\ByteStream\ClosedException; use Amp\ByteStream\ReadableStream; use Amp\ByteStream\WritableStream; use Amp\Cancellation; use Amp\Socket\Socket; use danog\MadelineProto\Logger; use danog\MadelineProto\Stream\ConnectionContext; use danog\MadelineProto\Stream\ProxyStreamInterface; use danog\MadelineProto\Stream\RawStreamInterface; use Throwable; use Webmozart\Assert\Assert; /** * Premade stream wrapper. * * Manages reading data in chunks * * @author Daniil Gentili * * @implements ProxyStreamInterface */ final class PremadeStream implements RawStreamInterface, ProxyStreamInterface { private Socket|ReadableStream|null $stream = null; public function __construct() { } public function setupTls(?Cancellation $cancellationToken = null): void { \assert($this->stream instanceof Socket); $this->stream->setupTls($cancellationToken); } public function getStream() { return $this->stream; } #[\Override] public function connect(ConnectionContext $ctx, string $header = ''): void { if ($header !== '') { $this->stream->write($header); } } /** * Async chunked read. */ #[\Override] public function read(?Cancellation $cancellation = null): ?string { return $this->stream ? $this->stream->read($cancellation) : null; } /** * Async write. * * @param string $data Data to write */ #[\Override] public function write(string $data): void { if (!$this->stream) { throw new ClosedException('MadelineProto stream was disconnected'); } \assert($this->stream instanceof WritableStream); /** @psalm-suppress UndefinedInterfaceMethod */ $this->stream->write($data); } /** * Async close. */ #[\Override] public function disconnect(): void { try { if ($this->stream) { if (method_exists($this->stream, 'close')) { $this->stream->close(); } $this->stream = null; } } catch (Throwable $e) { Logger::log('Got exception while closing stream: '.$e->getMessage()); } } public function close(): void { $this->disconnect(); } #[\Override] public function getSocket(): Socket { Assert::true($this->stream instanceof Socket); return $this->stream; } #[\Override] public function setExtra(mixed $extra): void { $this->stream = $extra; } public static function getName(): string { return self::class; } }