[utils/streams] implement some generic stream helpers

src/utils/index.js

@@ -4,3 +4,14 @@ export deepCloneWithBuffers from './deep_clone_with_buffers';
 export fromRoot from './from_root';
 export pkg from './package_json';
 export unset from './unset';
+
+export {
+  createConcatStream,
+  createIntersperseStream,
+  createJsonParseStream,
+  createJsonStringifyStream,
+  createListStream,
+  createPromiseFromStreams,
+  createReduceStream,
+  createSplitStream,
+} from './streams';

src/utils/streams/__tests__/concat_stream.js

@@ -0,0 +1,79 @@
+import expect from 'expect.js';
+
+import {
+  createListStream,
+  createPromiseFromStreams,
+  createConcatStream
+} from '../';
+
+describe('concatStream', () => {
+  it('accepts an initial value', async () => {
+    const output = await createPromiseFromStreams([
+      createListStream([1,2,3]),
+      createConcatStream([0])
+    ]);
+
+    expect(output).to.eql([0,1,2,3]);
+  });
+
+  describe(`combines using the previous value's concat method`, () => {
+    it('works with strings', async () => {
+      const output = await createPromiseFromStreams([
+        createListStream([
+          'a',
+          'b',
+          'c'
+        ]),
+        createConcatStream()
+      ]);
+      expect(output).to.eql('abc');
+    });
+
+    it('works with arrays', async () => {
+      const output = await createPromiseFromStreams([
+        createListStream([
+          [1],
+          [2,3,4],
+          [10]
+        ]),
+        createConcatStream()
+      ]);
+      expect(output).to.eql([1,2,3,4,10]);
+    });
+
+    it('works with a mixture, starting with array', async () => {
+      const output = await createPromiseFromStreams([
+        createListStream([
+          [],
+          1,
+          2,
+          3,
+          4,
+          [5,6,7]
+        ]),
+        createConcatStream()
+      ]);
+      expect(output).to.eql([1,2,3,4,5,6,7]);
+    });
+
+    it('fails when the value does not have a concat method', async () => {
+      let promise;
+      try {
+        promise = createPromiseFromStreams([
+          createListStream([1, '1']),
+          createConcatStream()
+        ]);
+      } catch (err) {
+        expect.fail('createPromiseFromStreams() should not fail synchronously');
+      }
+
+      try {
+        await promise;
+        expect.fail('Promise should have rejected');
+      } catch (err) {
+        expect(err).to.be.an(Error);
+        expect(err.message).to.contain('concat');
+      }
+    });
+  });
+});

src/utils/streams/__tests__/intersperse_stream.js

@@ -0,0 +1,38 @@
+import expect from 'expect.js';
+import sinon from 'sinon';
+
+import {
+  createPromiseFromStreams,
+  createListStream,
+  createIntersperseStream,
+  createConcatStream
+} from '../';
+
+describe('intersperseStream', () => {
+  it('places the intersperse value between each provided value', async () => {
+    expect(
+      await createPromiseFromStreams([
+        createListStream(['to', 'be', 'or', 'not', 'to', 'be']),
+        createIntersperseStream(' '),
+        createConcatStream()
+      ])
+    ).to.be('to be or not to be');
+  });
+
+  it('emits values as soon as possible, does not needlessly buffer', async () => {
+    const str = createIntersperseStream('y');
+    const stub = sinon.stub();
+    str.on('data', stub);
+
+    str.write('a');
+    sinon.assert.calledOnce(stub);
+    expect(stub.firstCall.args).to.eql(['a']);
+    stub.reset();
+
+    str.write('b');
+    sinon.assert.calledTwice(stub);
+    expect(stub.firstCall.args).to.eql(['y']);
+    sinon.assert.calledTwice(stub);
+    expect(stub.secondCall.args).to.eql(['b']);
+  });
+});

src/utils/streams/__tests__/json_parse_stream.js

