.editorconfig 0000644 00000000203 14476013052 0007216 0 ustar 00 root = true
[*]
end_of_line = lf
insert_final_newline = true
trim_trailing_whitespace = true
indent_style = space
charset = utf-8
.gitmodules 0000644 00000000140 14476013052 0006716 0 ustar 00 [submodule "docs/.shared"]
path = docs/.shared
url = https://github.com/amphp/amphp.github.io
.php_cs.dist 0000644 00000000344 14476013052 0006766 0 ustar 00 getFinder()->in(__DIR__);
$cacheDir = getenv('TRAVIS') ? getenv('HOME') . '/.php-cs-fixer' : __DIR__;
$config->setCacheFile($cacheDir . '/.php_cs.cache');
return $config;
LICENSE 0000644 00000002066 14476013052 0005557 0 ustar 00
The MIT License (MIT)
Copyright (c) 2016-2019 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.
README.md 0000644 00000002501 14476013052 0006023 0 ustar 00 [](https://amphp.org/byte-stream/)
`amphp/byte-stream` is a stream abstraction to make working with non-blocking I/O simple.
## Installation
This package can be installed as a [Composer](https://getcomposer.org/) dependency.
```bash
composer require amphp/byte-stream
```
## Requirements
- PHP 7.0+
## Documentation
Documentation is bundled within this repository in the [`./docs`](./docs) directory.
## Versioning
`amphp/byte-stream` follows the [semver](http://semver.org/) semantic versioning specification like all other `amphp` packages.
## Security
If you discover any security related issues, please email [`me@kelunik.com`](mailto:me@kelunik.com) instead of using the issue tracker.
## License
The MIT License (MIT). Please see [`LICENSE`](./LICENSE) for more information.
composer.json 0000644 00000002430 14476013052 0007267 0 ustar 00 {
"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": {
"amphp/amp": "^2"
},
"require-dev": {
"amphp/phpunit-util": "^1",
"phpunit/phpunit": "^6",
"friendsofphp/php-cs-fixer": "^2.3",
"amphp/php-cs-fixer-config": "dev-master",
"infection/infection": "^0.9.3"
},
"autoload": {
"psr-4": {
"Amp\\ByteStream\\": "lib"
},
"files": [
"lib/functions.php"
]
},
"autoload-dev": {
"psr-4": {
"Amp\\ByteStream\\Test\\": "test"
}
},
"config": {
"platform": {
"php": "7.0.0"
}
}
}
docs/Gemfile 0000644 00000000171 14476013052 0006770 0 ustar 00 source "https://rubygems.org"
gem "github-pages"
gem "kramdown"
gem "jekyll-github-metadata"
gem "jekyll-relative-links"
docs/README.md 0000644 00000001212 14476013052 0006751 0 ustar 00 # Documentation
This directory contains the documentation for `amphp/byte-stream`. Documentation and code are bundled within a single repository for easier maintenance. Additionally, this preserves the documentation for older versions.
## Reading
You can read this documentation either directly on GitHub or on our website. While the website will always contain the latest version, viewing on GitHub also works with older versions.
## Writing
Our documentation is built using Jekyll.
```
sudo gem install bundler jekyll
```
```
git submodule init
git submodule update
cd docs
bundle install --path vendor/bundle
bundle exec jekyll serve
```
docs/_config.yml 0000644 00000001371 14476013052 0007627 0 ustar 00 kramdown:
input: GFM
toc_levels: 2..3
baseurl: "/byte-stream"
layouts_dir: ".shared/layout"
includes_dir: ".shared/includes"
exclude: ["Gemfile", "Gemfile.lock", "README.md", "vendor"]
safe: true
repository: amphp/byte-stream
gems:
- "jekyll-github-metadata"
- "jekyll-relative-links"
defaults:
- scope:
path: ""
type: "pages"
values:
layout: "docs"
description: "amphp/byte-stream is a byte stream abstraction for Amp providing interfaces for InputStream and OutputStream."
keywords: ['amphp', 'amp', 'byte-stream', 'non-blocking', 'io', 'stream', 'compression']
shared_asset_path: "/byte-stream/asset"
navigation:
- resource-streams
- in-memory-stream
- iterator-stream
- compression-streams
- message
- payload
docs/compression-streams.md 0000644 00000002112 14476013052 0012031 0 ustar 00 ---
title: Compression Streams
permalink: /compression-streams
---
This package implements compression and decompression streams based on Zlib. `ZlibOutputStream` can be used for compression, while `ZlibInputStream` can be used for decompression. Both can simply wrap an existing stream to apply them. Both accept an `$encoding` and `$options` parameter in their constructor.
## Examples
```php
$inputStream = new ResourceInputStream(STDIN);
$gzInputStream = new ZlibInputStream($inputStream, \ZLIB_ENCODING_GZIP);
while (null !== $chunk = yield $gzInputStream) {
print $chunk;
}
```
```php
$outputStream = new ResourceOutputStream(STDOUT);
$gzOutputStream = new ZlibOutputStream($outputStream, \ZLIB_ENCODING_GZIP);
for ($i = 0; $i < 100; $i++) {
yield $gzOutputStream->write(bin2hex(random_bytes(32));
}
yield $gzOutputStream->end();
```
## See also
* [`./examples/gzip-compress.php`](https://github.com/amphp/byte-stream/blob/master/examples/gzip-compress.php)
* [`./examples/gzip-decompress.php`](https://github.com/amphp/byte-stream/blob/master/examples/gzip-decompress.php)
docs/in-memory-stream.md 0000644 00000000733 14476013052 0011230 0 ustar 00 ---
title: InMemoryStream
permalink: /in-memory-stream
---
An `InMemoryStream` allows creating an `InputStream` from a single known string chunk. This is helpful if the complete stream contents are already known.
```php
$inputStream = new InMemoryStream("foobar");
```
It also allows creating a stream without any chunks by passing `null` as chunk.
```php
$inputStream = new InMemoryStream;
// The stream ends immediately
assert(null === yield $inputStream->read());
```
docs/index.md 0000644 00000005444 14476013052 0007136 0 ustar 00 ---
title: Overview
permalink: /
---
Streams are an abstraction over ordered sequences of bytes. This package provides the fundamental interfaces `InputStream` and `OutputStream`.
## InputStream
`InputStream` offers a single method: `read()`. It returns a promise that gets either resolved with a `string` or `null`. `null` indicates that the stream has ended.
### Example
This example shows a simple `InputStream` consumption that buffers the complete stream contents inside a coroutine.
```php
$inputStream = ...;
$buffer = "";
while (($chunk = yield $inputStream->read()) !== null) {
$buffer .= $chunk;
}
// do something with $buffer
```
{:.note}
> While buffering a stream that way is relatively straightforward, you might want to use `yield new Message($inputStream)` to buffer a complete `InputStream`, making it even easier.
### Implementations
This package offers some basic implementations, other libraries might provide even more implementations, such as [`amphp/socket`](https://github.com/amphp/socket).
* [`InMemoryStream`](./in-memory-stream.md)
* [`IteratorStream`](./iterator-stream.md)
* [`Message`](./message.md)
* [`Payload`](./payload.md)
* [`ResourceInputStream`](./resource-streams.md)
* [`ZlibInputStream`](./compression-streams.md)
## OutputStream
`OutputStream` offers two methods: `write()` and `end()`.
### `write()`
`write()` writes the given string to the stream. The returned `Promise` might be used to wait for completion. Waiting for completion allows writing only as fast as the underlying stream can write and potentially send over a network. TCP streams will resolve the returned `Promise` immediately as long as the write buffer isn't full.
The write order is always ensured, even if the writer doesn't wait on the promise.
{:.note}
> Use `Amp\Promise\rethrow` on the returned `Promise` if you do not wait on it to get notified about write errors instead of silently doing nothing on errors.
### `end()`
`end()` marks the stream as ended, optionally writing a last data chunk before. TCP streams might close the underlying stream for writing, but MUST NOT close it. Instead, all resources should be freed and actual resource handles be closed by PHP's garbage collection process.
## Example
This example uses the previous example to read from a stream and simply writes all data to an `OutputStream`.
```php
$inputStream = ...;
$outputStream = ...;
$buffer = "";
while (($chunk = yield $inputStream->read()) !== null) {
yield $outputStream->write($chunk);
}
yield $outputStream->end();
```
### Implementations
This package offers some basic implementations, other libraries might provide even more implementations, such as [`amphp/socket`](https://github.com/amphp/socket).
* [`ResourceOutputStream`](./resource-streams.md)
* [`ZlibOutputStream`](./compression-streams.md)
docs/iterator-stream.md 0000644 00000000530 14476013052 0011140 0 ustar 00 ---
title: IteratorStream
permalink: /iterator-stream
---
`IteratorStream` allows converting an `Amp\Iterator` that yields strings into an `InputStream`.
```php
$inputStream = new IteratorStream(new Producer(function (callable $emit) {
for ($i = 0; $i < 10; $i++) {
yield new Delayed(1000);
yield $emit(".");
}
});
```
docs/message.md 0000644 00000003004 14476013052 0007441 0 ustar 00 ---
title: Message
permalink: /message
---
{:.note}
> `Message` has been deprecated. Use [`Payload`](./payload.md) instead.
`Message` implements both `InputStream` _and_ `Promise`. This allows consuming a message either in chunks (streaming) or consume everything at once (buffering).
## Buffering
Buffering a complete input stream is quite easy, you can simply `yield` a `Message` just like any other `Promise`. If you have an `InputStream` that's not a `Message`, simply create a new instance from it using `new Message($inputStream)`.
```php
$message = new Message($inputStream);
$content = yield $message;
```
## Streaming
Sometimes it's useful / possible to consume a message in chunks rather than first buffering it completely. An example might be streaming a large HTTP response body directly to disk.
```php
while (($chunk = yield $message->read()) !== null) {
// Use $chunk here, works just like any other InputStream
}
```
## Unwrapping
In some cases you might want to resolve a promise with an `InputStream` or your method explicitly declares `InputStream` as a return type. In these cases you should use `Message::getInputStream` to return the raw input stream. This makes it possible to resolve promises with the `InputStream` and not run into unexpected issues. Only return a `Message` if you declare that as a type, otherwise an API assumes `InputStream` and might try to resolve a promise with that, resulting in buffering the message's content instead of resolving the promise with an `InputStream` instance.
docs/payload.md 0000644 00000002235 14476013052 0007453 0 ustar 00 ---
title: Payload
permalink: /payload
---
`Payload` implements `InputStream` while also providing a method `buffer()` for buffering the entire contents. This allows consuming a message either in chunks (streaming) or consume everything at once (buffering). When the object is destructed, any remaining data in the stream is automatically consumed and discarded. This class is useful for small payloads or when the entire contents of a stream is needed before any processing can be done.
## Buffering
Buffering a complete input stream is quite easy, you can simply `yield` the promise returned from `buffer()` just like any other `Promise`. If you have an `InputStream` that's not a `Payload`, simply create a new instance from it using `new Payload($inputStream)`.
```php
$payload = new Payload($inputStream);
$content = yield $payload->buffer();
```
## Streaming
Sometimes it's useful / possible to consume a payload in chunks rather than first buffering it completely. An example might be streaming a large HTTP response body directly to disk.
```php
while (($chunk = yield $payload->read()) !== null) {
// Use $chunk here, works just like any other InputStream
}
```
docs/resource-streams.md 0000644 00000001007 14476013052 0011321 0 ustar 00 ---
title: Resource Streams
permalink: /resource-streams
---
This package abstracts PHP's stream resources with `ResourceInputStream` and `ResourceOutputStream`. They automatically set the passed resource to non-blocking and allow reading and writing like any other `InputStream` / `OutputStream`. They also handle backpressure automatically by disabling the read watcher in case there's no read request and only activate a write watcher if the underlying write buffer is already full, which makes them very efficient.
examples/benchmark-throughput.php 0000644 00000004001 14476013052 0013231 0 ustar 00 write('NOTICE: The "xdebug" extension is loaded, this has a major impact on performance.' . PHP_EOL);
}
try {
if (!@\assert(false)) {
$stderr->write("NOTICE: Assertions are enabled, this has a major impact on performance." . PHP_EOL);
}
} catch (AssertionError $exception) {
$stderr->write("NOTICE: Assertions are enabled, this has a major impact on performance." . PHP_EOL);
}
$stderr->write('piping from ' . $if . ' to ' . $of . ' (for max ' . $t . ' second(s)) ...'. PHP_EOL);
Loop::delay($t * 1000, [$in, "close"]);
Loop::run(function () use ($stderr, $in, $out) {
$start = \microtime(true);
while (($chunk = yield $in->read()) !== null) {
yield $out->write($chunk);
}
$t = \microtime(true) - $start;
$bytes = \ftell($out->getResource());
$stderr->write('read ' . $bytes . ' byte(s) in ' . \round($t, 3) . ' second(s) => ' . \round($bytes / 1024 / 1024 / $t, 1) . ' MiB/s' . PHP_EOL);
$stderr->write('peak memory usage of ' . \round(\memory_get_peak_usage(true) / 1024 / 1024, 1) . ' MiB' . PHP_EOL);
});
examples/gzip-compress.php 0000644 00000000727 14476013052 0011705 0 ustar 00 read()) !== null) {
yield $gzout->write($chunk);
}
});
examples/gzip-decompress.php 0000644 00000000723 14476013052 0012212 0 ustar 00 read()) !== null) {
yield $stdout->write($chunk);
}
});
infection.json.dist 0000644 00000000225 14476013052 0010360 0 ustar 00 {
"timeout": 5,
"source": {
"directories": [
"lib"
]
},
"logs": {
"text": "infection.log"
}
} lib/ClosedException.php 0000644 00000000132 14476013052 0011111 0 ustar 00 contents = $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.php 0000644 00000001360 14476013052 0010300 0 ustar 00 read()) !== 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.
*
* @throws PendingReadError Thrown if another read operation is still pending.
*/
public function read(): Promise;
}
lib/InputStreamChain.php 0000644 00000002040 14476013052 0011237 0 ustar 00 streams = $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.php 0000644 00000003037 14476013052 0010775 0 ustar 00 iterator = $iterator;
}
/** @inheritdoc */
public function read(): Promise
{
if ($this->exception) {
return new Failure($this->exception);
}
if ($this->pending) {
throw new PendingReadError;
}
$this->pending = true;
$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.php 0000644 00000002654 14476013052 0010046 0 ustar 00 source = $inputStream;
}
/**
* @return Promise
*/
public function readLine(): Promise
{
return call(function () {
if (($pos = \strpos($this->buffer, "\n")) !== false) {
$line = \substr($this->buffer, 0, $pos);
$this->buffer = \substr($this->buffer, $pos + 1);
return \rtrim($line, "\r");
}
while (null !== $chunk = yield $this->source->read()) {
$this->buffer .= $chunk;
if (($pos = \strpos($this->buffer, "\n")) !== false) {
$line = \substr($this->buffer, 0, $pos);
$this->buffer = \substr($this->buffer, $pos + 1);
return \rtrim($line, "\r");
}
}
if ($this->buffer === "") {
return null;
}
$line = $this->buffer;
$this->buffer = "";
return \rtrim($line, "\r");
});
}
public function getBuffer(): string
{
return $this->buffer;
}
public function clearBuffer()
{
$this->buffer = "";
}
}
lib/Message.php 0000644 00000011304 14476013052 0007410 0 ustar 00 read()) !== 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 \Amp\Deferred|null */
private $pendingRead;
/** @var \Amp\Coroutine */
private $coroutine;
/** @var bool True if onResolve() has been called. */
private $buffering = false;
/** @var \Amp\Deferred|null */
private $backpressure;
/** @var bool True if the iterator has completed. */
private $complete = false;
/** @var \Throwable 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.php 0000644 00000002162 14476013052 0010460 0 ustar 00 deferred = 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.php 0000644 00000002434 14476013052 0010504 0 ustar 00 stream = $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.php 0000644 00000000712 14476013052 0011217 0 ustar 00 useSingleRead = $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.
}
// 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 {
@\fclose($this->resource);
}
}
$this->free();
}
/**
* Nulls reference to resource, marks stream unreadable, and succeeds any pending read with null.
*/
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;
}
public function setChunkSize(int $chunkSize)
{
$this->chunkSize = $chunkSize;
}
/**
* References the read watcher, so the loop keeps running in case there's an active read.
*
* @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.
*
* @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.php 0000644 00000024230 14476013052 0012212 0 ustar 00 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 \Amp\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;
$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 {
@\fclose($this->resource);
}
}
$this->free();
}
/**
* Nulls reference to resource, marks stream unwritable, and fails any pending write.
*/
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 \Amp\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;
}
public function setChunkSize(int $chunkSize)
{
$this->chunkSize = $chunkSize;
}
public function __destruct()
{
if ($this->resource !== null) {
$this->free();
}
}
}
lib/StreamException.php 0000644 00000000117 14476013052 0011136 0 ustar 00 source = $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;
}
$data = yield $this->source->read();
// Needs a double guard, as stream might have been closed while reading
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 */
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.php 0000644 00000005534 14476013052 0011331 0 ustar 00 destination = $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");
}
$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");
}
$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 */
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.php 0000644 00000011034 14476013052 0010034 0 ustar 00 read()) !== 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);
}
});
}
middle 0000644 00000000000 14476013052 0005715 0 ustar 00 docs/.shared 0000777 00000000000 14476013052 0006747 5 ustar 00