LICENSE000064400000002066144760117130005561 0ustar00 The MIT License (MIT) Copyright (c) 2016-2020 amphp Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions: The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software. THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. composer.json000064400000002561144760117130007276 0ustar00{ "name": "amphp/byte-stream", "homepage": "http://amphp.org/byte-stream", "description": "A stream abstraction to make working with non-blocking I/O simple.", "support": { "issues": "https://github.com/amphp/byte-stream/issues", "irc": "irc://irc.freenode.org/amphp" }, "keywords": [ "stream", "async", "non-blocking", "amp", "amphp", "io" ], "license": "MIT", "authors": [ { "name": "Aaron Piotrowski", "email": "aaron@trowski.com" }, { "name": "Niklas Keller", "email": "me@kelunik.com" } ], "require": { "php": ">=7.1", "amphp/amp": "^2" }, "require-dev": { "amphp/phpunit-util": "^1.4", "phpunit/phpunit": "^6 || ^7 || ^8", "friendsofphp/php-cs-fixer": "^2.3", "amphp/php-cs-fixer-config": "dev-master", "psalm/phar": "^3.11.4", "jetbrains/phpstorm-stubs": "^2019.3" }, "autoload": { "psr-4": { "Amp\\ByteStream\\": "lib" }, "files": [ "lib/functions.php" ] }, "autoload-dev": { "psr-4": { "Amp\\ByteStream\\Test\\": "test" } }, "extra": { "branch-alias": { "dev-master": "1.x-dev" } } } lib/Base64/Base64DecodingInputStream.php000064400000003344144760117130013734 0ustar00source = $source; } public function read(): Promise { return call(function () { if ($this->source === null) { throw new StreamException('Failed to read stream chunk due to invalid base64 data'); } $chunk = yield $this->source->read(); if ($chunk === null) { if ($this->buffer === null) { return null; } $chunk = \base64_decode($this->buffer, true); if ($chunk === false) { $this->source = null; $this->buffer = null; throw new StreamException('Failed to read stream chunk due to invalid base64 data'); } $this->buffer = null; return $chunk; } $this->buffer .= $chunk; $length = \strlen($this->buffer); $chunk = \base64_decode(\substr($this->buffer, 0, $length - $length % 4), true); if ($chunk === false) { $this->source = null; $this->buffer = null; throw new StreamException('Failed to read stream chunk due to invalid base64 data'); } $this->buffer = \substr($this->buffer, $length - $length % 4); return $chunk; }); } } lib/Base64/Base64DecodingOutputStream.php000064400000002620144760117130014131 0ustar00destination = $destination; } public function write(string $data): Promise { $this->buffer .= $data; $length = \strlen($this->buffer); $chunk = \base64_decode(\substr($this->buffer, 0, $length - $length % 4), true); if ($chunk === false) { return new Failure(new StreamException('Invalid base64 near offset ' . $this->offset)); } $this->offset += $length - $length % 4; $this->buffer = \substr($this->buffer, $length - $length % 4); return $this->destination->write($chunk); } public function end(string $finalData = ""): Promise { $this->offset += \strlen($this->buffer); $chunk = \base64_decode($this->buffer . $finalData, true); if ($chunk === false) { return new Failure(new StreamException('Invalid base64 near offset ' . $this->offset)); } $this->buffer = ''; return $this->destination->end($chunk); } } lib/Base64/Base64EncodingInputStream.php000064400000002063144760117130013743 0ustar00source = $source; } public function read(): Promise { return call(function () { $chunk = yield $this->source->read(); if ($chunk === null) { if ($this->buffer === null) { return null; } $chunk = \base64_encode($this->buffer); $this->buffer = null; return $chunk; } $this->buffer .= $chunk; $length = \strlen($this->buffer); $chunk = \base64_encode(\substr($this->buffer, 0, $length - $length % 3)); $this->buffer = \substr($this->buffer, $length - $length % 3); return $chunk; }); } } lib/Base64/Base64EncodingOutputStream.php000064400000001641144760117130014145 0ustar00destination = $destination; } public function write(string $data): Promise { $this->buffer .= $data; $length = \strlen($this->buffer); $chunk = \base64_encode(\substr($this->buffer, 0, $length - $length % 3)); $this->buffer = \substr($this->buffer, $length - $length % 3); return $this->destination->write($chunk); } public function end(string $finalData = ""): Promise { $chunk = \base64_encode($this->buffer . $finalData); $this->buffer = ''; return $this->destination->end($chunk); } } lib/ClosedException.php000064400000000132144760117130011113 0ustar00contents = $contents; } /** * Reads data from the stream. * * @return Promise Resolves with the full contents or `null` if the stream has closed / already been consumed. */ public function read(): Promise { if ($this->contents === null) { return new Success; } $promise = new Success($this->contents); $this->contents = null; return $promise; } } lib/InputStream.php000064400000001441144760117130010302 0ustar00read()) !== null) { * $buffer .= $chunk; * } * * return $buffer; * }); * } * ``` */ interface InputStream { /** * Reads data from the stream. * * @return Promise Resolves with a string when new data is available or `null` if the stream has closed. * * @psalm-return Promise * * @throws PendingReadError Thrown if another read operation is still pending. */ public function read(): Promise; } lib/InputStreamChain.php000064400000002123144760117130011243 0ustar00streams = $streams; } /** @inheritDoc */ public function read(): Promise { if ($this->reading) { throw new PendingReadError; } if (!$this->streams) { return new Success(null); } return call(function () { $this->reading = true; try { while ($this->streams) { $chunk = yield $this->streams[0]->read(); if ($chunk === null) { \array_shift($this->streams); continue; } return $chunk; } return null; } finally { $this->reading = false; } }); } } lib/IteratorStream.php000064400000003347144760117130011003 0ustar00 */ private $iterator; /** @var \Throwable|null */ private $exception; /** @var bool */ private $pending = false; /** * @psam-param Iterator $iterator */ public function __construct(Iterator $iterator) { $this->iterator = $iterator; } /** @inheritdoc */ public function read(): Promise { if ($this->exception) { return new Failure($this->exception); } if ($this->pending) { throw new PendingReadError; } $this->pending = true; /** @var Deferred $deferred */ $deferred = new Deferred; $this->iterator->advance()->onResolve(function ($error, $hasNextElement) use ($deferred) { $this->pending = false; if ($error) { $this->exception = $error; $deferred->fail($error); } elseif ($hasNextElement) { $chunk = $this->iterator->getCurrent(); if (!\is_string($chunk)) { $this->exception = new StreamException(\sprintf( "Unexpected iterator value of type '%s', expected string", \is_object($chunk) ? \get_class($chunk) : \gettype($chunk) )); $deferred->fail($this->exception); return; } $deferred->resolve($chunk); } else { $deferred->resolve(); } }); return $deferred->promise(); } } lib/LineReader.php000064400000003312144760117130010040 0ustar00source = $inputStream; $this->delimiter = $delimiter === null ? "\n" : $delimiter; $this->lineMode = $delimiter === null; } /** * @return Promise */ public function readLine(): Promise { return call(function () { if (false !== \strpos($this->buffer, $this->delimiter)) { list($line, $this->buffer) = \explode($this->delimiter, $this->buffer, 2); return $this->lineMode ? \rtrim($line, "\r") : $line; } while (null !== $chunk = yield $this->source->read()) { $this->buffer .= $chunk; if (false !== \strpos($this->buffer, $this->delimiter)) { list($line, $this->buffer) = \explode($this->delimiter, $this->buffer, 2); return $this->lineMode ? \rtrim($line, "\r") : $line; } } if ($this->buffer === "") { return null; } $line = $this->buffer; $this->buffer = ""; return $this->lineMode ? \rtrim($line, "\r") : $line; }); } public function getBuffer(): string { return $this->buffer; } /** * @return void */ public function clearBuffer() { $this->buffer = ""; } } lib/Message.php000064400000011277144760117130007423 0ustar00read()) !== null) { * // Immediately use $chunk, reducing memory consumption since the entire message is never buffered. * } * * @deprecated Use Amp\ByteStream\Payload instead. */ class Message implements InputStream, Promise { /** @var InputStream */ private $source; /** @var string */ private $buffer = ""; /** @var Deferred|null */ private $pendingRead; /** @var Coroutine|null */ private $coroutine; /** @var bool True if onResolve() has been called. */ private $buffering = false; /** @var Deferred|null */ private $backpressure; /** @var bool True if the iterator has completed. */ private $complete = false; /** @var \Throwable|null Used to fail future reads on failure. */ private $error; /** * @param InputStream $source An iterator that only emits strings. */ public function __construct(InputStream $source) { $this->source = $source; } private function consume(): \Generator { while (($chunk = yield $this->source->read()) !== null) { $buffer = $this->buffer .= $chunk; if ($buffer === "") { continue; // Do not succeed reads with empty string. } elseif ($this->pendingRead) { $deferred = $this->pendingRead; $this->pendingRead = null; $this->buffer = ""; $deferred->resolve($buffer); $buffer = ""; // Destroy last emitted chunk to free memory. } elseif (!$this->buffering) { $buffer = ""; // Destroy last emitted chunk to free memory. $this->backpressure = new Deferred; yield $this->backpressure->promise(); } } $this->complete = true; if ($this->pendingRead) { $deferred = $this->pendingRead; $this->pendingRead = null; $deferred->resolve($this->buffer !== "" ? $this->buffer : null); $this->buffer = ""; } return $this->buffer; } /** @inheritdoc */ final public function read(): Promise { if ($this->pendingRead) { throw new PendingReadError; } if ($this->coroutine === null) { $this->coroutine = new Coroutine($this->consume()); $this->coroutine->onResolve(function ($error) { if ($error) { $this->error = $error; } if ($this->pendingRead) { $deferred = $this->pendingRead; $this->pendingRead = null; $deferred->fail($error); } }); } if ($this->error) { return new Failure($this->error); } if ($this->buffer !== "") { $buffer = $this->buffer; $this->buffer = ""; if ($this->backpressure) { $backpressure = $this->backpressure; $this->backpressure = null; $backpressure->resolve(); } return new Success($buffer); } if ($this->complete) { return new Success; } $this->pendingRead = new Deferred; return $this->pendingRead->promise(); } /** @inheritdoc */ final public function onResolve(callable $onResolved) { $this->buffering = true; if ($this->coroutine === null) { $this->coroutine = new Coroutine($this->consume()); } if ($this->backpressure) { $backpressure = $this->backpressure; $this->backpressure = null; $backpressure->resolve(); } $this->coroutine->onResolve($onResolved); } /** * Exposes the source input stream. * * This might be required to resolve a promise with an InputStream, because promises in Amp can't be resolved with * other promises. * * @return InputStream */ final public function getInputStream(): InputStream { return $this->source; } } lib/OutputBuffer.php000064400000002202144760117130010455 0ustar00deferred = new Deferred; } public function write(string $data): Promise { if ($this->closed) { throw new ClosedException("The stream has already been closed."); } $this->contents .= $data; return new Success(\strlen($data)); } public function end(string $finalData = ""): Promise { if ($this->closed) { throw new ClosedException("The stream has already been closed."); } $this->contents .= $finalData; $this->closed = true; $this->deferred->resolve($this->contents); $this->contents = ""; return new Success(\strlen($finalData)); } public function onResolve(callable $onResolved) { $this->deferred->promise()->onResolve($onResolved); } } lib/OutputStream.php000064400000002434144760117130010506 0ustar00stream = $stream; } public function __destruct() { if (!$this->promise) { Promise\rethrow(new Coroutine($this->consume())); } } private function consume(): \Generator { try { if ($this->lastRead && null === yield $this->lastRead) { return; } while (null !== yield $this->stream->read()) { // Discard unread bytes from message. } } catch (\Throwable $exception) { // If exception is thrown here the connection closed anyway. } } /** * @inheritdoc * * @throws \Error If a buffered message was requested by calling buffer(). */ final public function read(): Promise { if ($this->promise) { throw new \Error("Cannot stream message data once a buffered message has been requested"); } return $this->lastRead = $this->stream->read(); } /** * Buffers the entire message and resolves the returned promise then. * * @return Promise Resolves with the entire message contents. */ final public function buffer(): Promise { if ($this->promise) { return $this->promise; } return $this->promise = call(function () { $buffer = ''; if ($this->lastRead && null === yield $this->lastRead) { return $buffer; } while (null !== $chunk = yield $this->stream->read()) { $buffer .= $chunk; } return $buffer; }); } } lib/PendingReadError.php000064400000000712144760117130011221 0ustar00useSingleRead = $useSingleRead; if (\strpos($meta["mode"], "r") === false && \strpos($meta["mode"], "+") === false) { throw new \Error("Expected a readable stream"); } \stream_set_blocking($stream, false); \stream_set_read_buffer($stream, 0); $this->resource = &$stream; $this->chunkSize = &$chunkSize; $deferred = &$this->deferred; $readable = &$this->readable; $this->watcher = Loop::onReadable($this->resource, static function ($watcher) use ( &$deferred, &$readable, &$stream, &$chunkSize, $useSingleRead ) { if ($useSingleRead) { $data = @\fread($stream, $chunkSize); } else { $data = @\stream_get_contents($stream, $chunkSize); } \assert($data !== false, "Trying to read from a previously fclose()'d resource. Do NOT manually fclose() resources the loop still has a reference to."); // Error suppression, because pthreads does crazy things with resources, // which might be closed during two operations. // See https://github.com/amphp/byte-stream/issues/32 if ($data === '' && @\feof($stream)) { $readable = false; $stream = null; $data = null; // Stream closed, resolve read with null. Loop::cancel($watcher); } else { Loop::disable($watcher); } $temp = $deferred; $deferred = null; \assert($temp instanceof Deferred); $temp->resolve($data); }); $this->immediateCallable = static function ($watcherId, $data) use (&$deferred) { $temp = $deferred; $deferred = null; \assert($temp instanceof Deferred); $temp->resolve($data); }; Loop::disable($this->watcher); } /** @inheritdoc */ public function read(): Promise { if ($this->deferred !== null) { throw new PendingReadError; } if (!$this->readable) { return new Success; // Resolve with null on closed stream. } \assert($this->resource !== null); // Attempt a direct read, because Windows suffers from slow I/O on STDIN otherwise. if ($this->useSingleRead) { $data = @\fread($this->resource, $this->chunkSize); } else { $data = @\stream_get_contents($this->resource, $this->chunkSize); } \assert($data !== false, "Trying to read from a previously fclose()'d resource. Do NOT manually fclose() resources the loop still has a reference to."); if ($data === '') { // Error suppression, because pthreads does crazy things with resources, // which might be closed during two operations. // See https://github.com/amphp/byte-stream/issues/32 if (@\feof($this->resource)) { $this->readable = false; $this->resource = null; Loop::cancel($this->watcher); return new Success; // Stream closed, resolve read with null. } $this->deferred = new Deferred; Loop::enable($this->watcher); return $this->deferred->promise(); } // Prevent an immediate read → write loop from blocking everything // See e.g. examples/benchmark-throughput.php $this->deferred = new Deferred; $this->immediateWatcher = Loop::defer($this->immediateCallable, $data); return $this->deferred->promise(); } /** * Closes the stream forcefully. Multiple `close()` calls are ignored. * * @return void */ public function close() { if ($this->resource) { // Error suppression, as resource might already be closed $meta = @\stream_get_meta_data($this->resource); if ($meta && \strpos($meta["mode"], "+") !== false) { @\stream_socket_shutdown($this->resource, \STREAM_SHUT_RD); } else { /** @psalm-suppress InvalidPropertyAssignmentValue */ @\fclose($this->resource); } } $this->free(); } /** * Nulls reference to resource, marks stream unreadable, and succeeds any pending read with null. * * @return void */ private function free() { $this->readable = false; $this->resource = null; if ($this->deferred !== null) { $deferred = $this->deferred; $this->deferred = null; $deferred->resolve(); } Loop::cancel($this->watcher); if ($this->immediateWatcher !== null) { Loop::cancel($this->immediateWatcher); } } /** * @return resource|null The stream resource or null if the stream has closed. */ public function getResource() { return $this->resource; } /** * @return void */ public function setChunkSize(int $chunkSize) { $this->chunkSize = $chunkSize; } /** * References the read watcher, so the loop keeps running in case there's an active read. * * @return void * * @see Loop::reference() */ public function reference() { if (!$this->resource) { throw new \Error("Resource has already been freed"); } Loop::reference($this->watcher); } /** * Unreferences the read watcher, so the loop doesn't keep running even if there are active reads. * * @return void * * @see Loop::unreference() */ public function unreference() { if (!$this->resource) { throw new \Error("Resource has already been freed"); } Loop::unreference($this->watcher); } public function __destruct() { if ($this->resource !== null) { $this->free(); } } } lib/ResourceOutputStream.php000064400000024606144760117130012223 0ustar00 */ private $writes; /** @var bool */ private $writable = true; /** @var int|null */ private $chunkSize; /** * @param resource $stream Stream resource. * @param int|null $chunkSize Chunk size per `fwrite()` operation. */ public function __construct($stream, int $chunkSize = null) { if (!\is_resource($stream) || \get_resource_type($stream) !== 'stream') { throw new \Error("Expected a valid stream"); } $meta = \stream_get_meta_data($stream); if (\strpos($meta["mode"], "r") !== false && \strpos($meta["mode"], "+") === false) { throw new \Error("Expected a writable stream"); } \stream_set_blocking($stream, false); \stream_set_write_buffer($stream, 0); $this->resource = $stream; $this->chunkSize = &$chunkSize; $writes = $this->writes = new \SplQueue; $writable = &$this->writable; $resource = &$this->resource; $this->watcher = Loop::onWritable($stream, static function ($watcher, $stream) use ($writes, &$chunkSize, &$writable, &$resource) { static $emptyWrites = 0; try { while (!$writes->isEmpty()) { /** @var Deferred $deferred */ list($data, $previous, $deferred) = $writes->shift(); $length = \strlen($data); if ($length === 0) { $deferred->resolve(0); continue; } if (!\is_resource($stream) || (($metaData = @\stream_get_meta_data($stream)) && $metaData['eof'])) { throw new ClosedException("The stream was closed by the peer"); } // Error reporting suppressed since fwrite() emits E_WARNING if the pipe is broken or the buffer is full. // Use conditional, because PHP doesn't like getting null passed if ($chunkSize) { $written = @\fwrite($stream, $data, $chunkSize); } else { $written = @\fwrite($stream, $data); } \assert( $written !== false || \PHP_VERSION_ID >= 70400, // PHP 7.4+ returns false on EPIPE. "Trying to write on a previously fclose()'d resource. Do NOT manually fclose() resources the still referenced in the loop." ); // PHP 7.4.0 and 7.4.1 may return false on EAGAIN. if ($written === false && \PHP_VERSION_ID >= 70402) { $message = "Failed to write to stream"; if ($error = \error_get_last()) { $message .= \sprintf("; %s", $error["message"]); } throw new StreamException($message); } // Broken pipes between processes on macOS/FreeBSD do not detect EOF properly. if ($written === 0 || $written === false) { if ($emptyWrites++ > self::MAX_CONSECUTIVE_EMPTY_WRITES) { $message = "Failed to write to stream after multiple attempts"; if ($error = \error_get_last()) { $message .= \sprintf("; %s", $error["message"]); } throw new StreamException($message); } $writes->unshift([$data, $previous, $deferred]); return; } $emptyWrites = 0; if ($length > $written) { $data = \substr($data, $written); $writes->unshift([$data, $written + $previous, $deferred]); return; } $deferred->resolve($written + $previous); } } catch (\Throwable $exception) { $resource = null; $writable = false; /** @psalm-suppress PossiblyUndefinedVariable */ $deferred->fail($exception); while (!$writes->isEmpty()) { list(, , $deferred) = $writes->shift(); $deferred->fail($exception); } Loop::cancel($watcher); } finally { if ($writes->isEmpty()) { Loop::disable($watcher); } } }); Loop::disable($this->watcher); } /** * Writes data to the stream. * * @param string $data Bytes to write. * * @return Promise Succeeds once the data has been successfully written to the stream. * * @throws ClosedException If the stream has already been closed. */ public function write(string $data): Promise { return $this->send($data, false); } /** * Closes the stream after all pending writes have been completed. Optionally writes a final data chunk before. * * @param string $finalData Bytes to write. * * @return Promise Succeeds once the data has been successfully written to the stream. * * @throws ClosedException If the stream has already been closed. */ public function end(string $finalData = ""): Promise { return $this->send($finalData, true); } private function send(string $data, bool $end = false): Promise { if (!$this->writable) { return new Failure(new ClosedException("The stream is not writable")); } $length = \strlen($data); $written = 0; if ($end) { $this->writable = false; } if ($this->writes->isEmpty()) { if ($length === 0) { if ($end) { $this->close(); } return new Success(0); } if (!\is_resource($this->resource) || (($metaData = @\stream_get_meta_data($this->resource)) && $metaData['eof'])) { return new Failure(new ClosedException("The stream was closed by the peer")); } // Error reporting suppressed since fwrite() emits E_WARNING if the pipe is broken or the buffer is full. // Use conditional, because PHP doesn't like getting null passed. if ($this->chunkSize) { $written = @\fwrite($this->resource, $data, $this->chunkSize); } else { $written = @\fwrite($this->resource, $data); } \assert( $written !== false || \PHP_VERSION_ID >= 70400, // PHP 7.4+ returns false on EPIPE. "Trying to write on a previously fclose()'d resource. Do NOT manually fclose() resources the still referenced in the loop." ); // PHP 7.4.0 and 7.4.1 may return false on EAGAIN. if ($written === false && \PHP_VERSION_ID >= 70402) { $message = "Failed to write to stream"; if ($error = \error_get_last()) { $message .= \sprintf("; %s", $error["message"]); } return new Failure(new StreamException($message)); } $written = (int) $written; // Cast potential false to 0. if ($length === $written) { if ($end) { $this->close(); } return new Success($written); } $data = \substr($data, $written); } $deferred = new Deferred; if ($length - $written > self::LARGE_CHUNK_SIZE) { $chunks = \str_split($data, self::LARGE_CHUNK_SIZE); $data = \array_pop($chunks); foreach ($chunks as $chunk) { $this->writes->push([$chunk, $written, new Deferred]); $written += self::LARGE_CHUNK_SIZE; } } $this->writes->push([$data, $written, $deferred]); Loop::enable($this->watcher); $promise = $deferred->promise(); if ($end) { $promise->onResolve([$this, "close"]); } return $promise; } /** * Closes the stream forcefully. Multiple `close()` calls are ignored. * * @return void */ public function close() { if ($this->resource) { // Error suppression, as resource might already be closed $meta = @\stream_get_meta_data($this->resource); if ($meta && \strpos($meta["mode"], "+") !== false) { @\stream_socket_shutdown($this->resource, \STREAM_SHUT_WR); } else { /** @psalm-suppress InvalidPropertyAssignmentValue psalm reports this as closed-resource */ @\fclose($this->resource); } } $this->free(); } /** * Nulls reference to resource, marks stream unwritable, and fails any pending write. * * @return void */ private function free() { $this->resource = null; $this->writable = false; if (!$this->writes->isEmpty()) { $exception = new ClosedException("The socket was closed before writing completed"); do { /** @var Deferred $deferred */ list(, , $deferred) = $this->writes->shift(); $deferred->fail($exception); } while (!$this->writes->isEmpty()); } Loop::cancel($this->watcher); } /** * @return resource|null Stream resource or null if end() has been called or the stream closed. */ public function getResource() { return $this->resource; } /** * @return void */ public function setChunkSize(int $chunkSize) { $this->chunkSize = $chunkSize; } public function __destruct() { if ($this->resource !== null) { $this->free(); } } } lib/StreamException.php000064400000000117144760117130011140 0ustar00source = $source; $this->encoding = $encoding; $this->options = $options; $this->resource = @\inflate_init($encoding, $options); if ($this->resource === false) { throw new StreamException("Failed initializing deflate context"); } } /** @inheritdoc */ public function read(): Promise { return call(function () { if ($this->resource === null) { return null; } \assert($this->source !== null); $data = yield $this->source->read(); // Needs a double guard, as stream might have been closed while reading /** @psalm-suppress ParadoxicalCondition */ if ($this->resource === null) { return null; } if ($data === null) { $decompressed = @\inflate_add($this->resource, "", \ZLIB_FINISH); if ($decompressed === false) { throw new StreamException("Failed adding data to deflate context"); } $this->close(); return $decompressed; } $decompressed = @\inflate_add($this->resource, $data, \ZLIB_SYNC_FLUSH); if ($decompressed === false) { throw new StreamException("Failed adding data to deflate context"); } return $decompressed; }); } /** * @internal * @return void */ private function close() { $this->resource = null; $this->source = null; } /** * Gets the used compression encoding. * * @return int Encoding specified on construction time. */ public function getEncoding(): int { return $this->encoding; } /** * Gets the used compression options. * * @return array Options array passed on construction time. */ public function getOptions(): array { return $this->options; } } lib/ZlibOutputStream.php000064400000006104144760117130011325 0ustar00destination = $destination; $this->encoding = $encoding; $this->options = $options; $this->resource = @\deflate_init($encoding, $options); if ($this->resource === false) { throw new StreamException("Failed initializing deflate context"); } } /** @inheritdoc */ public function write(string $data): Promise { if ($this->resource === null) { throw new ClosedException("The stream has already been closed"); } \assert($this->destination !== null); $compressed = \deflate_add($this->resource, $data, \ZLIB_SYNC_FLUSH); if ($compressed === false) { throw new StreamException("Failed adding data to deflate context"); } $promise = $this->destination->write($compressed); $promise->onResolve(function ($error) { if ($error) { $this->close(); } }); return $promise; } /** @inheritdoc */ public function end(string $finalData = ""): Promise { if ($this->resource === null) { throw new ClosedException("The stream has already been closed"); } \assert($this->destination !== null); $compressed = \deflate_add($this->resource, $finalData, \ZLIB_FINISH); if ($compressed === false) { throw new StreamException("Failed adding data to deflate context"); } $promise = $this->destination->end($compressed); $promise->onResolve(function () { $this->close(); }); return $promise; } /** * @internal * @return void */ private function close() { $this->resource = null; $this->destination = null; } /** * Gets the used compression encoding. * * @return int Encoding specified on construction time. */ public function getEncoding(): int { return $this->encoding; } /** * Gets the used compression options. * * @return array Options array passed on construction time. */ public function getOptions(): array { return $this->options; } } lib/functions.php000064400000011034144760117130010036 0ustar00read()) !== null) { $written += \strlen($chunk); $writePromise = $destination->write($chunk); $chunk = null; // free memory yield $writePromise; } return $written; }); } /** * @param \Amp\ByteStream\InputStream $source * * @return \Amp\Promise */ function buffer(InputStream $source): Promise { return call(function () use ($source): \Generator { $buffer = ""; while (($chunk = yield $source->read()) !== null) { $buffer .= $chunk; $chunk = null; // free memory } return $buffer; }); } /** * The php://input input buffer stream for the process associated with the currently active event loop. * * @return ResourceInputStream */ function getInputBufferStream(): ResourceInputStream { static $key = InputStream::class . '\\input'; $stream = Loop::getState($key); if (!$stream) { $stream = new ResourceInputStream(\fopen('php://input', 'rb')); Loop::setState($key, $stream); } return $stream; } /** * The php://output output buffer stream for the process associated with the currently active event loop. * * @return ResourceOutputStream */ function getOutputBufferStream(): ResourceOutputStream { static $key = OutputStream::class . '\\output'; $stream = Loop::getState($key); if (!$stream) { $stream = new ResourceOutputStream(\fopen('php://output', 'wb')); Loop::setState($key, $stream); } return $stream; } /** * The STDIN stream for the process associated with the currently active event loop. * * @return ResourceInputStream */ function getStdin(): ResourceInputStream { static $key = InputStream::class . '\\stdin'; $stream = Loop::getState($key); if (!$stream) { $stream = new ResourceInputStream(\STDIN); Loop::setState($key, $stream); } return $stream; } /** * The STDOUT stream for the process associated with the currently active event loop. * * @return ResourceOutputStream */ function getStdout(): ResourceOutputStream { static $key = OutputStream::class . '\\stdout'; $stream = Loop::getState($key); if (!$stream) { $stream = new ResourceOutputStream(\STDOUT); Loop::setState($key, $stream); } return $stream; } /** * The STDERR stream for the process associated with the currently active event loop. * * @return ResourceOutputStream */ function getStderr(): ResourceOutputStream { static $key = OutputStream::class . '\\stderr'; $stream = Loop::getState($key); if (!$stream) { $stream = new ResourceOutputStream(\STDERR); Loop::setState($key, $stream); } return $stream; } function parseLineDelimitedJson(InputStream $stream, bool $assoc = false, int $depth = 512, int $options = 0): Iterator { return new Producer(static function (callable $emit) use ($stream, $assoc, $depth, $options) { $reader = new LineReader($stream); while (null !== $line = yield $reader->readLine()) { $line = \trim($line); if ($line === '') { continue; } /** @noinspection PhpComposerExtensionStubsInspection */ $data = \json_decode($line, $assoc, $depth, $options); /** @noinspection PhpComposerExtensionStubsInspection */ $error = \json_last_error(); /** @noinspection PhpComposerExtensionStubsInspection */ if ($error !== \JSON_ERROR_NONE) { /** @noinspection PhpComposerExtensionStubsInspection */ throw new StreamException('Failed to parse JSON: ' . \json_last_error_msg(), $error); } yield $emit($data); } }); } psalm.xml000064400000003134144760117130006407 0ustar00