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) {
    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}>`);
    do {
      const xml = theCharacters
        .slice(
          nextOpenTagIndex,
          nextCloseTagIndex + this.options.entryTagName.length + 3,
        )
        .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 + this.options.entryTagName.length + 3);
      nextOpenTagIndex = theCharacters.indexOf(`<${this.options.entryTagName}>`);
      nextCloseTagIndex = theCharacters.indexOf(`</${this.options.entryTagName}>`);
    } while (nextOpenTagIndex !== -1 && nextCloseTagIndex !== -1);
    this.charactersToParse = theCharacters;

    callback(null, '');
  }
}

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