feat: refactor stream/Readable things

ReadableTyped is now better typed to support
iterator helpers

Author: kirillgroshkov

src/index.ts

@@ -41,8 +41,6 @@ export * from './stream/pipeline/pipeline'
 export * from './stream/readable/readableCreate'
 export * from './stream/readable/readableForEach'
 export * from './stream/readable/readableFromArray'
-export * from './stream/readable/readableMap'
-export * from './stream/readable/readableMapToArray'
 export * from './stream/readable/readableToArray'
 export * from './stream/stream.model'
 export * from './stream/progressLogger'

src/stream/progressLogger.ts

@@ -10,8 +10,9 @@ import {
 } from '@naturalcycles/js-lib'
 import { boldWhite, dimGrey, hasColors, white, yellow } from '../colors/colors'
 import { SizeStack } from './sizeStack'
+import { ReadableMapper } from './stream.model'
 
-export interface ProgressLoggerCfg<IN = any> {
+export interface ProgressLoggerCfg<T = any> {
   /**
    * Progress metric
    *
@@ -98,7 +99,7 @@ export interface ProgressLoggerCfg<IN = any> {
    *
    * chunk is undefined for "final" stats, otherwise is defined.
    */
-  extra?: (chunk: IN | undefined, index: number) => AnyObject
+  extra?: (chunk: T | undefined, index: number) => AnyObject
 
   /**
    * If specified - will multiply the counter by this number.
@@ -150,8 +151,8 @@ const inspectOpt: InspectOptions = {
   breakLength: 300,
 }
 
-export class ProgressLogger<IN> implements Disposable {
-  constructor(cfg: ProgressLoggerCfg<IN> = {}) {
+export class ProgressLogger<T> implements Disposable {
+  constructor(cfg: ProgressLoggerCfg<T> = {}) {
     this.cfg = {
       metric: 'progress',
       rss: true,
@@ -170,7 +171,7 @@ export class ProgressLogger<IN> implements Disposable {
     this.logStats() // initial
   }
 
-  cfg!: ProgressLoggerCfg<IN> & {
+  cfg!: ProgressLoggerCfg<T> & {
     logEvery: number
     logSizesBuffer: number
     batchSize: number
@@ -201,7 +202,7 @@ export class ProgressLogger<IN> implements Disposable {
       : undefined
   }
 
-  log(chunk?: IN): void {
+  log(chunk?: T): void {
     this.progress++
     this.processedLastSecond++
 
@@ -223,7 +224,7 @@ export class ProgressLogger<IN> implements Disposable {
     this.done()
   }
 
-  private logStats(chunk?: IN, final = false, tenx = false): void {
+  private logStats(chunk?: T, final = false, tenx = false): void {
     if (!this.cfg.logProgress) return
 
     const {
@@ -304,6 +305,20 @@ export class ProgressLogger<IN> implements Disposable {
 /**
  * Create new ProgressLogger.
  */
-export function progressLogger<IN>(cfg: ProgressLoggerCfg<IN> = {}): ProgressLogger<IN> {
+export function progressLogger<T>(cfg: ProgressLoggerCfg<T> = {}): ProgressLogger<T> {
   return new ProgressLogger(cfg)
 }
+
+/**
+ * Limitation: I don't know how to catch the `final` callback to log final stats.
+ *
+ * @experimental
+ */
+export function progressReadableMapper<T>(cfg: ProgressLoggerCfg<T> = {}): ReadableMapper<T, T> {
+  const progress = new ProgressLogger(cfg)
+
+  return chunk => {
+    progress.log(chunk)
+    return chunk
+  }
+}

src/stream/readable/readableCreate.ts

@@ -34,8 +34,8 @@ export function readableCreate<T>(
  * Convenience type-safe wrapper around Readable.from() that infers the Type of input.
  */
 export function readableFrom<T>(
-  items: Iterable<T> | AsyncIterable<T>,
+  iterable: Iterable<T> | AsyncIterable<T>,
   opt?: ReadableOptions,
 ): ReadableTyped<T> {
-  return Readable.from(items, opt)
+  return Readable.from(iterable, opt)
 }

src/stream/readable/readableFromArray.test.ts

@@ -1,14 +1,12 @@
 import { _range, pDelay } from '@naturalcycles/js-lib'
-import { _pipelineToArray, readableFromArray } from '../../index'
+import { readableFromArray } from '../../index'
 
 test('readableFromArray', async () => {
   const items = _range(1, 11)
 
   const readable = readableFromArray(items, async item => await pDelay(10, item))
 
-  const r = await _pipelineToArray([readable])
-
-  // jestLog('pipeline done')
+  const r = await readable.toArray()
 
   expect(r).toEqual(items)
 })

src/stream/readable/readableMap.test.ts

@@ -3,13 +3,17 @@ import { ReadableTyped } from '../stream.model'
 /**
  * Convenience function to read the whole Readable stream into Array (in-memory)
  * and return that array.
+ *
+ * Native `await readable.toArray()` can be used instead.
+ * This helper is kept for type-safery support.
  */
 export async function readableToArray<T>(readable: ReadableTyped<T>): Promise<T[]> {
-  const a: T[] = []
-
-  for await (const item of readable) {
-    a.push(item)
-  }
-
-  return a
+  return await readable.toArray()
+  // const a: T[] = []
+  //
+  // for await (const item of readable) {
+  //   a.push(item)
+  // }
+  //
+  // return a
 }

src/stream/readable/readableMap.ts

@@ -1,13 +1,55 @@
-import { Readable, Transform, Writable } from 'node:stream'
+import type { Readable, Transform, Writable } from 'node:stream'
+import type { Promisable } from '@naturalcycles/js-lib'
 
-// eslint-disable-next-line unused-imports/no-unused-vars
-export interface ReadableTyped<T> extends Readable {}
+export interface ReadableSignalOptions {
+  /** allows destroying the stream if the signal is aborted. */
+  signal?: AbortSignal
+}
+
+export interface ReadableArrayOptions {
+  /** the maximum concurrent invocations of `fn` to call on the stream at once. **Default: 1**. */
+  concurrency?: number
+  /** allows destroying the stream if the signal is aborted. */
+  signal?: AbortSignal
+}
+
+export type ReadableMapper<IN, OUT> = (data: IN, opt?: ReadableSignalOptions) => Promisable<OUT>
+
+export type ReadableFlatMapper<IN, OUT> = (
+  data: IN,
+  opt?: ReadableSignalOptions,
+) => Promisable<OUT[]>
+
+export type ReadableVoidMapper<IN> = (data: IN, opt?: ReadableSignalOptions) => void | Promise<void>
+
+export type ReadablePredicate<IN> = (
+  data: IN,
+  opt?: ReadableSignalOptions,
+) => boolean | Promise<boolean>
+
+export interface ReadableTyped<T> extends Readable {
+  toArray: (opt?: ReadableSignalOptions) => Promise<T[]>
+
+  map: <OUT>(mapper: ReadableMapper<T, OUT>, opt?: ReadableArrayOptions) => ReadableTyped<OUT>
+
+  flatMap: <OUT>(
+    mapper: ReadableFlatMapper<T, OUT>,
+    opt?: ReadableArrayOptions,
+  ) => ReadableTyped<OUT>
+
+  filter: (predicate: ReadablePredicate<T>, opt?: ReadableArrayOptions) => ReadableTyped<T>
+
+  forEach: (mapper: ReadableVoidMapper<T>, opt?: ReadableArrayOptions) => Promise<void>
+
+  take: (limit: number, opt?: ReadableSignalOptions) => ReadableTyped<T>
+  drop: (limit: number, opt?: ReadableSignalOptions) => ReadableTyped<T>
+}
 
 // eslint-disable-next-line unused-imports/no-unused-vars
 export interface WritableTyped<T> extends Writable {}
 
 // eslint-disable-next-line unused-imports/no-unused-vars
-export interface TransformTyped<IN, OUT = IN> extends Transform {}
+export interface TransformTyped<IN, OUT> extends Transform {}
 
 export interface TransformOptions {
   /**