public function parseJsonStream(PromiseInterface $promise)
    {
        // application/json

        $in = $this->parsePlainStream($promise);
        $out = new ReadableStream();

        // invalid/closing input stream => return closed output stream
        if (!$in->isReadable()) {
            $out->close();

            return $out;
        }

        // forward each data chunk to the streaming JSON parser
        $parser = new StreamingJsonParser();
        $in->on('data', function ($data) use ($parser, $out) {
            $objects = $parser->push($data);

            foreach ($objects as $object) {
                if (isset($object['error'])) {
                    $out->emit('error', array(new JsonProgressException($object), $out));
                    $out->close();
                    return;
                }
                $out->emit('progress', array($object, $out));
            }
        });

        // forward error and make sure stream closes
        $in->on('error', function ($error) use ($out) {
            $out->emit('error', array($error, $out));
            $out->close();
        });

        // closing either stream closes the other one
        $in->on('close', array($out, 'close'));
        $out->on('close', array($in, 'close'));

        return $out;
    }
public function parseJsonStream(PromiseInterface $promise)
    {
        // application/json

        $in = $this->parsePlainStream($promise);
        $out = new ReadableStream();

        // invalid/closing input stream => return closed output stream
        if (!$in->isReadable()) {
            $out->close();

            return $out;
        }

        // forward each data chunk to the streaming JSON parser
        $parser = new StreamingJsonParser();
        $in->on('data', function ($data) use ($parser, $out) {
            $objects = $parser->push($data);

            foreach ($objects as $object) {
                if (isset($object['error'])) {
                    $out->emit('error', array(new JsonProgressException($object), $out));
                    $out->close();
                    return;
                }
                $out->emit('progress', array($object, $out));
            }
        });

        // forward error and make sure stream closes
        $in->on('error', function ($error) use ($out) {
            $out->emit('error', array($error, $out));
            $out->close();
        });

        // closing either stream closes the other one
        $in->on('close', array($out, 'close'));
        $out->on('close', array($in, 'close'));

        return $out;
    }
public function parseJsonStream ( PromiseInterface $ promise ) { // application/json $ in = $ this -> parsePlainStream ( $ promise ) ; $ out = new ReadableStream ( ) ; // invalid/closing input stream => return closed output stream if ( ! $ in -> isReadable ( ) ) { $ out -> close ( ) ; return $ out ; } // forward each data chunk to the streaming JSON parser $ parser = new StreamingJsonParser ( ) ; $ in -> on ( 'data' , function ( $ data ) use ( $ parser , $ out ) { $ objects = $ parser -> push ( $ data ) ; foreach ( $ objects as $ object ) { if ( isset ( $ object [ 'error' ] ) ) { $ out -> emit ( 'error' , array ( new JsonProgressException ( $ object ) , $ out ) ) ; $ out -> close ( ) ; return ; } $ out -> emit ( 'progress' , array ( $ object , $ out ) ) ; } } ) ; // forward error and make sure stream closes $ in -> on ( 'error' , function ( $ error ) use ( $ out ) { $ out -> emit ( 'error' , array ( $ error , $ out ) ) ; $ out -> close ( ) ; } ) ; // closing either stream closes the other one $ in -> on ( 'close' , array ( $ out , 'close' ) ) ; $ out -> on ( 'close' , array ( $ in , 'close' ) ) ; return $ out ; }
Returns a readable JSON stream for the given ResponseInterface

@param PromiseInterface $promise Promise<ResponseInterface>
@return ReadableStreamInterface
@uses self::parsePlainSream()
Returns a readable JSON stream for the given ResponseInterface