File stream ends before node-expat stream ends

[antoniobusrod]

It seems when you use pause and resume functions to run async operations when you get an endElement event, xml-stream continues reading. I sometimes have problems when readable file stream ends before node-expat stream completes.

In my code, I transform detected endElement objects and write them asynchronously to a file using pause and resume xml-stream methods during write operation. In large files, when I handle end event I end the writable file stream. But after that, I get more endElement events and I get next error (cannot write in a closed writable stream).

Error: write after end
    at writeAfterEnd (_stream_writable.js:166:12)
    at WriteStream.Writable.write (_stream_writable.js:211:5)

I am working in a PR to mitigate this error.

[antoniobusrod]

#67

[cjuega]

This issue still happens, is there any plan to fix it or should I create a PR?

[vbarbarosh]

This is what I did in order to get around this bug:

import Promise from 'bluebird';
import XmlStream from 'xml-stream'; // https://github.com/vbarbarosh/xml-stream

/**
 * Parse input stream as XML data. Call `callback` for each record.
 * Wait until it returns. Throw if it fails.
 *
 * Structure of XML data should be as follows:
 *
 *     <root>
 *         <item>...</item>
 *         <item>...</item>
 *         <item>...</item>
 *         <item>...</item>
 *     </root>
 *
 * @param stream
 * @param callback
 * @returns {Promise<void>}
 * @link https://github.com/assistunion/xml-stream/issues/66
 */
async function stream_parse_xml(stream, callback)
{
    const queue = [];
    let paused = false;
    let finished = false;
    await new Promise(function (resolve, reject) {
        const xs = new XmlStream(stream);
        xs.on('endItem', function (item) {
            queue.push(item);
            if (paused) {
                return;
            }
            paused = true;
            stream.pause();
            next();
        });
        // BUG Using `xs.on('end', ...)` might produce `endElement/endItem` events afterwards
        xs.on('end', function () {
            finished = true;
        });
        xs.on('error', panic);
        function next() {
            if (queue.length) {
                const item = queue.shift();
                Promise.method(callback).call(null, item).then(next).catch(panic);
                return;
            }
            if (paused) {
                paused = false;
                stream.resume();
            }
            if (finished) {
                resolve();
            }
        }
        function panic(error) {
            if (!paused) {
                stream.pause();
                paused = true;
            }
            reject(error);
        }
    });
}

export default stream_parse_xml;