feat(asasynciterable): add AsyncIterableTransform stream

AsyncIterableTransform allows constructing an AsyncIterableX instance from NodeJS ReadableStreams
via pipe()

Author: trxcllnt

docs/asynciterable/converting.md

@@ -97,7 +97,7 @@ We can then introduce IxJS by using the `fromNodeStream` which allows us then to
 
 ```typescript
 import * as fs from 'fs';
-import { fromNodeStream } from 'ix/asynciterable';
+import { fromNodeStream } from 'ix/asynciterable/fromnodestream';
 
 const readable = fs.createReadStream('tmp.txt', {encoding: 'utf8'});
 const source = fromNodeStream(readable);
@@ -107,6 +107,27 @@ for await (const chunk in source) {
 }
 ```
 
+Or we can use `asAsyncIterable()` to take advantage of Node Streams' fluent `pipe` API:
+
+```typescript
+import * as fs from 'fs';
+import { map } from 'ix/asynciterable/operators/map';
+import { flatMap } from 'ix/asynciterable/operators/flatmap';
+import { asAsyncIterable } from 'ix/asynciterable/asasynciterable';
+
+const source = fs
+    .createReadStream('tmp.txt', {encoding: 'utf8'})
+    // Transform a Node stream into an AsyncIterable
+    .pipe(asAsyncIterable({ objectMode: false }))
+    // The result here is an AsyncIterableX
+    .pipe(map((chunk, index) => `${index}: ${chunk}`));
+
+for await (const chunk in source) {
+  console.log(chunk);
+}
+
+```
+
 ## Creating a sequence from Events
 
 Although we traditionally think of events being push only such as Subject/Observer or Observables, we can also bridge to events using AsyncIterables.  To do that, we have a couple of mechanisms, one called `fromEvent` which binds either your DOM `EventTarget` or Node.js `EventEmitter` to a given event.  Then you can iterate over it just like any other async-iterable sequence.

spec/asynciterable/asasynciterable-spec.ts

@@ -0,0 +1,50 @@
+import '../asynciterablehelpers';
+import { from } from 'ix/asynciterable';
+import { Readable, ReadableOptions } from 'stream';
+import { asAsyncIterable } from 'ix/asynciterable/asasynciterable';
+
+// eslint-disable-next-line consistent-return
+(() => {
+  if (!asAsyncIterable || process.env.TEST_NODE_STREAMS !== 'true') {
+    return test('not testing node streams because process.env.TEST_NODE_STREAMS !== "true"', () => {
+      /**/
+    });
+  }
+
+  class Counter extends Readable {
+    private _index: number;
+    private _max: number;
+
+    constructor(options?: ReadableOptions) {
+      super(options);
+      this._max = 3;
+      this._index = 0;
+    }
+
+    _read() {
+      this.push(++this._index > this._max ? null : `${this._index}`);
+    }
+  }
+
+  const compare = (a: string, b: string) => Buffer.from(a).compare(Buffer.from(b)) === 0;
+
+  describe('AsyncIterable#fromNodeStream', () => {
+    test('objectMode: true', async () => {
+      const c = new Counter({ objectMode: true });
+      const xs = c.pipe(
+        asAsyncIterable<string>({ objectMode: true })
+      );
+      const expected = from(['1', '2', '3']);
+      await expect(xs).toEqualStream(expected, compare);
+    });
+
+    test('objectMode: false', async () => {
+      const c = new Counter({ objectMode: false });
+      const xs = c.pipe(
+        asAsyncIterable<string>({ objectMode: false })
+      );
+      const expected = from(['123']);
+      await expect(xs).toEqualStream(expected, compare);
+    });
+  });
+})();

src/Ix.node.ts

@@ -2,6 +2,7 @@ export * from './Ix';
 
 export { IterableReadable } from './iterable/tonodestream';
 export { AsyncIterableReadable } from './asynciterable/tonodestream';
+export { asAsyncIterable, AsyncIterableTransform } from './asynciterable/asasynciterable';
 export { fromNodeStream, ReadableStreamAsyncIterable } from './asynciterable/fromnodestream';
 
 import './add/asynciterable-operators/skipwhile';

src/asynciterable/asasynciterable.ts

@@ -0,0 +1,60 @@
+import { AsyncIterableX } from './asynciterablex';
+import { OperatorAsyncFunction, UnaryFunction } from '../interfaces';
+import { Transform, TransformCallback } from 'stream';
+
+export interface TransformOptions {
+  // ReadableOptions/WritableOptions
+  highWaterMark?: number;
+  objectMode?: boolean;
+  autoDestroy?: boolean;
+  // ReadableOptions
+  encoding?: string;
+  // WritableOptions
+  decodeStrings?: boolean;
+  defaultEncoding?: string;
+  emitClose?: boolean;
+  // DuplexOptions
+  allowHalfOpen?: boolean;
+  readableObjectMode?: boolean;
+  writableObjectMode?: boolean;
+  readableHighWaterMark?: number;
+  writableHighWaterMark?: number;
+  writableCorked?: number;
+}
+
+export interface AsyncIterableTransform<T> extends AsyncIterableX<T>, Transform {
+  pipe<R>(...operations: UnaryFunction<AsyncIterable<T>, R>[]): R;
+  pipe<R>(...operations: OperatorAsyncFunction<T, R>[]): AsyncIterableX<R>;
+  pipe<R extends NodeJS.WritableStream>(writable: R, options?: { end?: boolean }): R;
+  [Symbol.asyncIterator](): AsyncIterableIterator<T>;
+}
+
+const asyncIterableMixin = Symbol('asyncIterableMixin');
+
+export class AsyncIterableTransform<T> extends Transform {
+  private static [asyncIterableMixin] = false;
+  constructor(options?: TransformOptions) {
+    super(options);
+    // If this is the first time AsyncIterableTransform is being constructed,
+    // mixin the methods from the AsyncIterableX's prototype.
+    if (!AsyncIterableTransform[asyncIterableMixin]) {
+      AsyncIterableTransform[asyncIterableMixin] = true;
+      Object.defineProperties(
+        AsyncIterableTransform.prototype,
+        Object.getOwnPropertyDescriptors(AsyncIterableX.prototype)
+      );
+    }
+  }
+  /** @nocollapse */
+  _flush(callback: TransformCallback): void {
+    callback();
+  }
+  /** @nocollapse */
+  _transform(chunk: any, _encoding: string, callback: TransformCallback): void {
+    callback(null, chunk);
+  }
+}
+
+export function asAsyncIterable<T>(options: TransformOptions = {}) {
+  return new AsyncIterableTransform<T>(options);
+}

src/asynciterable/index.node.ts

@@ -1,4 +1,4 @@
 export * from './index';
-
 export * from './tonodestream';
 export * from './fromnodestream';
+export * from './asasynciterable';