@@ -0,0 +1,72 @@
+import expect from 'expect.js';
+
+import {
+  createPromiseFromStreams,
+  createListStream,
+  createConcatStream,
+  createJsonParseStream
+} from '../';
+
+describe('jsonParseStream', () => {
+  describe('standard usage', () => {
+    it('parses json strings', async () => {
+      const str = createJsonParseStream();
+      const dataPromise = new Promise((resolve, reject) => {
+        str.on('data', resolve);
+        str.on('error', reject);
+      });
+      str.write('{ "foo": "bar" }');
+
+      expect(await dataPromise).to.eql({
+        foo: 'bar'
+      });
+    });
+
+    it('parses json value passed to it from a list stream', async () => {
+      expect(await createPromiseFromStreams([
+        createListStream([
+          '"foo"',
+          '1'
+        ]),
+        createJsonParseStream(),
+        createConcatStream([])
+      ]))
+      .to.eql(['foo', 1]);
+    });
+  });
+
+  describe('error handling', () => {
+    it('emits an error when there is a parse failure', async () => {
+      const str = createJsonParseStream();
+      const errorPromise = new Promise(resolve => str.once('error', resolve));
+      str.write('{"partial');
+      const err = await errorPromise;
+      expect(err).to.be.an(Error);
+      expect(err).to.have.property('name', 'SyntaxError');
+    });
+
+    it('continues parsing after an error', async () => {
+      const str = createJsonParseStream();
+
+      const firstEmitPromise = new Promise(resolve => {
+        str.once('error', v => resolve({ name: 'error', value: v }));
+        str.once('data', v => resolve({ name: 'data', value: v }));
+      });
+
+      str.write('{"partial');
+      const firstEmit = await firstEmitPromise;
+      expect(firstEmit).to.have.property('name', 'error');
+      expect(firstEmit.value).to.be.an(Error);
+
+      const secondEmitPromise = new Promise(resolve => {
+        str.once('error', v => resolve({ name: 'error', value: v }));
+        str.once('data', v => resolve({ name: 'data', value: v }));
+      });
+
+      str.write('42');
+      const secondEmit = await secondEmitPromise;
+      expect(secondEmit).to.have.property('name', 'data');
+      expect(secondEmit).to.have.property('value', 42);
+    });
+  });
+});

src/utils/streams/__tests__/json_stringify_stream.js

@@ -0,0 +1,79 @@
+import expect from 'expect.js';
+
+import {
+  createPromiseFromStreams,
+  createListStream,
+  createConcatStream,
+  createJsonStringifyStream
+} from '../';
+
+function createCircularStructure() {
+  const obj = {};
+  obj.obj = obj; // create circular reference
+  return obj;
+}
+
+describe('jsonStringifyStream', () => {
+  describe('standard usage', () => {
+    it('stringifys js values', async () => {
+      const str = createJsonStringifyStream();
+      const dataPromise = new Promise((resolve, reject) => {
+        str.on('data', resolve);
+        str.on('error', reject);
+      });
+      str.write({ foo: 'bar' });
+
+      expect(await dataPromise).to.be('{"foo":"bar"}');
+    });
+
+    it('stringifys js values passed from a list stream', async () => {
+      const all = await createPromiseFromStreams([
+        createListStream([
+          'foo',
+          1
+        ]),
+        createJsonStringifyStream(),
+        createConcatStream([])
+      ]);
+
+      expect(all).to.eql(['"foo"', '1']);
+    });
+  });
+
+  describe('error handling', () => {
+    it('emits an error when there is a parse failure', async () => {
+      const str = createJsonStringifyStream();
+      const errorPromise = new Promise(resolve => str.once('error', resolve));
+      str.write(createCircularStructure());
+      const err = await errorPromise;
+      expect(err).to.be.an(Error);
+      expect(err).to.have.property('name', 'TypeError');
+      expect(err.message).to.contain('circular');
+    });
+
+    it('continues parsing after an error', async () => {
+      const str = createJsonStringifyStream();
+
+      const firstEmitPromise = new Promise(resolve => {
+        str.once('error', v => resolve({ name: 'error', value: v }));
+        str.once('data', v => resolve({ name: 'data', value: v }));
+      });
+
+      str.write(createCircularStructure());
+
+      const firstEmit = await firstEmitPromise;
+      expect(firstEmit).to.have.property('name', 'error');
+      expect(firstEmit.value).to.be.an(Error);
+
+      const secondEmitPromise = new Promise(resolve => {
+        str.once('error', v => resolve({ name: 'error', value: v }));
+        str.once('data', v => resolve({ name: 'data', value: v }));
+      });
+
+      str.write('foo');
+      const secondEmit = await secondEmitPromise;
+      expect(secondEmit).to.have.property('name', 'data');
+      expect(secondEmit).to.have.property('value', '"foo"');
+    });
+  });
+});

src/utils/streams/__tests__/list_stream.js

@@ -0,0 +1,28 @@
+import expect from 'expect.js';
+import sinon from 'sinon';
+
+import { createListStream } from '../';
+
+describe('listStream', () => {
+  it('provides the values in the initial list', async () => {
+    const str = createListStream([1,2,3,4]);
+    const stub = sinon.stub();
+    str.on('data', stub);
+
+    await new Promise(resolve => str.on('end', resolve));
+
+    sinon.assert.callCount(stub, 4);
+    expect(stub.getCall(0).args).to.eql([1]);
+    expect(stub.getCall(1).args).to.eql([2]);
+    expect(stub.getCall(2).args).to.eql([3]);
+    expect(stub.getCall(3).args).to.eql([4]);
+  });
+
+  it('does not modify the list passed', async () => {
+    const list = [1,2,3,4];
+    const str = createListStream(list);
+    str.resume();
+    await new Promise(resolve => str.on('end', resolve));
+    expect(list).to.eql([1,2,3,4]);
+  });
+});