.github/workflows/ci.yml000064400000007774144760122720011303 0ustar00name: Continuous Integration on: push: null pull_request: branches: - master jobs: unit_tests: strategy: matrix: include: - operating-system: 'ubuntu-latest' php-version: '7.1' - operating-system: 'ubuntu-latest' php-version: '7.2' - operating-system: 'ubuntu-latest' php-version: '7.3' - operating-system: 'ubuntu-latest' php-version: '7.4' - operating-system: 'ubuntu-latest' php-version: '8.0' composer-flags: '--ignore-platform-req=php' - operating-system: 'windows-latest' php-version: '8.0' composer-flags: '--ignore-platform-req=php' - operating-system: 'macos-latest' php-version: '8.0' composer-flags: '--ignore-platform-req=php' name: PHP ${{ matrix.php-version }} on ${{ matrix.operating-system }} runs-on: ${{ matrix.operating-system }} steps: - name: Setup PHP uses: shivammathur/setup-php@v2 with: php-version: ${{ matrix.php-version }} - name: Use LF line ends run: | git config --global core.autocrlf false git config --global core.eol lf - name: Checkout code uses: actions/checkout@v2 - name: Get Composer cache directory id: composer-cache run: echo "::set-output name=dir::$(composer config cache-dir)" - name: Cache dependencies uses: actions/cache@v2 with: path: ${{ steps.composer-cache.outputs.dir }} key: composer-${{ runner.os }}-${{ matrix.php-version }}-${{ hashFiles('**/composer.*') }}-${{ matrix.composer-flags }} restore-keys: | composer-${{ runner.os }}-${{ matrix.php-version }}-${{ hashFiles('**/composer.*') }}- composer-${{ runner.os }}-${{ matrix.php-version }}- - name: Install dependencies uses: nick-invision/retry@v2 with: timeout_minutes: 5 max_attempts: 5 retry_wait_seconds: 30 command: | php_version=$(php -v) composer update --optimize-autoloader --no-interaction --no-progress ${{ matrix.composer-flags }} composer info -D - name: Run unit tests run: vendor/bin/phpunit --verbose coding_standards: strategy: matrix: include: - operating-system: 'ubuntu-latest' php-version: '8.0' composer-flags: '--ignore-platform-req=php' name: Coding standards runs-on: ${{ matrix.operating-system }} steps: - name: Setup PHP uses: shivammathur/setup-php@v2 with: php-version: ${{ matrix.php-version }} - name: Use LF line ends run: | git config --global core.autocrlf false git config --global core.eol lf - name: Checkout code uses: actions/checkout@v2 - name: Get Composer cache directory id: composer-cache run: echo "::set-output name=dir::$(composer config cache-dir)" - name: Cache dependencies uses: actions/cache@v2 with: path: ${{ steps.composer-cache.outputs.dir }} key: composer-${{ runner.os }}-${{ matrix.php-version }}-${{ hashFiles('**/composer.*') }}-${{ matrix.composer-flags }} restore-keys: | composer-${{ runner.os }}-${{ matrix.php-version }}-${{ hashFiles('**/composer.*') }}- composer-${{ runner.os }}-${{ matrix.php-version }}- - name: Install dependencies uses: nick-invision/retry@v2 with: timeout_minutes: 5 max_attempts: 5 retry_wait_seconds: 30 command: | php_version=$(php -v) composer update --optimize-autoloader --no-interaction --no-progress ${{ matrix.composer-flags }} composer info -D - name: Run style fixer env: PHP_CS_FIXER_IGNORE_ENV: 1 run: vendor/bin/php-cs-fixer --diff --dry-run -v fix LICENSE000064400000002066144760122720005562 0ustar00 The MIT License (MIT) Copyright (c) 2016-2021 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.json000064400000002561144760122720007277 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.php000064400000003344144760122720013735 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.php000064400000002620144760122720014132 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.php000064400000002063144760122720013744 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.php000064400000001641144760122720014146 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.php000064400000000132144760122720011114 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.php000064400000001441144760122720010303 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.php000064400000002123144760122720011244 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.php000064400000003347144760122720011004 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.php000064400000003312144760122720010041 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.php000064400000011277144760122720007424 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.php000064400000002202144760122720010456 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.php000064400000002434144760122720010507 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.php000064400000000712144760122720011222 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 (\is_resource($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.php000064400000024624144760122720012224 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 (\is_resource($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.php000064400000000117144760122720011141 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.php000064400000006104144760122720011326 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.php000064400000011034144760122720010037 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.xml000064400000003134144760122720006410 0ustar00