Feature/node streams (#257)

* update fromNodeStream test, add toNodeStream tests

* update fromNodeStream to only add one extra end event handler

* add iterable/asynciterable toNodeStream implementations

* ugh, update npm because semver

* fix lint

* update Ix.node AsyncIterable pipe implementation to support piping streams

Author: trxcllnt

.travis.yml

@@ -25,6 +25,7 @@ before_install:
   - export PATH=$HOME/.yarn/bin:$PATH
   - which yarn # for debugging Travis
   - echo $PATH # for debugging Travis
+  - node -v && npm i -g npm && npm -v
 
 script:
   - |

.vscode/launch.json

@@ -20,6 +20,7 @@
             "request": "launch",
             "name": "Debug Unit Tests",
             "cwd": "${workspaceRoot}",
+            "console": "integratedTerminal",
             "program": "${workspaceFolder}/node_modules/.bin/jest",
             "skipFiles": [
                 "<node_internals>/**/*.js",

gulp/closure-task.js

@@ -45,8 +45,8 @@ const closureTask = ((cache) => memoizeTask(cache, async function closure(target
     await mkdirp(out);
 
     await Promise.all([
-        `${mainExport}`,
-        `${mainExport}.internal`
+        `${mainExport}.dom`,
+        `${mainExport}.dom.internal`
     ].map(closureCompile));
 
     async function closureCompile(entry) {

gulp/minify-task.js

@@ -43,7 +43,7 @@ const minifyTask = ((cache, commonConfig) => memoizeTask(cache, function minifyJ
         output: { ...commonConfig.output,
             path: path.resolve(`./${out}`) } };
 
-    const webpackConfigs = [mainExport, `${mainExport}.internal`].map((entry) => ({
+    const webpackConfigs = [`${mainExport}.dom`, `${mainExport}.dom.internal`].map((entry) => ({
         ...targetConfig,
         name: entry,
         entry: { [entry]: path.resolve(`${src}/${entry}.js`) },

gulp/package-task.js

@@ -46,8 +46,8 @@ const createMainPackageJson = (target, format) => (orig) => ({
     ...createTypeScriptPackageJson(target, format)(orig),
     bin: orig.bin,
     name: npmPkgName,
-    browser: `${mainExport}`,
     main: `${mainExport}.node`,
+    browser: `${mainExport}.dom`,
     types: `${mainExport}.node.d.ts`,
     unpkg: `${mainExport}.es5.min.js`,
     [`esm`]: { mode: `all`, sourceMap: true }
@@ -56,9 +56,9 @@ const createMainPackageJson = (target, format) => (orig) => ({
 const createTypeScriptPackageJson = (target, format) => (orig) => ({
     ...createScopedPackageJSON(target, format)(orig),
     bin: undefined,
-    browser: `${mainExport}.ts`,
     main: `${mainExport}.node.ts`,
     types: `${mainExport}.node.ts`,
+    browser: `${mainExport}.dom.ts`,
     dependencies: {
         '@types/node': '*',
         ...orig.dependencies
@@ -71,9 +71,9 @@ const createScopedPackageJSON = (target, format) => (({ name, ...orig }) =>
             (xs, key) => ({ ...xs, [key]: xs[key] || orig[key] }),
             {
                 name: `${npmOrgName}/${packageName(target, format)}`,
-                browser: format === 'umd' ? undefined : `${mainExport}`,
-                main: format === 'umd' ? `${mainExport}` : `${mainExport}.node`,
-                types: format === 'umd' ? undefined : `${mainExport}.node.d.ts`,
+                browser: format === 'umd' ? undefined : `${mainExport}.dom`,
+                main: format === 'umd' ? `${mainExport}.dom` : `${mainExport}.node`,
+                types: format === 'umd' ? `${mainExport}.d.ts` : `${mainExport}.node.d.ts`,
                 version: undefined, unpkg: undefined, module: undefined, [`esm`]: undefined,
             }
         )

package.json

@@ -45,6 +45,7 @@
   ],
   "dependencies": {
     "@types/node": "10.12.18",
+    "is-stream": "1.1.0",
     "tslib": "^1.9.3"
   },
   "devDependencies": {

spec/Ix.ts

@@ -1,6 +1,6 @@
 /* tslint:disable */
 
-(<any> global).window = (<any> global).window || global;
+(<any>global).window = (<any>global).window || global;
 
 // these are duplicated in the gulpfile :<
 const targets = [`es5`, `es2015`, `esnext`];
@@ -20,7 +20,7 @@ else if (target === `ts` || target === `apache-arrow`) modulePath = target;
 else modulePath = path.join(target, format);
 
 modulePath = path.resolve(`./targets`, modulePath);
-const IxPath = path.join(modulePath, `Ix${format === 'umd' ? '' : '.node'}.internal`);
+const IxPath = path.join(modulePath, `Ix${format === 'umd' ? '.dom' : '.node'}.internal`);
 const IxInternal: typeof import('../src/Ix.node.internal') = require(IxPath);
 
 export = IxInternal;

spec/asynciterable-operators/tonodestream-spec.ts

@@ -0,0 +1,68 @@
+import '../asynciterablehelpers';
+import { AsyncIterable, AsyncIterableReadable } from '../Ix';
+
+(() => {
+  if (!AsyncIterableReadable || process.env.TEST_NODE_STREAMS !== 'true') {
+    return test('not testing node streams because process.env.TEST_NODE_STREAMS !== "true"', () => {
+      /**/
+    });
+  }
+
+  const stringsItr = () => AsyncIterable.from([1, 2, 3]).map(i => `${i}`);
+  const buffersItr = () => stringsItr().map(val => Buffer.from(val));
+  const objectsItr = () => stringsItr().map(val => ({ val }));
+  const compare = <T>(a: T, b: T) => {
+    let aVal = ArrayBuffer.isView(a) ? `${Buffer.from(a.buffer, a.byteOffset, a.byteLength)}` : a;
+    let bVal = ArrayBuffer.isView(b) ? `${Buffer.from(b.buffer, b.byteOffset, b.byteLength)}` : b;
+    // poor man's deep-equals
+    try {
+      expect(aVal).toEqual(bVal);
+    } catch (e) {
+      return false;
+    }
+    return true;
+  };
+
+  describe(`AsyncIterable#toNodeStream`, () => {
+    describe(`objectMode: true`, () => {
+      const expectedStrings = ['1', '2', '3'];
+      const expectedObjects = expectedStrings.map(val => ({ val }));
+      const expectedBuffers = expectedStrings.map(x => Buffer.from(x));
+      test(`yields Strings`, async () => {
+        await expect(stringsItr().toNodeStream({ objectMode: true })).toEqualStream(
+          AsyncIterable.from(expectedStrings),
+          compare
+        );
+      });
+      test(`yields Buffers`, async () => {
+        await expect(buffersItr().toNodeStream({ objectMode: true })).toEqualStream(
+          AsyncIterable.from(expectedBuffers),
+          compare
+        );
+      });
+      test(`yields Objects`, async () => {
+        await expect(objectsItr().toNodeStream({ objectMode: true })).toEqualStream(
+          AsyncIterable.from(expectedObjects),
+          compare
+        );
+      });
+    });
+
+    describe(`objectMode: false`, () => {
+      const expectedStrings = ['123'];
+      const expectedBuffers = expectedStrings.map(x => Buffer.from(x));
+      test(`yields Strings`, async () => {
+        await expect(stringsItr().toNodeStream({ objectMode: false })).toEqualStream(
+          AsyncIterable.from(expectedStrings),
+          compare
+        );
+      });
+      test(`yields Buffers`, async () => {
+        await expect(buffersItr().toNodeStream({ objectMode: false })).toEqualStream(
+          AsyncIterable.from(expectedBuffers),
+          compare
+        );
+      });
+    });
+  });
+})();

spec/asynciterable-operators/writable-stream-pipe-spec.ts

@@ -0,0 +1,93 @@
+import * as Ix from '../Ix';
+import { PassThrough } from 'stream';
+
+(() => {
+  if (process.env.TEST_NODE_STREAMS !== 'true') {
+    return test('not testing node streams because process.env.TEST_NODE_STREAMS !== "true"', () => {
+      /**/
+    });
+  }
+
+  const { of } = Ix.AsyncIterable;
+  const { empty, map } = Ix.asynciterable;
+  const { sequenceEqual } = Ix.asynciterable;
+  const through = () => {
+    return new PassThrough({
+      objectMode: true,
+      readableObjectMode: true,
+      writableObjectMode: true
+    });
+  };
+
+  test('AsyncIterable#pipe writable-stream single element', async () => {
+    const source = of({ name: 'Frank', custId: 98088 });
+    const expected = of('Frank');
+
+    expect(await sequenceEqual(expected, map(source.pipe(through()), x => x.name))).toBeTruthy();
+  });
+
+  test('AsyncIterable#pipe writable-stream maps property', async () => {
+    const source = of<any>(
+      { name: 'Frank', custId: 98088 },
+      { name: 'Bob', custId: 29099 },
+      { name: 'Chris', custId: 39033 },
+      { name: null, custId: 30349 },
+      { name: 'Frank', custId: 39030 }
+    );
+    const expected = of('Frank', 'Bob', 'Chris', null, 'Frank');
+
+    expect(await sequenceEqual(expected, map(source.pipe(through()), x => x.name))).toBeTruthy();
+  });
+
+  test('AsyncIterable#pipe writable-stream empty', async () => {
+    expect(
+      await sequenceEqual(empty<number>(), map(empty<string>(), (s, i) => s.length + i))
+    ).toBeTruthy();
+  });
+
+  test('AsyncIterable#pipe writable-stream map property using index', async () => {
+    const source = of(
+      { name: 'Frank', custId: 98088 },
+      { name: 'Bob', custId: 29099 },
+      { name: 'Chris', custId: 39033 }
+    );
+    const expected = of('Frank', null, null);
+
+    expect(
+      await sequenceEqual(
+        expected,
+        map(source.pipe(through()), (x, i) => (i === 0 ? x.name : null))
+      )
+    ).toBeTruthy();
+  });
+
+  test('AsyncIterable#pipe writable-stream map property using index on last', async () => {
+    const source = of(
+      { name: 'Frank', custId: 98088 },
+      { name: 'Bob', custId: 29099 },
+      { name: 'Chris', custId: 39033 },
+      { name: 'Bill', custId: 30349 },
+      { name: 'Frank', custId: 39030 }
+    );
+    const expected = of(null, null, null, null, 'Frank');
+
+    expect(
+      await sequenceEqual(
+        expected,
+        map(source.pipe(through()), (x, i) => (i === 4 ? x.name : null))
+      )
+    ).toBeTruthy();
+  });
+
+  test('AsyncIterable#pipe writable-stream execution is deferred', async () => {
+    let fnCalled = false;
+    const source = of(() => {
+      fnCalled = true;
+      return 1;
+    });
+
+    map(source.pipe(through()), x => x());
+
+    expect(fnCalled).toBeFalsy();
+  });
+})();

spec/asynciterable/fromnodestream-spec.ts

@@ -1,5 +1,6 @@
-import { fromNodeStream } from '../Ix';
+import '../asynciterablehelpers';
 import { Readable, ReadableOptions } from 'stream';
+import { fromNodeStream, AsyncIterable } from '../Ix';
 
 (() => {
   if (!fromNodeStream || process.env.TEST_NODE_STREAMS !== 'true') {
@@ -19,34 +20,25 @@ import { Readable, ReadableOptions } from 'stream';
     }
 
     _read() {
-      const i = ++this._index;
-      if (i > this._max) {
-        this.push(null);
-      } else {
-        const buf = Buffer.from(`${i}`, 'utf8');
-        this.push(buf);
-      }
+      this.push(++this._index > this._max ? null : `${this._index}`);
     }
   }
 
-  test('AsyncIterable#fromNodeStream with readable', async () => {
-    const c = new Counter({ objectMode: true });
-    const xs = fromNodeStream(c);
+  const compare = (a: string, b: string) => Buffer.from(a).compare(Buffer.from(b)) === 0;
 
-    const it = xs[Symbol.asyncIterator]();
-    let next = await it.next();
-    expect(next.done).toBeFalsy();
-    expect((next.value as Buffer).compare(Buffer.from('1', 'utf8'))).toBe(0);
-
-    next = await it.next();
-    expect(next.done).toBeFalsy();
-    expect((next.value as Buffer).compare(Buffer.from('2', 'utf8'))).toBe(0);
-
-    next = await it.next();
-    expect(next.done).toBeFalsy();
-    expect((next.value as Buffer).compare(Buffer.from('3', 'utf8'))).toBe(0);
+  describe(`AsyncIterable#fromNodeStream`, () => {
+    test('objectMode: true', async () => {
+      const c = new Counter({ objectMode: true });
+      const xs = fromNodeStream(c) as AsyncIterable<string>;
+      const expected = AsyncIterable.from(['1', '2', '3']);
+      await expect(xs).toEqualStream(expected, compare);
+    });
 
-    next = await it.next();
-    expect(next.done).toBeTruthy();
+    test('objectMode: false', async () => {
+      const c = new Counter({ objectMode: false });
+      const xs = fromNodeStream(c) as AsyncIterable<string>;
+      const expected = AsyncIterable.from(['123']);
+      await expect(xs).toEqualStream(expected, compare);
+    });
   });
 })();