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;
    }