Have stream helpers accept retry options

Author: jdalton

registry/lib/arrays.d.ts

@@ -1,5 +1,5 @@
 declare const Arrays: {
-  arrayChunk<T>(arr: T[] | readonly T[], size?: number): T[][]
+  arrayChunk<T>(arr: T[] | readonly T[], size?: number | undefined): T[][]
   arrayUnique<T>(arr: T[] | readonly T[]): T[]
   joinAnd(arr: string[] | readonly string[]): string
   joinOr(arr: string[] | readonly string[]): string

registry/lib/promises.d.ts

@@ -1,53 +1,85 @@
+declare type OnRetry = (attempt: number, error: unknown, delay: number) => void
+declare type pIterationContext = {
+  signal: AbortSignal
+}
+declare type pIterationOptions = {
+  concurrency?: number | undefined
+  retries?: pRetryOptions | undefined
+  signal?: AbortSignal | undefined
+}
 declare type pRetryOptions =
   | number
   | {
+      args?: any[] | undefined
       backoffFactor?: number | undefined
       baseDelayMs?: number | undefined
       jitter?: boolean | undefined
       maxDelayMs?: number | undefined
-      onRetry?: (
-        attempt: number,
-        error: unknown,
-        delay: number
-      ) => void | undefined
+      onRetry?: OnRetry | undefined
       onRetryCancelOnFalse?: boolean | undefined
       onRetryRethrow?: boolean | undefined
       retries?: number | undefined
       signal?: AbortSignal | undefined
     }
-declare type pOptions = {
-  retries?: pRetryOptions | undefined
-  signal?: AbortSignal | undefined
+declare type pNormalizedIterationOptions = {
+  concurrency: number
+  retries: pRetryOptions
+  signal: AbortSignal
+}
+declare type pNormalizedRetryOptions = {
+  args: any[]
+  backoffFactor: number
+  baseDelayMs: number
+  concurrency: number
+  jitter: boolean
+  maxDelayMs: number
+  onRetry?: OnRetry | undefined
+  onRetryCancelOnFalse: boolean
+  onRetryRethrow: boolean
+  retries: number
+  signal: AbortSignal
 }
 declare const Promises: {
+  normalizeIterationOptions(
+    options?: pIterationOptions | undefined
+  ): pNormalizedIterationOptions
+  normalizeRetryOptions(
+    options?: pRetryOptions | undefined
+  ): pNormalizedRetryOptions
   pEach<T>(
     array: T[],
-    concurrency: number,
-    callbackFn: (value: T, options: pOptions) => Promise<any>,
-    options?: pOptions | undefined
+    callbackFn: (value: T, context: pIterationContext) => Promise<any>,
+    options?: pIterationOptions | undefined
   ): Promise<void>
   pEachChunk<T>(
     chunks: T[][],
-    callbackFn: (value: T, options: pOptions) => Promise<any>,
-    options?: pOptions | undefined
+    callbackFn: (value: T, context: pIterationContext) => Promise<any>,
+    options?: pRetryOptions | undefined
   ): Promise<void>
   pFilter<T>(
     array: T[],
-    concurrency: number,
-    callbackFn: (value: T, options: pOptions) => Promise<boolean>,
-    options?: pOptions | undefined
+    callbackFn: (value: T, context: pIterationContext) => Promise<boolean>,
+    options?: pIterationOptions | undefined
   ): Promise<T[]>
   pFilterChunk<T>(
     chunks: T[][],
-    callbackFn: (value: T, options: pOptions) => Promise<boolean>,
-    options?: pOptions | undefined
+    callbackFn: (value: T, context: pIterationContext) => Promise<boolean>,
+    options?: pRetryOptions | undefined
   ): Promise<T[][]>
-  pRetry<T, P extends (value: T, options: pOptions) => Promise<any>>(
+  pRetry<T, P extends (value: T, context: pIterationContext) => Promise<any>>(
     callbackFn: P,
     options?: pRetryOptions | undefined
   ): ReturnType<P>
+  resolveRetryOptions(options?: pRetryOptions | undefined): pRetryOptions
 }
 declare namespace Promises {
-  export { pOptions, pRetryOptions }
+  export {
+    OnRetry,
+    pIterationContext,
+    pIterationOptions,
+    pNormalizedIterationOptions,
+    pNormalizedRetryOptions,
+    pRetryOptions
+  }
 }
 export = Promises

registry/lib/promises.js

@@ -13,6 +13,25 @@ function getTimers() {
   return _timers
 }
 
+/*@__NO_SIDE_EFFECTS__*/
+function normalizeIterationOptions(options) {
+  const {
+    // The number of concurrent executions performed at one time.
+    concurrency = 1,
+    // Retries as a number or options object.
+    retries,
+    // AbortSignal used to support cancellation.
+    signal = /*@__PURE__*/ require('./constants/abort-signal')
+  } = { __proto__: null, ...options }
+  const retryOpts = resolveRetryOptions(retries)
+  return {
+    __proto__: null,
+    concurrency,
+    retries: normalizeRetryOptions({ signal, ...retryOpts }),
+    signal
+  }
+}
+
 /*@__NO_SIDE_EFFECTS__*/
 function normalizeRetryOptions(options) {
   const {
@@ -62,23 +81,31 @@ function resolveRetryOptions(options) {
 }
 
 /*@__NO_SIDE_EFFECTS__*/
-async function pEach(array, concurrency, callbackFn, options) {
-  await pEachChunk(arrayChunk(array, concurrency), callbackFn, options)
+async function pEach(array, callbackFn, options) {
+  const iterOpts = normalizeIterationOptions(options)
+  await pEachChunk(
+    arrayChunk(array, iterOpts.concurrency),
+    callbackFn,
+    iterOpts.retries
+  )
 }
 
 /*@__NO_SIDE_EFFECTS__*/
-async function pFilter(array, concurrency, callbackFn, options) {
+async function pFilter(array, callbackFn, options) {
+  const iterOpts = normalizeIterationOptions(options)
   return (
-    await pFilterChunk(arrayChunk(array, concurrency), callbackFn, options)
+    await pFilterChunk(
+      arrayChunk(array, iterOpts.concurrency),
+      callbackFn,
+      iterOpts.retries
+    )
   ).flat()
 }
 
 /*@__NO_SIDE_EFFECTS__*/
 async function pEachChunk(chunks, callbackFn, options) {
-  const {
-    retries,
-    signal = /*@__PURE__*/ require('./constants/abort-signal')
-  } = { __proto__: null, ...options }
+  const retryOpts = normalizeRetryOptions(options)
+  const { signal } = retryOpts
   for (const chunk of chunks) {
     if (signal?.aborted) {
       return
@@ -87,8 +114,7 @@ async function pEachChunk(chunks, callbackFn, options) {
     await Promise.all(
       chunk.map(value =>
         pRetry(callbackFn, {
-          signal,
-          ...resolveRetryOptions(retries),
+          ...retryOpts,
           args: [value]
         })
       )
@@ -98,10 +124,8 @@ async function pEachChunk(chunks, callbackFn, options) {
 
 /*@__NO_SIDE_EFFECTS__*/
 async function pFilterChunk(chunks, callbackFn, options) {
-  const {
-    retries,
-    signal = /*@__PURE__*/ require('./constants/abort-signal')
-  } = { __proto__: null, ...options }
+  const retryOpts = normalizeRetryOptions(options)
+  const { signal } = retryOpts
   const { length } = chunks
   const filteredChunks = Array(length)
   for (let i = 0; i < length; i += 1) {
@@ -114,8 +138,7 @@ async function pFilterChunk(chunks, callbackFn, options) {
       const predicateResults = await Promise.all(
         chunk.map(value =>
           pRetry(callbackFn, {
-            signal,
-            ...resolveRetryOptions(retries),
+            ...retryOpts,
             args: [value]
           })
         )
@@ -197,9 +220,12 @@ async function pRetry(callbackFn, options) {
 }
 
 module.exports = {
+  normalizeIterationOptions,
+  normalizeRetryOptions,
   pEach,
   pEachChunk,
   pFilter,
   pFilterChunk,
-  pRetry
+  pRetry,
+  resolveRetryOptions
 }

registry/lib/streams.d.ts

@@ -1,54 +1,29 @@
 /// <reference types="node" />
+import { pIterationOptions } from './promises'
+
+declare type sIterationOptions = pIterationOptions
 declare type AnyIterable<T> =
   | Iterable<T>
   | AsyncIterable<T>
   | NodeJS.ReadableStream
 declare const Streams: {
   parallelForEach<T>(
-    concurrency: number,
+    iterable: AnyIterable<T>,
     func: (data: T) => Promise<void>,
-    iterable: AnyIterable<T>
+    options?: sIterationOptions | undefined
   ): Promise<void>
   parallelMap<T, R>(
-    concurrency: number
-  ): {
-    (
-      func: (data: T) => R | Promise<R>,
-      iterable: AnyIterable<T>
-    ): AsyncIterableIterator<R>
-    (
-      func: (data: T) => R | Promise<R>
-    ): (iterable: AnyIterable<T>) => AsyncIterableIterator<R>
-  }
-  parallelMap<T, R>(
-    concurrency: number,
-    func: (data: T) => R | Promise<R>
-  ): (iterable: AnyIterable<T>) => AsyncIterableIterator<R>
-  parallelMap<T, R>(
-    concurrency: number,
+    iterable: AnyIterable<T>,
     func: (data: T) => R | Promise<R>,
-    iterable: AnyIterable<T>
+    options?: sIterationOptions | undefined
   ): AsyncIterableIterator<R>
-  transform(concurrency: number): {
-    <T, R>(
-      func: (data: T) => R | Promise<R>,
-      iterable: AnyIterable<T>
-    ): AsyncIterableIterator<R>
-    <T, R>(
-      func: (data: T) => R | Promise<R>
-    ): (iterable: AnyIterable<T>) => AsyncIterableIterator<R>
-  }
-  transform<T, R>(
-    concurrency: number,
-    func: (data: T) => R | Promise<R>
-  ): (iterable: AnyIterable<T>) => AsyncIterableIterator<R>
   transform<T, R>(
-    concurrency: number,
+    iterable: AnyIterable<T>,
     func: (data: T) => R | Promise<R>,
-    iterable: AnyIterable<T>
+    options?: sIterationOptions | undefined
   ): AsyncIterableIterator<R>
 }
 declare namespace Streams {
-  export { AnyIterable }
+  export { AnyIterable, sIterationOptions }
 }
 export = Streams

registry/lib/streams.js

@@ -1,6 +1,9 @@
 'use strict'
 
-const { apply: ReflectApply } = Reflect
+const {
+  normalizeIterationOptions,
+  pRetry
+} = /*@__PURE__*/ require('./promises')
 
 let _streamingIterables
 /*@__NO_SIDE_EFFECTS__*/
@@ -12,22 +15,40 @@ function getStreamingIterables() {
 }
 
 /*@__NO_SIDE_EFFECTS__*/
-async function parallelForEach(concurrency, func, iterable) {
-  for await (const _ of parallelMap(concurrency, func, iterable)) {
+async function parallelForEach(iterable, func, options) {
+  for await (const _ of parallelMap(iterable, func, options)) {
     /* empty block */
   }
 }
 
 /*@__NO_SIDE_EFFECTS__*/
-function parallelMap(...args) {
+function parallelMap(iterable, func, options) {
   const streamingIterables = getStreamingIterables()
-  return ReflectApply(streamingIterables.parallelMap, undefined, args)
+  const opts = normalizeIterationOptions(options)
+  return streamingIterables.parallelMap(
+    opts.concurrency,
+    item =>
+      pRetry(func, {
+        ...opts.retries,
+        args: [item]
+      }),
+    iterable
+  )
 }
 
 /*@__NO_SIDE_EFFECTS__*/
-function transform(...args) {
+function transform(iterable, func, options) {
   const streamingIterables = getStreamingIterables()
-  return ReflectApply(streamingIterables.transform, undefined, args)
+  const opts = normalizeIterationOptions(options)
+  return streamingIterables.transform(
+    opts.concurrency,
+    item =>
+      pRetry(func, {
+        ...opts.retries,
+        args: [item]
+      }),
+    iterable
+  )
 }
 
 module.exports = {