danog/byte-stream
1
0
Fork 0
mirror of https://github.com/danog/byte-stream.git synced 2024-11-26 11:54:54 +01:00
Code Issues Projects Releases Wiki Activity

Apply PSR-2

Browse Source
This commit is contained in:
Niklas Keller 2018-09-21 22:45:13 +02:00
parent 73257690b0
commit d161589772
29 changed files with 308 additions and 196 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

46
.php_cs.dist
View File

@ -1,40 +1,10 @@
<?php
return PhpCsFixer\Config::create()
->setRiskyAllowed(true)
->setRules([
"@PSR1" => true,
"@PSR2" => true,
"braces" => [
"allow_single_line_closure" => true,
"position_after_functions_and_oop_constructs" => "same",
],
"array_syntax" => ["syntax" => "short"],
"cast_spaces" => true,
"combine_consecutive_unsets" => true,
"function_to_constant" => true,
"no_multiline_whitespace_before_semicolons" => true,
"no_unused_imports" => true,
"no_useless_else" => true,
"no_useless_return" => true,
"no_whitespace_before_comma_in_array" => true,
"no_whitespace_in_blank_line" => true,
"non_printable_character" => true,
"normalize_index_brace" => true,
"ordered_imports" => true,
"php_unit_construct" => true,
"php_unit_dedicate_assert" => true,
"php_unit_fqcn_annotation" => true,
"phpdoc_summary" => true,
"phpdoc_types" => true,
"psr4" => true,
"return_type_declaration" => ["space_before" => "none"],
"short_scalar_cast" => true,
"single_blank_line_before_namespace" => true,
])
->setFinder(
PhpCsFixer\Finder::create()
->in(__DIR__ . "/examples")
->in(__DIR__ . "/lib")
->in(__DIR__ . "/test")
);
$config = new Amp\CodeStyle\Config();
$config->getFinder()->in(__DIR__);
$cacheDir = getenv('TRAVIS') ? getenv('HOME') . '/.php-cs-fixer' : __DIR__;
$config->setCacheFile($cacheDir . '/.php_cs.cache');
return $config;

3
composer.json
View File

