Node Streams JSON Transformer

JsonLinesTransform.js

var stream = require('stream');

class JsonLinesTransform extends stream.Transform {
  _transform(chunk, env, cb) {
    if(!this.chunks) {
      this.chunks = "";
    }

    this.chunks += chunk;

    var lines = this.chunks.split(/\n/);

    this.chunks = lines.pop();

    for(let i = 0; i < lines.length; i++) {
      this.push(lines[i]);
    }

    cb();
  }
}

exports.JsonLinesTransform=JsonLinesTransform;

JsonLinesTransform.spec.js

const { expect } = require('chai');

const { PassThrough, Writable, Readable } = require('stream');
const { JsonLinesTransform } = require('./JsonLinesTransform');

class ChunkyReadable extends Readable {
  constructor(chunks) {
    super();
    this.chunks = JSON.parse(JSON.stringify(chunks));
  }

  _read() {
    this.push(this.chunks.shift());
  }
}

class ChunkyWritable extends Writable {
  constructor() {
    super();
    this.chunks = [];
  }

  _write(chunk, encoding, callback) {
    this.chunks.push(chunk.toString());
    callback();
  }
}

let transform;

beforeEach(() => {
  transform = new JsonLinesTransform();
});

it('should transform json chunks', async () => {

  const chunks = [
    `{"foo":"bar"}\n`,
    `{"foo":"bar2"}\n`,
    null
  ];

  const readable = new ChunkyReadable(chunks);

  const writable = new ChunkyWritable();

  const p = new Promise((resolve, reject) => {
    readable
      .pipe(transform)
      .pipe(writable)
      .on('error', reject)
      .on('finish', resolve)
    ;
  });

  await p;

  expect(writable.chunks[0]).to.eql(chunks[0].replace(/\n/,''));
});

it('should transform json partial chunks', async () => {

  const chunks = [
    `{"foo"`,
    `:"bar2"}\n`,
    null
  ];

  const readable = new ChunkyReadable(chunks);

  const writable = new ChunkyWritable();

  const p = new Promise((resolve, reject) => {
    readable
      .pipe(transform)
      .pipe(writable)
      .on('error', reject)
      .on('finish', resolve)
    ;
  });

  await p;

  expect(writable.chunks[0]).to.eql(`{"foo":"bar2"}`);
});