public function deferredStream(ReadableStreamInterface $stream, $progressEventName)
    {
        // cancelling the deferred will (try to) close the stream
        $deferred = new Deferred(function () use ($stream) {
            $stream->close();

            throw new RuntimeException('Cancelled');
        });

        if ($stream->isReadable()) {
            // buffer all data events for deferred resolving
            $buffered = array();
            $stream->on($progressEventName, function ($data) use (&$buffered) {
                $buffered []= $data;
            });

            // error event rejects
            $stream->on('error', function ($error) use ($deferred) {
                $deferred->reject($error);
            });

            // close event resolves with buffered events (unless already error'ed)
            $stream->on('close', function () use ($deferred, &$buffered) {
                $deferred->resolve($buffered);
            });
        } else {
            $deferred->reject(new RuntimeException('Stream already ended, looks like it could not be opened'));
        }

        return $deferred->promise();
    }
public function deferredStream(ReadableStreamInterface $stream, $progressEventName)
    {
        // cancelling the deferred will (try to) close the stream
        $deferred = new Deferred(function () use ($stream) {
            $stream->close();

            throw new RuntimeException('Cancelled');
        });

        if ($stream->isReadable()) {
            // buffer all data events for deferred resolving
            $buffered = array();
            $stream->on($progressEventName, function ($data) use (&$buffered) {
                $buffered []= $data;
            });

            // error event rejects
            $stream->on('error', function ($error) use ($deferred) {
                $deferred->reject($error);
            });

            // close event resolves with buffered events (unless already error'ed)
            $stream->on('close', function () use ($deferred, &$buffered) {
                $deferred->resolve($buffered);
            });
        } else {
            $deferred->reject(new RuntimeException('Stream already ended, looks like it could not be opened'));
        }

        return $deferred->promise();
    }
public function deferredStream ( ReadableStreamInterface $ stream , $ progressEventName ) { // cancelling the deferred will (try to) close the stream $ deferred = new Deferred ( function ( ) use ( $ stream ) { $ stream -> close ( ) ; throw new RuntimeException ( 'Cancelled' ) ; } ) ; if ( $ stream -> isReadable ( ) ) { // buffer all data events for deferred resolving $ buffered = array ( ) ; $ stream -> on ( $ progressEventName , function ( $ data ) use ( & $ buffered ) { $ buffered [ ] = $ data ; } ) ; // error event rejects $ stream -> on ( 'error' , function ( $ error ) use ( $ deferred ) { $ deferred -> reject ( $ error ) ; } ) ; // close event resolves with buffered events (unless already error'ed) $ stream -> on ( 'close' , function ( ) use ( $ deferred , & $ buffered ) { $ deferred -> resolve ( $ buffered ) ; } ) ; } else { $ deferred -> reject ( new RuntimeException ( 'Stream already ended, looks like it could not be opened' ) ) ; } return $ deferred -> promise ( ) ; }
Returns a promise which resolves with an array of all "progress" events

@param ReadableStreamInterface $stream
@param string                  $progressEventName the name of the event to collect
@return PromiseInterface Promise<array, Exception>
Returns a promise which resolves with an array of all progress events