@ -31,7 +31,8 @@
"require-dev": {
"amphp/phpunit-util": "^1",
"phpunit/phpunit": "^6",
"friendsofphp/php-cs-fixer": "^2.3"
"friendsofphp/php-cs-fixer": "^2.3",
"amphp/php-cs-fixer-config": "dev-master"
},
"autoload": {
"psr-4": {

22
examples/benchmark-throughput.php
View File

@ -11,20 +11,20 @@ require __DIR__ . '/../vendor/autoload.php';
Loop::set(new Loop\NativeDriver());
$args = getopt('i:o:t:');
$args = \getopt('i:o:t:');
$if = isset($args['i']) ? $args['i'] : '/dev/zero';
$of = isset($args['o']) ? $args['o'] : '/dev/null';
$t = isset($args['t']) ? $args['t'] : 30;
// passing file descriptors requires mapping paths (https://bugs.php.net/bug.php?id=53465)
$if = preg_replace('(^/dev/fd/)', 'php://fd/', $if);
$of = preg_replace('(^/dev/fd/)', 'php://fd/', $of);
$if = \preg_replace('(^/dev/fd/)', 'php://fd/', $if);
$of = \preg_replace('(^/dev/fd/)', 'php://fd/', $of);
$stderr = new ResourceOutputStream(STDERR);
$in = new ResourceInputStream(fopen($if, 'r'), 65536 /* Default size used by React to allow comparisons */);
$out = new ResourceOutputStream(fopen($of, 'w'));
$in = new ResourceInputStream(\fopen($if, 'r'), 65536 /* Default size used by React to allow comparisons */);
$out = new ResourceOutputStream(\fopen($of, 'w'));
if (extension_loaded('xdebug')) {
if (\extension_loaded('xdebug')) {
$stderr->write('NOTICE: The "xdebug" extension is loaded, this has a major impact on performance.' . PHP_EOL);
}
@ -41,16 +41,16 @@ $stderr->write('piping from ' . $if . ' to ' . $of . ' (for max ' . $t . ' secon
Loop::delay($t * 1000, [$in, "close"]);
Loop::run(function () use ($stderr, $in, $out) {
$start = microtime(true);
$start = \microtime(true);
while (($chunk = yield $in->read()) !== null) {
yield $out->write($chunk);
}
$t = microtime(true) - $start;
$t = \microtime(true) - $start;
$bytes = ftell($out->getResource());
$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);
$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);
});

3
lib/ClosedException.php
View File

@ -2,5 +2,6 @@
namespace Amp\ByteStream;
final class ClosedException extends StreamException {
final class ClosedException extends StreamException
{
}

9
lib/InMemoryStream.php
View File

@ -8,13 +8,15 @@ use Amp\Success;
/**
* Input stream with a single already known data chunk.
*/
final class InMemoryStream implements InputStream {
final class InMemoryStream implements InputStream
{
private $contents;
/**
* @param string|null $contents Data chunk or `null` for no data chunk.
*/
public function __construct(string $contents = null) {
public function __construct(string $contents = null)
{
$this->contents = $contents;
}
@ -23,7 +25,8 @@ final class InMemoryStream implements InputStream {
*
* @return Promise Resolves with the full contents or `null` if the stream has closed / already been consumed.
*/
public function read(): Promise {
public function read(): Promise
{
if ($this->contents === null) {
return new Success;
}

3
lib/InputStream.php
View File

@ -23,7 +23,8 @@ use Amp\Promise;
* }
* ```
*/
interface InputStream {
interface InputStream
{
/**
* Reads data from the stream.
*

9
lib/IteratorStream.php
View File

@ -7,17 +7,20 @@ use Amp\Failure;
use Amp\Iterator;
use Amp\Promise;
final class IteratorStream implements InputStream {
final class IteratorStream implements InputStream
{
private $iterator;
private $exception;
private $pending = false;
public function __construct(Iterator $iterator) {
public function __construct(Iterator $iterator)
{
$this->iterator = $iterator;
}
/** @inheritdoc */
public function read(): Promise {
public function read(): Promise
{
if ($this->exception) {
return new Failure($this->exception);
}

18
lib/Message.php
View File

@ -30,7 +30,8 @@ use Amp\Success;
*
* @deprecated Use Amp\ByteStream\Payload instead.
*/
class Message implements InputStream, Promise {
class Message implements InputStream, Promise
{
/** @var InputStream */
private $source;
@ -58,11 +59,13 @@ class Message implements InputStream, Promise {
/**
* @param InputStream $source An iterator that only emits strings.
*/
public function __construct(InputStream $source) {
public function __construct(InputStream $source)
{
$this->source = $source;
}
private function consume(): \Generator {
private function consume(): \Generator
{
while (($chunk = yield $this->source->read()) !== null) {
$buffer = $this->buffer .= $chunk;
@ -94,7 +97,8 @@ class Message implements InputStream, Promise {
}
/** @inheritdoc */
final public function read(): Promise {
final public function read(): Promise
{
if ($this->pendingRead) {
throw new PendingReadError;
}
@ -140,7 +144,8 @@ class Message implements InputStream, Promise {
}
/** @inheritdoc */
final public function onResolve(callable $onResolved) {
final public function onResolve(callable $onResolved)
{
$this->buffering = true;
if ($this->coroutine === null) {
@ -164,7 +169,8 @@ class Message implements InputStream, Promise {
*
* @return InputStream
*/
final public function getInputStream(): InputStream {
final public function getInputStream(): InputStream
{
return $this->source;
}
}

15
lib/OutputBuffer.php
View File

@ -6,7 +6,8 @@ use Amp\Deferred;
use Amp\Promise;
use Amp\Success;
class OutputBuffer implements OutputStream, Promise {
class OutputBuffer implements OutputStream, Promise
{
/** @var \Amp\Deferred|null */
private $deferred;
@ -15,11 +16,13 @@ class OutputBuffer implements OutputStream, Promise {
private $closed = false;
public function __construct() {
public function __construct()
{
$this->deferred = new Deferred;
}
public function write(string $data): Promise {
public function write(string $data): Promise
{
if ($this->closed) {
throw new ClosedException("The stream has already been closed.");
}
@ -29,7 +32,8 @@ class OutputBuffer implements OutputStream, Promise {
return new Success(\strlen($data));
}
public function end(string $finalData = ""): Promise {
public function end(string $finalData = ""): Promise
{
if ($this->closed) {
throw new ClosedException("The stream has already been closed.");
}
@ -43,7 +47,8 @@ class OutputBuffer implements OutputStream, Promise {
return new Success(\strlen($finalData));
}
public function onResolve(callable $onResolved) {
public function onResolve(callable $onResolved)
{
$this->deferred->promise()->onResolve($onResolved);
}
}

3
lib/OutputStream.php
View File

@ -7,7 +7,8 @@ use Amp\Promise;
/**
* An `OutputStream` allows writing data in chunks. Writers can wait on the returned promises to feel the backpressure.
*/
interface OutputStream {
interface OutputStream
{
/**
* Writes data to the stream.
*

18
lib/Payload.php
View File

@ -11,7 +11,8 @@ use function Amp\call;
* be buffered and accessed in its entirety by calling buffer(). Once buffering is requested through buffer(), the
* stream cannot be read in chunks. On destruct any remaining data is read from the InputStream given to this class.
*/
class Payload implements InputStream {
class Payload implements InputStream
{
/** @var InputStream */
private $stream;
@ -24,17 +25,20 @@ class Payload implements InputStream {
/**
* @param \Amp\ByteStream\InputStream $stream
*/
public function __construct(InputStream $stream) {
public function __construct(InputStream $stream)
{
$this->stream = $stream;
}
public function __destruct() {
public function __destruct()
{
if (!$this->promise) {
Promise\rethrow(new Coroutine($this->consume()));
}
}
private function consume(): \Generator {
private function consume(): \Generator
{
try {
if ($this->lastRead && null === yield $this->lastRead) {
return;
@ -53,7 +57,8 @@ class Payload implements InputStream {
*
* @throws \Error If a buffered message was requested by calling buffer().
*/
final public function read(): Promise {
final public function read(): Promise
{
if ($this->promise) {
throw new \Error("Cannot stream message data once a buffered message has been requested");
}
@ -66,7 +71,8 @@ class Payload implements InputStream {
*
* @return Promise<string> Resolves with the entire message contents.
*/
final public function buffer(): Promise {
final public function buffer(): Promise
{
if ($this->promise) {
return $this->promise;
}

3
lib/PendingReadError.php
View File

@ -5,7 +5,8 @@ namespace Amp\ByteStream;
/**
* Thrown in case a second read operation is attempted while another read operation is still pending.
*/
final class PendingReadError extends \Error {
final class PendingReadError extends \Error
{
public function __construct(
string $message = "The previous read operation must complete before read can be called again",
int $code = 0,

27
lib/ResourceInputStream.php
View File

@ -10,7 +10,8 @@ use Amp\Success;
/**
* Input stream abstraction for PHP's stream resources.
*/
final class ResourceInputStream implements InputStream {
final class ResourceInputStream implements InputStream
{
const DEFAULT_CHUNK_SIZE = 8192;
/** @var resource */
@ -43,7 +44,8 @@ final class ResourceInputStream implements InputStream {
*
* @throws \Error If an invalid stream or parameter has been passed.
*/
public function __construct($stream, int $chunkSize = self::DEFAULT_CHUNK_SIZE) {
public function __construct($stream, int $chunkSize = self::DEFAULT_CHUNK_SIZE)
{
if (!\is_resource($stream) || \get_resource_type($stream) !== 'stream') {
throw new \Error("Expected a valid stream");
}
@ -102,7 +104,8 @@ final class ResourceInputStream implements InputStream {
}
/** @inheritdoc */
public function read(): Promise {
public function read(): Promise
{
if ($this->deferred !== null) {
throw new PendingReadError;
}
@ -149,7 +152,8 @@ final class ResourceInputStream implements InputStream {
*
* @return void
*/
public function close() {
public function close()
{
if ($this->resource) {
// Error suppression, as resource might already be closed
$meta = @\stream_get_meta_data($this->resource);
@ -168,7 +172,8 @@ final class ResourceInputStream implements InputStream {
/**
* Nulls reference to resource, marks stream unreadable, and succeeds any pending read with null.
*/
private function free() {
private function free()
{
$this->readable = false;
if ($this->deferred !== null) {
@ -187,7 +192,8 @@ final class ResourceInputStream implements InputStream {
/**
* @return resource|null The stream resource or null if the stream has closed.
*/
public function getResource() {
public function getResource()
{
return $this->resource;
}
@ -196,7 +202,8 @@ final class ResourceInputStream implements InputStream {
*
* @see Loop::reference()
*/
public function reference() {
public function reference()
{
if (!$this->resource) {
throw new \Error("Resource has already been freed");
}
@ -209,7 +216,8 @@ final class ResourceInputStream implements InputStream {
*
* @see Loop::unreference()
*/
public function unreference() {
public function unreference()
{
if (!$this->resource) {
throw new \Error("Resource has already been freed");
}
@ -217,7 +225,8 @@ final class ResourceInputStream implements InputStream {
Loop::unreference($this->watcher);
}
public function __destruct() {
public function __destruct()
{
if ($this->resource !== null) {
$this->free();
}

27
lib/ResourceOutputStream.php
View File

@ -11,7 +11,8 @@ use Amp\Success;
/**
* Output stream abstraction for PHP's stream resources.
*/
final class ResourceOutputStream implements OutputStream {
final class ResourceOutputStream implements OutputStream
{
const MAX_CONSECUTIVE_EMPTY_WRITES = 3;
/** @var resource */
@ -33,7 +34,8 @@ final class ResourceOutputStream implements OutputStream {
* @param resource $stream Stream resource.
* @param int|null $chunkSize Chunk size per `fwrite()` operation.
*/
public function __construct($stream, int $chunkSize = null) {
public function __construct($stream, int $chunkSize = null)
{
if (!\is_resource($stream) || \get_resource_type($stream) !== 'stream') {
throw new \Error("Expected a valid stream");
}
@ -136,7 +138,8 @@ final class ResourceOutputStream implements OutputStream {
*
* @throws ClosedException If the stream has already been closed.
*/
public function write(string $data): Promise {
public function write(string $data): Promise
{
return $this->send($data, false);
}
@ -149,11 +152,13 @@ final class ResourceOutputStream implements OutputStream {
*
* @throws ClosedException If the stream has already been closed.
*/
public function end(string $finalData = ""): Promise {
public function end(string $finalData = ""): Promise
{
return $this->send($finalData, true);
}
private function send(string $data, bool $end = false): Promise {
private function send(string $data, bool $end = false): Promise
{
if (!$this->writable) {
return new Failure(new ClosedException("The stream is not writable"));
}
@ -214,7 +219,8 @@ final class ResourceOutputStream implements OutputStream {
*
* @return void
*/
public function close() {
public function close()
{
if ($this->resource) {
// Error suppression, as resource might already be closed
$meta = @\stream_get_meta_data($this->resource);
@ -232,7 +238,8 @@ final class ResourceOutputStream implements OutputStream {
/**
* Nulls reference to resource, marks stream unwritable, and fails any pending write.
*/
private function free() {
private function free()
{
$this->resource = null;
$this->writable = false;
@ -251,11 +258,13 @@ final class ResourceOutputStream implements OutputStream {
/**
* @return resource|null Stream resource or null if end() has been called or the stream closed.
*/
public function getResource() {
public function getResource()
{
return $this->resource;
}
public function __destruct() {
public function __destruct()
{
if ($this->resource !== null) {
$this->free();
}

3
lib/StreamException.php
View File

@ -2,5 +2,6 @@
namespace Amp\ByteStream;
class StreamException extends \Exception {
class StreamException extends \Exception
{
}

18
lib/ZlibInputStream.php
View File

@ -8,7 +8,8 @@ use function Amp\call;
/**
* Allows decompression of input streams using Zlib.
*/
final class ZlibInputStream implements InputStream {
final class ZlibInputStream implements InputStream
{
private $source;
private $encoding;
private $options;
@ -24,7 +25,8 @@ final class ZlibInputStream implements InputStream {
*
* @see http://php.net/manual/en/function.inflate-init.php
*/
public function __construct(InputStream $source, int $encoding, array $options = []) {
public function __construct(InputStream $source, int $encoding, array $options = [])
{
$this->source = $source;
$this->encoding = $encoding;
$this->options = $options;
@ -36,7 +38,8 @@ final class ZlibInputStream implements InputStream {
}
/** @inheritdoc */
public function read(): Promise {
public function read(): Promise
{
return call(function () {
if ($this->resource === null) {
return null;
@ -72,7 +75,8 @@ final class ZlibInputStream implements InputStream {
}
/** @internal */
private function close() {
private function close()
{
$this->resource = null;
$this->source = null;
}
@ -82,7 +86,8 @@ final class ZlibInputStream implements InputStream {
*
* @return int Encoding specified on construction time.
*/
public function getEncoding(): int {
public function getEncoding(): int
{
return $this->encoding;
}
/**
@ -90,7 +95,8 @@ final class ZlibInputStream implements InputStream {
*
* @return array Options array passed on construction time.
*/
public function getOptions(): array {
public function getOptions(): array
{
return $this->options;
}
}

21
lib/ZlibOutputStream.php
View File

@ -7,7 +7,8 @@ use Amp\Promise;
/**
* Allows compression of output streams using Zlib.
*/
final class ZlibOutputStream implements OutputStream {
final class ZlibOutputStream implements OutputStream
{
private $destination;
private $encoding;
private $options;
@ -22,7 +23,8 @@ final class ZlibOutputStream implements OutputStream {
*
* @see http://php.net/manual/en/function.deflate-init.php
*/
public function __construct(OutputStream $destination, int $encoding, array $options = []) {
public function __construct(OutputStream $destination, int $encoding, array $options = [])
{
$this->destination = $destination;
$this->encoding = $encoding;
$this->options = $options;
@ -34,7 +36,8 @@ final class ZlibOutputStream implements OutputStream {
}
/** @inheritdoc */
public function write(string $data): Promise {
public function write(string $data): Promise
{
if ($this->resource === null) {
throw new ClosedException("The stream has already been closed");
}
@ -56,7 +59,8 @@ final class ZlibOutputStream implements OutputStream {
}
/** @inheritdoc */
public function end(string $finalData = ""): Promise {
public function end(string $finalData = ""): Promise
{
if ($this->resource === null) {
throw new ClosedException("The stream has already been closed");
}
@ -76,7 +80,8 @@ final class ZlibOutputStream implements OutputStream {
}
/** @internal */
private function close() {
private function close()
{
$this->resource = null;
$this->destination = null;
}
@ -86,7 +91,8 @@ final class ZlibOutputStream implements OutputStream {
*
* @return int Encoding specified on construction time.
*/
public function getEncoding(): int {
public function getEncoding(): int
{
return $this->encoding;
}
@ -95,7 +101,8 @@ final class ZlibOutputStream implements OutputStream {
*
* @return array Options array passed on construction time.
*/
public function getOptions(): array {
public function getOptions(): array
{
return $this->options;
}
}

6
lib/functions.php
View File

@ -26,7 +26,8 @@ if (!\defined('STDERR')) {
*
* @return \Amp\Promise
*/
function pipe(InputStream $source, OutputStream $destination): Promise {
function pipe(InputStream $source, OutputStream $destination): Promise
{
return call(function () use ($source, $destination): \Generator {
$written = 0;
@ -46,7 +47,8 @@ function pipe(InputStream $source, OutputStream $destination): Promise {
*
* @return \Amp\Promise
*/
function buffer(InputStream $source): Promise {
function buffer(InputStream $source): Promise
{
return call(function () use ($source): \Generator {
$buffer = "";

1
test/BufferTest.php
View File

@ -1,6 +1,5 @@
<?php
namespace Amp\ByteStream\Test;
use Amp\ByteStream\IteratorStream;

6
test/InMemoryStreamTest.php
View File

@ -6,8 +6,10 @@ use Amp\ByteStream\InMemoryStream;
use Amp\Loop;
use PHPUnit\Framework\TestCase;
class InMemoryStreamTest extends TestCase {
public function testSingleReadConsumesEverything() {
class InMemoryStreamTest extends TestCase
{
public function testSingleReadConsumesEverything()
{
Loop::run(function () {
$stream = new InMemoryStream("foobar");
$this->assertSame("foobar", yield $stream->read());

15
test/IteratorStreamTest.php
View File

@ -9,8 +9,10 @@ use Amp\Loop;
use Amp\PHPUnit\TestCase;
use Amp\PHPUnit\TestException;
class IteratorStreamTest extends TestCase {
public function testReadIterator() {
class IteratorStreamTest extends TestCase
{
public function testReadIterator()
{
Loop::run(function () {
$values = ["abc", "def", "ghi"];
@ -33,7 +35,8 @@ class IteratorStreamTest extends TestCase {
});
}
public function testFailingIterator() {
public function testFailingIterator()
{
Loop::run(function () {
$exception = new TestException;
$value = "abc";
@ -59,7 +62,8 @@ class IteratorStreamTest extends TestCase {
});
}
public function testThrowsOnNonStringIteration() {
public function testThrowsOnNonStringIteration()
{
$this->expectException(StreamException::class);
Loop::run(function () {
$value = 42;
@ -73,7 +77,8 @@ class IteratorStreamTest extends TestCase {
});
}
public function testFailsAfterException() {
public function testFailsAfterException()
{
$this->expectException(StreamException::class);
Loop::run(function () {
$value = 42;

42
test/MessageTest.php
View File

@ -11,8 +11,10 @@ use Amp\Loop;
use Amp\PHPUnit\TestCase;
use Amp\PHPUnit\TestException;
class MessageTest extends TestCase {
public function testBufferingAll() {
class MessageTest extends TestCase
{
public function testBufferingAll()
{
Loop::run(function () {
$values = ["abc", "def", "ghi"];
@ -29,7 +31,8 @@ class MessageTest extends TestCase {
});
}
public function testFullStreamConsumption() {
public function testFullStreamConsumption()
{
Loop::run(function () use (&$invoked) {
$values = ["abc", "def", "ghi"];
@ -54,7 +57,8 @@ class MessageTest extends TestCase {
});
}
public function testFastResolvingStream() {
public function testFastResolvingStream()
{
Loop::run(function () {
$values = ["abc", "def", "ghi"];
@ -77,7 +81,8 @@ class MessageTest extends TestCase {
});
}
public function testFastResolvingStreamBufferingOnly() {
public function testFastResolvingStreamBufferingOnly()
{
Loop::run(function () {
$values = ["abc", "def", "ghi"];
@ -94,7 +99,8 @@ class MessageTest extends TestCase {
});
}
public function testPartialStreamConsumption() {
public function testPartialStreamConsumption()
{
Loop::run(function () {
$values = ["abc", "def", "ghi"];
@ -117,7 +123,8 @@ class MessageTest extends TestCase {
});
}
public function testFailingStream() {
public function testFailingStream()
{
Loop::run(function () {
$exception = new TestException;
$value = "abc";
@ -143,7 +150,8 @@ class MessageTest extends TestCase {
});
}
public function testFailingStreamWithPendingRead() {
public function testFailingStreamWithPendingRead()
{
Loop::run(function () {
$exception = new TestException;
$value = "abc";
@ -167,7 +175,8 @@ class MessageTest extends TestCase {
});
}
public function testEmptyStream() {
public function testEmptyStream()
{
Loop::run(function () {
$emitter = new Emitter;
$emitter->complete();
@ -177,7 +186,8 @@ class MessageTest extends TestCase {
});
}
public function testEmptyStringStream() {
public function testEmptyStringStream()
{
Loop::run(function () {
$value = "";
@ -192,7 +202,8 @@ class MessageTest extends TestCase {
});
}
public function testReadAfterCompletion() {
public function testReadAfterCompletion()
{
Loop::run(function () {
$value = "abc";
@ -207,7 +218,8 @@ class MessageTest extends TestCase {
});
}
public function testGetInputStream() {
public function testGetInputStream()
{
Loop::run(function () {
$inputStream = new InMemoryStream("");
$message = new Message($inputStream);
@ -217,7 +229,8 @@ class MessageTest extends TestCase {
});
}
public function testPendingRead() {
public function testPendingRead()
{
Loop::run(function () {
$emitter = new Emitter;
$stream = new Message(new IteratorStream($emitter->iterate()));
@ -230,7 +243,8 @@ class MessageTest extends TestCase {
});
}
public function testPendingReadError() {
public function testPendingReadError()
{
Loop::run(function () {
$emitter = new Emitter;
$stream = new Message(new IteratorStream($emitter->iterate()));

15
test/OutputBufferTest.php
View File

@ -7,8 +7,10 @@ use Amp\ByteStream\OutputBuffer;
use Amp\Loop;
use Amp\PHPUnit\TestCase;
class OutputBufferTest extends TestCase {
public function testWrite() {
class OutputBufferTest extends TestCase
{
public function testWrite()
{
Loop::run(function () {
$output = new OutputBuffer();
$output->write('foo');
@ -18,7 +20,8 @@ class OutputBufferTest extends TestCase {
});
}
public function testEnd() {
public function testEnd()
{
Loop::run(function () {
$output = new OutputBuffer();
$output->write('foo');
@ -28,7 +31,8 @@ class OutputBufferTest extends TestCase {
});
}
public function testThrowsOnWritingToClosedBuffer() {
public function testThrowsOnWritingToClosedBuffer()
{
$this->expectException(ClosedException::class);
Loop::run(function () {
@ -38,7 +42,8 @@ class OutputBufferTest extends TestCase {
});
}
public function testThrowsOnEndingToClosedBuffer() {
public function testThrowsOnEndingToClosedBuffer()
{
$this->expectException(ClosedException::class);
Loop::run(function () {

45
test/PayloadTest.php
View File

@ -11,8 +11,10 @@ use Amp\Loop;
use Amp\PHPUnit\TestCase;
use Amp\PHPUnit\TestException;
class PayloadTest extends TestCase {
public function testBufferingAll() {
class PayloadTest extends TestCase
{
public function testBufferingAll()
{
Loop::run(function () {
$values = ["abc", "def", "ghi"];
@ -29,7 +31,8 @@ class PayloadTest extends TestCase {
});
}
public function testFullStreamConsumption() {
public function testFullStreamConsumption()
{
Loop::run(function () use (&$invoked) {
$values = ["abc", "def", "ghi"];
@ -54,7 +57,8 @@ class PayloadTest extends TestCase {
});
}
public function testFastResolvingStream() {
public function testFastResolvingStream()
{
Loop::run(function () {
$values = ["abc", "def", "ghi"];
@ -77,7 +81,8 @@ class PayloadTest extends TestCase {
});
}
public function testFastResolvingStreamBufferingOnly() {
public function testFastResolvingStreamBufferingOnly()
{
Loop::run(function () {
$values = ["abc", "def", "ghi"];
@ -94,7 +99,8 @@ class PayloadTest extends TestCase {
});
}
public function testPartialStreamConsumption() {
public function testPartialStreamConsumption()
{
Loop::run(function () {
$values = ["abc", "def", "ghi"];
@ -117,7 +123,8 @@ class PayloadTest extends TestCase {
});
}
public function testFailingStream() {
public function testFailingStream()
{
Loop::run(function () {
$exception = new TestException;
$value = "abc";
@ -143,7 +150,8 @@ class PayloadTest extends TestCase {
});
}
public function testFailingStreamWithPendingRead() {
public function testFailingStreamWithPendingRead()
{
Loop::run(function () {
$exception = new TestException;
$value = "abc";
@ -167,7 +175,8 @@ class PayloadTest extends TestCase {
});
}
public function testEmptyStream() {
public function testEmptyStream()
{
Loop::run(function () {
$emitter = new Emitter;
$emitter->complete();
@ -177,7 +186,8 @@ class PayloadTest extends TestCase {
});
}
public function testEmptyStringStream() {
public function testEmptyStringStream()
{
Loop::run(function () {
$value = "";
@ -192,7 +202,8 @@ class PayloadTest extends TestCase {
});
}
public function testReadAfterCompletion() {
public function testReadAfterCompletion()
{
Loop::run(function () {
$value = "abc";
@ -207,7 +218,8 @@ class PayloadTest extends TestCase {
});
}
public function testPendingRead() {
public function testPendingRead()
{
Loop::run(function () {
$emitter = new Emitter;
$stream = new Payload(new IteratorStream($emitter->iterate()));
@ -220,7 +232,8 @@ class PayloadTest extends TestCase {
});
}
public function testPendingReadError() {
public function testPendingReadError()
{
Loop::run(function () {
$emitter = new Emitter;
$stream = new Payload(new IteratorStream($emitter->iterate()));
@ -232,7 +245,8 @@ class PayloadTest extends TestCase {
});
}
public function testReadAfterBuffer() {
public function testReadAfterBuffer()
{
Loop::run(function () {
$stream = new Payload(new InMemoryStream("test"));
$stream->buffer();
@ -244,7 +258,8 @@ class PayloadTest extends TestCase {
});
}
public function testFurtherCallsToBufferReturnSameData() {
public function testFurtherCallsToBufferReturnSameData()
{
Loop::run(function () {
$data = "test";
$stream = new Payload(new InMemoryStream($data));

12
test/ResourceInputStreamTest.php
View File

@ -5,21 +5,25 @@ namespace Amp\ByteStream\Test;
use Amp\ByteStream\ResourceInputStream;
use PHPUnit\Framework\TestCase;
class ResourceInputStreamTest extends TestCase {
public function testGetResource() {
class ResourceInputStreamTest extends TestCase
{
public function testGetResource()
{
$stream = new ResourceInputStream(\STDIN);
$this->assertSame(\STDIN, $stream->getResource());
}
public function testNonStream() {
public function testNonStream()
{
$this->expectException(\Error::class);
$this->expectExceptionMessage("Expected a valid stream");
new ResourceInputStream(42);
}
public function testNotReadable() {
public function testNotReadable()
{
$this->expectException(\Error::class);
$this->expectExceptionMessage("Expected a readable stream");

21
test/ResourceOutputStreamTest.php
View File

@ -7,28 +7,33 @@ use Amp\ByteStream\StreamException;
use PHPUnit\Framework\TestCase;
use function Amp\Promise\wait;
class ResourceOutputStreamTest extends TestCase {
public function testGetResource() {
class ResourceOutputStreamTest extends TestCase
{
public function testGetResource()
{
$stream = new ResourceOutputStream(\STDOUT);
$this->assertSame(\STDOUT, $stream->getResource());
}
public function testNonStream() {
public function testNonStream()
{
$this->expectException(\Error::class);
$this->expectExceptionMessage("Expected a valid stream");
new ResourceOutputStream(42);
}
public function testNotWritable() {
public function testNotWritable()
{
$this->expectException(\Error::class);
$this->expectExceptionMessage("Expected a writable stream");
new ResourceOutputStream(\STDIN);
}
public function testBrokenPipe() {
public function testBrokenPipe()
{
if (($sockets = @\stream_socket_pair(\stripos(PHP_OS, "win") === 0 ? STREAM_PF_INET : STREAM_PF_UNIX, STREAM_SOCK_STREAM, STREAM_IPPROTO_IP)) === false) {
$this->fail("Failed to create socket pair.");
}
@ -43,7 +48,8 @@ class ResourceOutputStreamTest extends TestCase {
wait($stream->write("foobar"));
}
public function testClosedRemoteSocket() {
public function testClosedRemoteSocket()
{
$server = \stream_socket_server("tcp://127.0.0.1:0");
$address = \stream_socket_get_name($server, false);
@ -61,7 +67,8 @@ class ResourceOutputStreamTest extends TestCase {
wait($stream->write("foobar"));
}
public function testClosedRemoteSocketWithFork() {
public function testClosedRemoteSocketWithFork()
{
$server = \stream_socket_server("tcp://127.0.0.1:0");
$address = \stream_socket_get_name($server, false);

49
test/ResourceStreamTest.php
View File

@ -11,10 +11,12 @@ use Amp\Loop;
use Amp\Success;
use PHPUnit\Framework\TestCase;
class ResourceStreamTest extends TestCase {
class ResourceStreamTest extends TestCase
{
const LARGE_MESSAGE_SIZE = 1 << 20; // 1 MB
public function getStreamPair($outputChunkSize = null, $inputChunkSize = ResourceInputStream::DEFAULT_CHUNK_SIZE) {
public function getStreamPair($outputChunkSize = null, $inputChunkSize = ResourceInputStream::DEFAULT_CHUNK_SIZE)
{
$domain = \stripos(PHP_OS, "win") === 0 ? STREAM_PF_INET : STREAM_PF_UNIX;
list($left, $right) = @\stream_socket_pair($domain, \STREAM_SOCK_STREAM, \STREAM_IPPROTO_IP);
@ -24,7 +26,8 @@ class ResourceStreamTest extends TestCase {
return [$a, $b];
}
public function testLargePayloads() {
public function testLargePayloads()
{
Loop::run(function () {
list($a, $b) = $this->getStreamPair();
@ -41,7 +44,8 @@ class ResourceStreamTest extends TestCase {
});
}
public function testManySmallPayloads() {
public function testManySmallPayloads()
{
Loop::run(function () {
list($a, $b) = $this->getStreamPair();
@ -61,12 +65,13 @@ class ResourceStreamTest extends TestCase {
});
}
public function testThrowsOnExternallyShutdownStreamWithLargePayload() {
public function testThrowsOnExternallyShutdownStreamWithLargePayload()
{
$this->expectException(StreamException::class);
Loop::run(function () {
try { /* prevent crashes with phpdbg due to SIGPIPE not being handled... */
Loop::onSignal(defined("SIGPIPE") ? SIGPIPE : 13, function () {});
Loop::onSignal(\defined("SIGPIPE") ? SIGPIPE : 13, function () {});
} catch (Loop\UnsupportedFeatureException $e) {
}
@ -83,12 +88,13 @@ class ResourceStreamTest extends TestCase {
});
}
public function testThrowsOnExternallyShutdownStreamWithSmallPayloads() {
public function testThrowsOnExternallyShutdownStreamWithSmallPayloads()
{
$this->expectException(StreamException::class);
Loop::run(function () {
try { /* prevent crashes with phpdbg due to SIGPIPE not being handled... */
Loop::onSignal(defined("SIGPIPE") ? SIGPIPE : 13, function () {});
Loop::onSignal(\defined("SIGPIPE") ? SIGPIPE : 13, function () {});
} catch (Loop\UnsupportedFeatureException $e) {
}
@ -107,7 +113,8 @@ class ResourceStreamTest extends TestCase {
});
}
public function testThrowsOnCloseBeforeWritingComplete() {
public function testThrowsOnCloseBeforeWritingComplete()
{
$this->expectException(ClosedException::class);
Loop::run(function () {
@ -123,7 +130,8 @@ class ResourceStreamTest extends TestCase {
});
}
public function testThrowsOnStreamNotWritable() {
public function testThrowsOnStreamNotWritable()
{
$this->expectException(StreamException::class);
Loop::run(function () {
@ -139,7 +147,8 @@ class ResourceStreamTest extends TestCase {
});
}
public function testThrowsOnReferencingClosedStream() {
public function testThrowsOnReferencingClosedStream()
{
$this->expectException(\Error::class);
Loop::run(function () {
@ -151,7 +160,8 @@ class ResourceStreamTest extends TestCase {
});
}
public function testThrowsOnUnreferencingClosedStream() {
public function testThrowsOnUnreferencingClosedStream()
{
$this->expectException(\Error::class);
Loop::run(function () {
@ -163,7 +173,8 @@ class ResourceStreamTest extends TestCase {
});
}
public function testThrowsOnPendingRead() {
public function testThrowsOnPendingRead()
{
$this->expectException(PendingReadError::class);
Loop::run(function () {
@ -174,7 +185,8 @@ class ResourceStreamTest extends TestCase {
});
}
public function testResolveSuccessOnClosedStream() {
public function testResolveSuccessOnClosedStream()
{
Loop::run(function () {
list($a, $b) = $this->getStreamPair();
@ -184,7 +196,8 @@ class ResourceStreamTest extends TestCase {
});
}
public function testChunkedPayload() {
public function testChunkedPayload()
{
Loop::run(function () {
list($a, $b) = $this->getStreamPair(4096);
@ -201,7 +214,8 @@ class ResourceStreamTest extends TestCase {
});
}
public function testEmptyPayload() {
public function testEmptyPayload()
{
Loop::run(function () {
list($a, $b) = $this->getStreamPair(4096);
@ -218,7 +232,8 @@ class ResourceStreamTest extends TestCase {
});
}
public function testCloseStreamAfterEndPayload() {
public function testCloseStreamAfterEndPayload()
{
Loop::run(function () {
list($a, $b) = $this->getStreamPair();

18
test/ZlibInputStreamTest.php
View File

@ -10,8 +10,10 @@ use Amp\Loop;
use Amp\PHPUnit\TestCase;
use Amp\Producer;
class ZlibInputStreamTest extends TestCase {
public function testRead() {
class ZlibInputStreamTest extends TestCase
{
public function testRead()
{
Loop::run(function () {
$file1 = __DIR__ . "/fixtures/foobar.txt";
$file2 = __DIR__ . "/fixtures/foobar.txt.gz";
@ -37,19 +39,22 @@ class ZlibInputStreamTest extends TestCase {
});
}
public function testGetEncoding() {
public function testGetEncoding()
{
$gzStream = new ZlibInputStream(new InMemoryStream(""), \ZLIB_ENCODING_GZIP);
$this->assertSame(\ZLIB_ENCODING_GZIP, $gzStream->getEncoding());
}
public function testInvalidEncoding() {
public function testInvalidEncoding()
{
$this->expectException(StreamException::class);
new ZlibInputStream(new InMemoryStream(""), 1337);
}
public function testGetOptions() {
public function testGetOptions()
{
$options = [
"level" => -1,
"memory" => 8,
@ -62,7 +67,8 @@ class ZlibInputStreamTest extends TestCase {
$this->assertSame($options, $gzStream->getOptions());
}
public function testInvalidStream() {
public function testInvalidStream()
{
$this->expectException(StreamException::class);
Loop::run(function () {

26
test/ZlibOutputStreamTest.php
View File

@ -12,8 +12,10 @@ use Amp\ByteStream\ZlibOutputStream;
use Amp\Loop;
use Amp\PHPUnit\TestCase;
class ZlibOutputStreamTest extends TestCase {
public function testWrite() {
class ZlibOutputStreamTest extends TestCase
{
public function testWrite()
{
Loop::run(function () {
$file1 = __DIR__ . "/fixtures/foobar.txt";
$file2 = __DIR__ . "/fixtures/foobar.txt.gz";
@ -21,7 +23,7 @@ class ZlibOutputStreamTest extends TestCase {
$bufferStream = new OutputBuffer();
$outputStream = new ZlibOutputStream($bufferStream, \ZLIB_ENCODING_GZIP);
$fileStream = new ResourceInputStream(fopen($file1, "r"));
$fileStream = new ResourceInputStream(\fopen($file1, "r"));
while (($chunk = yield $fileStream->read()) !== null) {
yield $outputStream->write($chunk);
}
@ -39,7 +41,8 @@ class ZlibOutputStreamTest extends TestCase {
});
}
public function testThrowsOnWritingToClosedContext() {
public function testThrowsOnWritingToClosedContext()
{
$this->expectException(ClosedException::class);
Loop::run(function () {
@ -49,7 +52,8 @@ class ZlibOutputStreamTest extends TestCase {
});
}
public function testThrowsOnEndingToClosedContext() {
public function testThrowsOnEndingToClosedContext()
{
$this->expectException(ClosedException::class);
Loop::run(function () {
@ -59,19 +63,22 @@ class ZlibOutputStreamTest extends TestCase {
});
}
public function testGetEncoding() {
public function testGetEncoding()
{
$gzStream = new ZlibOutputStream(new OutputBuffer(), \ZLIB_ENCODING_GZIP);
$this->assertSame(\ZLIB_ENCODING_GZIP, $gzStream->getEncoding());
}
public function testInvalidEncoding() {
public function testInvalidEncoding()
{
$this->expectException(StreamException::class);
new ZlibOutputStream(new OutputBuffer(), 1337);
}
public function testGetOptions() {
public function testGetOptions()
{
$options = [
"level" => -1,
"memory" => 8,
@ -84,7 +91,8 @@ class ZlibOutputStreamTest extends TestCase {
$this->assertSame($options, $gzStream->getOptions());
}
public function testInvalidOptions() {
public function testInvalidOptions()
{
$this->expectException(StreamException::class);
new ZlibOutputStream(new OutputBuffer(), \ZLIB_ENCODING_GZIP, ["level" => 42]);