. * * @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\Common; use Amp\ByteStream\ClosedException; use Amp\Cancellation; use Amp\File\File; use Amp\Socket\Socket; use danog\MadelineProto\Exception; use danog\MadelineProto\Stream\BufferedStreamInterface; use danog\MadelineProto\Stream\BufferInterface; use danog\MadelineProto\Stream\ConnectionContext; use danog\MadelineProto\Stream\ProxyStreamInterface; use danog\MadelineProto\Stream\RawStreamInterface; use RuntimeException; /** * Buffered raw stream. * * @author Daniil Gentili * * @implements ProxyStreamInterface */ final class FileBufferedStream implements BufferedStreamInterface, BufferInterface, ProxyStreamInterface, RawStreamInterface { private ?File $stream = null; private int $append_after; private string $append; /** * Connect. */ #[\Override] public function connect(ConnectionContext $ctx, string $header = ''): void { if ($header !== '') { $this->stream->write($header); } } /** * Async chunked read. */ #[\Override] public function read(?Cancellation $token = null): ?string { if (!$this->stream) { throw new ClosedException('MadelineProto stream was disconnected'); } return $this->stream->read($token); } /** * 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'); } $this->stream->write($data); } /** * Async close. */ #[\Override] public function disconnect(): void { if ($this->stream) { $this->stream = null; } } /** * Get read buffer asynchronously. * * @param int $length Length of payload, as detected by this layer */ #[\Override] public function getReadBuffer(?int &$length): \danog\MadelineProto\Stream\ReadBufferInterface { if (!$this->stream) { throw new ClosedException('MadelineProto stream was disconnected'); } return $this; } /** * Get write buffer asynchronously. * * @param int $length Total length of data that is going to be piped in the buffer */ #[\Override] public function getWriteBuffer(int $length, string $append = ''): \danog\MadelineProto\Stream\WriteBufferInterface { if (\strlen($append)) { $this->append = $append; $this->append_after = $length - \strlen($append); } return $this; } /** * Read data asynchronously. * * @param int $length Amount of data to read */ #[\Override] public function bufferRead(int $length, ?Cancellation $cancellation = null): ?string { return $this->stream->read($cancellation, $length); } /** * Async write. * * @param string $data Data to write */ #[\Override] public function bufferWrite(string $data): void { if ($this->append_after) { $this->append_after -= \strlen($data); if ($this->append_after === 0) { $data .= $this->append; $this->append = ''; } elseif ($this->append_after < 0) { $this->append_after = 0; $this->append = ''; throw new Exception('Tried to send too much out of frame data, cannot append'); } } $this->write($data); } /** * Set file handle. */ #[\Override] public function setExtra($extra): void { $this->stream = $extra; } /** * {@inheritDoc} */ #[\Override] public function getStream(): RawStreamInterface { throw new RuntimeException("Can't get underlying RawStreamInterface, is a File handle!"); } /** * {@inheritDoc} */ #[\Override] public function getSocket(): Socket { throw new RuntimeException("Can't get underlying socket, is a File handle!"); } /** * Get class name. */ #[\Override] public static function getName(): string { return self::class; } }