murasaki-core/src/streams.ts

import { Transform, TransformCallback } from 'stream';
import { xml2json } from 'xml-js';

export interface XmlToJsonLinesTransformStreamOptions {
  entryTagName: string;
}

class XmlToJsonLinesTransformStream extends Transform {
  private charactersToParse?: string;

  constructor(private readonly options: XmlToJsonLinesTransformStreamOptions) {
    super();
    // noop
  }

  // eslint-disable-next-line no-underscore-dangle
  _transform(chunk: Buffer, _encoding: BufferEncoding, callback: TransformCallback) {
    try {
      const chunkStr = chunk.toString('utf-8');
      if (typeof this.charactersToParse !== 'string') {
        const firstEntryIndex = chunkStr.indexOf(`<${this.options.entryTagName}>`);
        this.charactersToParse = chunkStr.slice(firstEntryIndex);
      } else {
        this.charactersToParse += chunkStr;
      }

      let theCharacters = `${this.charactersToParse}`;
      let nextOpenTagIndex = theCharacters.indexOf(`<${this.options.entryTagName}>`);
      let nextCloseTagIndex = theCharacters.indexOf(`</${this.options.entryTagName}>`);
      const closeTagLength = this.options.entryTagName.length + 3;
      do {
        const xml = theCharacters
          .slice(nextOpenTagIndex, nextCloseTagIndex + closeTagLength)
          .replace(/&(.+?);/g, '$1'); // FIXME better handling of XML entities??? This makes the pipe hang for some reason
        const json = xml2json(xml, { compact: true });
        this.push(`${json}\n`);
        theCharacters = theCharacters.slice(nextCloseTagIndex + closeTagLength);
        nextOpenTagIndex = theCharacters.indexOf(`<${this.options.entryTagName}>`);
        nextCloseTagIndex = theCharacters.indexOf(`</${this.options.entryTagName}>`);
      } while (nextOpenTagIndex !== -1 && nextCloseTagIndex !== -1);
      this.charactersToParse = theCharacters;
      callback(null, '');
    } catch (err) {
      callback(err as Error);
    }
  }
}

export const createXmlToJsonLines = (options = {
  entryTagName: 'entry',
} as XmlToJsonLinesTransformStreamOptions) => new XmlToJsonLinesTransformStream(options);