Repository files navigation

The missing link between Promise-land and Stream-landforReactPHP.

Table of Contents

• Usage
  • buffer()
  • first()
  • all()
  • unwrapReadable()
  • unwrapWritable()
• Install
• Tests
• License

This lightweight library consists only of a few simple functions.All functions reside under theReact\Promise\Stream namespace.

The below examples refer to all functions with their fully-qualified names like this:

React\Promise\Stream\buffer(…);

As of PHP 5.6+ you can also import each required function into your code like this:

usefunctionReact\Promise\Stream\buffer;buffer(…);

Alternatively, you can also use an import statement similar to this:

useReact\Promise\Stream;Stream\buffer(…);

Thebuffer(ReadableStreamInterface<string> $stream, ?int $maxLength = null): PromiseInterface<string> function can be used tocreate aPromise which will be fulfilled with the stream data buffer.

$stream =accessSomeJsonStream();React\Promise\Stream\buffer($stream)->then(function (string$contents) {var_dump(json_decode($contents));});

The promise will be fulfilled with astring of all data chunks concatenated once the stream closes.

The promise will be fulfilled with an emptystring if the stream is already closed.

The promise will be rejected with aRuntimeException if the stream emits an error.

The promise will be rejected with aRuntimeException if it is cancelled.

The optional$maxLength argument defaults to no limit. In case the maximumlength is given and the stream emits more data before the end, the promisewill be rejected with anOverflowException.

$stream =accessSomeToLargeStream();React\Promise\Stream\buffer($stream,1024)->then(function ($contents) {var_dump(json_decode($contents));},function ($error) {// Reaching here when the stream buffer goes above the max size,// in this example that is 1024 bytes,// or when the stream emits an error.});

Thefirst(ReadableStreamInterface|WritableStreamInterface $stream, string $event = 'data'): PromiseInterface<mixed> function can be used tocreate aPromise which will be fulfilled once the given event triggers for the first time.

$stream =accessSomeJsonStream();React\Promise\Stream\first($stream)->then(function (string$chunk) {echo'The first chunk arrived:' .$chunk;});

The promise will be fulfilled with amixed value of whatever the first eventemitted ornull if the event does not pass any data.If you do not pass a custom event name, then it will wait for the first "data"event.For common streams of typeReadableStreamInterface<string>, this means it will befulfilled with astring containing the first data chunk.

The promise will be rejected with aRuntimeException if the stream emits an error– unless you're waiting for the "error" event, in which case it will be fulfilled.

The promise will be rejected with aRuntimeException once the stream closes– unless you're waiting for the "close" event, in which case it will be fulfilled.

The promise will be rejected with aRuntimeException if the stream is already closed.

The promise will be rejected with aRuntimeException if it is cancelled.

Theall(ReadableStreamInterface|WritableStreamInterface $stream, string $event = 'data'): PromiseInterface<array> function can be used tocreate aPromise which will be fulfilled with an array of all the event data.

$stream =accessSomeJsonStream();React\Promise\Stream\all($stream)->then(function (array$chunks) {echo'The stream consists of' .count($chunks) .' chunk(s)';});

The promise will be fulfilled with anarray once the stream closes. The arraywill contain whatever all events emitted ornull values if the events do not pass any data.If you do not pass a custom event name, then it will wait for all the "data"events.For common streams of typeReadableStreamInterface<string>, this means it will befulfilled with astring[] array containing all the data chunk.

The promise will be fulfilled with an emptyarray if the stream is already closed.

The promise will be rejected with aRuntimeException if the stream emits an error.

The promise will be rejected with aRuntimeException if it is cancelled.

TheunwrapReadable(PromiseInterface<ReadableStreamInterface<T>> $promise): ReadableStreamInterface<T> function can be used tounwrap aPromise which will be fulfilled with aReadableStreamInterface<T>.

This function returns a readable stream instance (implementingReadableStreamInterface<T>)right away which acts as a proxy for the future promise resolution.Once the given Promise will be fulfilled with aReadableStreamInterface<T>, itsdata will be piped to the output stream.

//$promise = someFunctionWhichResolvesWithAStream();$promise =startDownloadStream($uri);$stream =React\Promise\Stream\unwrapReadable($promise);$stream->on('data',function (string$data) {echo$data;});$stream->on('end',function () {echo'DONE';});

If the given promise is either rejected or fulfilled with anything but aninstance ofReadableStreamInterface, then the output stream will emitanerror event and close:

$promise =startDownloadStream($invalidUri);$stream =React\Promise\Stream\unwrapReadable($promise);$stream->on('error',function (Exception$error) {echo'Error:' .$error->getMessage();});

The given$promise SHOULD be pending, i.e. it SHOULD NOT be fulfilled or rejectedat the time of invoking this function.If the given promise is already settled and does not fulfill with an instance ofReadableStreamInterface, then you will not be able to receive theerror event.

You canclose() the resulting stream at any time, which will either try tocancel() the pending promise or try toclose() the underlying stream.

$promise =startDownloadStream($uri);$stream =React\Promise\Stream\unwrapReadable($promise);$loop->addTimer(2.0,function ()use ($stream) {$stream->close();});

TheunwrapWritable(PromiseInterface<WritableStreamInterface<T>> $promise): WritableStreamInterface<T> function can be used tounwrap aPromise which will be fulfilled with aWritableStreamInterface<T>.

This function returns a writable stream instance (implementingWritableStreamInterface<T>)right away which acts as a proxy for the future promise resolution.Any writes to this instance will be buffered in memory for when the promise willbe fulfilled.Once the given Promise will be fulfilled with aWritableStreamInterface<T>, anydata you have written to the proxy will be forwarded transparently to the innerstream.

//$promise = someFunctionWhichResolvesWithAStream();$promise =startUploadStream($uri);$stream =React\Promise\Stream\unwrapWritable($promise);$stream->write('hello');$stream->end('world');$stream->on('close',function () {echo'DONE';});

If the given promise is either rejected or fulfilled with anything but aninstance ofWritableStreamInterface, then the output stream will emitanerror event and close:

$promise =startUploadStream($invalidUri);$stream =React\Promise\Stream\unwrapWritable($promise);$stream->on('error',function (Exception$error) {echo'Error:' .$error->getMessage();});

The given$promise SHOULD be pending, i.e. it SHOULD NOT be fulfilled or rejectedat the time of invoking this function.If the given promise is already settled and does not fulfill with an instance ofWritableStreamInterface, then you will not be able to receive theerror event.

You canclose() the resulting stream at any time, which will either try tocancel() the pending promise or try toclose() the underlying stream.

$promise =startUploadStream($uri);$stream =React\Promise\Stream\unwrapWritable($promise);$loop->addTimer(2.0,function ()use ($stream) {$stream->close();});

The recommended way to install this library isthrough Composer.New to Composer?

This project followsSemVer.This will install the latest supported version:

composer require react/promise-stream:^1.7

See also theCHANGELOG for details about version upgrades.

This project aims to run on any platform and thus does not require any PHPextensions and supports running on legacy PHP 5.3 through current PHP 8+ andHHVM.It'shighly recommended to use the latest supported PHP version for this project.

To run the test suite, you first need to clone this repo and then install alldependenciesthrough Composer:

composer install

To run the test suite, go to the project root and run:

vendor/bin/phpunit

MIT, seeLICENSE file.